Skip to content

Commit 7a36b1b

Browse files
mattkaemmattkaem
committed
Add PubSub batch initialization to improve start up time in multi tenant instances
Fix PubSub tenant topic/subscription clean up on tenant delete Signed-off-by: Matthias Kaemmer <[email protected]>
1 parent ed7c646 commit 7a36b1b

File tree

11 files changed

+664
-98
lines changed

11 files changed

+664
-98
lines changed

Diff for: ‎device-communication/src/main/java/org/eclipse/hono/communication/api/config/PubSubConstants.java

+13-3
Original file line numberDiff line numberDiff line change
@@ -29,20 +29,30 @@ public final class PubSubConstants {
2929
public static final String EVENT_STATES_SUBTOPIC_ENDPOINT = "event.state";
3030
public static final String COMMAND_ENDPOINT = "command";
3131
public static final String COMMAND_RESPONSE_ENDPOINT = "command_response";
32+
public static final String COMMUNICATION_API_SUBSCRIPTION_NAME = "%s-communication-api";
3233

3334
private PubSubConstants() {
3435
}
3536

3637
/**
37-
* Gets the list of all topics need to be created per tenant.
38+
* Gets the list of all endpoints for which a topic per tenant has to be created.
3839
*
39-
* @return List of all topics.
40+
* @return List of all topic endpoints.
4041
*/
41-
public static List<String> getTenantTopics() {
42+
public static List<String> getTenantEndpoints() {
4243
return List.of(EVENT_ENDPOINT,
4344
COMMAND_ENDPOINT,
4445
COMMAND_RESPONSE_ENDPOINT,
4546
EVENT_STATES_SUBTOPIC_ENDPOINT,
4647
TELEMETRY_ENDPOINT);
4748
}
49+
50+
/**
51+
* Gets the list of all endpoints for which an additional subscription per tenant has to be created.
52+
*
53+
* @return List of all endpoints that need an additional subscription.
54+
*/
55+
public static List<String> getEndpointsWithAdditionalSubscription() {
56+
return List.of(EVENT_ENDPOINT, COMMAND_RESPONSE_ENDPOINT, EVENT_STATES_SUBTOPIC_ENDPOINT);
57+
}
4858
}

Diff for: ‎device-communication/src/main/java/org/eclipse/hono/communication/api/service/communication/InternalMessageSubscriber.java

+7
Original file line numberDiff line numberDiff line change
@@ -30,4 +30,11 @@ public interface InternalMessageSubscriber {
3030
* @param callbackHandler The function to be called when a message is received
3131
*/
3232
void subscribe(String topic, MessageReceiver callbackHandler);
33+
34+
/**
35+
* Closes all active subscribers for the given tenant.
36+
*
37+
* @param tenant The tenant whose active subscribers should be closed.
38+
*/
39+
void closeSubscribersForTenant(String tenant);
3340
}

Diff for: ‎device-communication/src/main/java/org/eclipse/hono/communication/api/service/communication/InternalTopicManagerImpl.java

+137-15
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,13 @@
1717
package org.eclipse.hono.communication.api.service.communication;
1818

1919
import java.util.ArrayList;
20+
import java.util.HashMap;
2021
import java.util.List;
22+
import java.util.Map;
23+
import java.util.Set;
24+
import java.util.stream.Stream;
2125

22-
import org.eclipse.hono.client.pubsub.PubSubBasedAdminClientManager;
26+
import org.apache.commons.lang3.StringUtils;
2327
import org.eclipse.hono.client.pubsub.PubSubMessageHelper;
2428
import org.eclipse.hono.communication.api.config.PubSubConstants;
2529
import org.eclipse.hono.communication.api.handler.CommandTopicEventHandler;
@@ -37,6 +41,7 @@
3741
import com.google.cloud.pubsub.v1.AckReplyConsumer;
3842
import com.google.common.base.Strings;
3943
import com.google.pubsub.v1.PubsubMessage;
44+
import com.google.pubsub.v1.Subscription;
4045
import com.google.pubsub.v1.SubscriptionName;
4146
import com.google.pubsub.v1.TopicName;
4247

@@ -96,16 +101,115 @@ public InternalTopicManagerImpl(final DeviceRepository deviceRepository,
96101
@Override
97102
public void initPubSub() {
98103
log.info("Initialize tenant topics and subscriptions.");
99-
internalMessaging.subscribe(PubSubConstants.TENANT_NOTIFICATIONS, this::onTenantChanges);
104+
vertx.executeBlocking(promise -> {
105+
internalMessaging.subscribe(PubSubConstants.TENANT_NOTIFICATIONS, this::onTenantChanges);
106+
promise.complete();
107+
});
100108
deviceRepository.listDistinctTenants()
101109
.onFailure(err -> log.error("Error getting tenants for topic creation: {}", err.getMessage()))
102110
.onSuccess(tenants -> {
111+
if (tenants.size() < internalMessagingConfig.getBatchInitTenantThreshold()) {
112+
for (String tenant : tenants) {
113+
initPubSubForTenant(tenant);
114+
}
115+
} else {
116+
batchInitPubSubResources(tenants);
117+
}
118+
});
119+
}
120+
121+
private void batchInitPubSubResources(final List<String> tenants) {
122+
log.info("Start batchInitPubSubResources");
123+
findMissingPubSubResources(tenants).compose(map -> {
124+
final PubSubBasedAdminClientManager adminClientManager = adminClientManagerFactory
125+
.createAdminClientManager();
126+
final List<Future> pubsubRequests = new ArrayList<>();
127+
for (Map.Entry<String, String> missingTopic : map.get("missingTopics").entrySet()) {
128+
final TopicName topic = TopicName.parse(missingTopic.getKey());
129+
pubsubRequests.add(adminClientManager.createTopic(topic));
130+
}
131+
for (Map.Entry<String, String> missingSubscription : map.get("missingSubscriptions").entrySet()) {
132+
final SubscriptionName subscription = SubscriptionName.parse(missingSubscription.getKey());
133+
final TopicName topic = TopicName.parse(missingSubscription.getValue());
134+
pubsubRequests.add(adminClientManager.createSubscription(subscription, topic));
135+
}
136+
for (Map.Entry<String, String> faultySubscription : map.get("faultySubscriptions").entrySet()) {
137+
final SubscriptionName subscription = SubscriptionName.parse(faultySubscription.getKey());
138+
final TopicName topic = TopicName.parse(faultySubscription.getValue());
139+
pubsubRequests.add(adminClientManager.updateSubscriptionTopic(subscription, topic));
140+
}
141+
return CompositeFuture.join(pubsubRequests).onComplete(i -> {
142+
adminClientManager.closeAdminClients();
143+
log.info("Finished batchInitPubSubResources");
144+
});
145+
}).onFailure(e -> log.error("batchInitPubSubResources failed", e))
146+
.onSuccess(i -> {
103147
for (String tenant : tenants) {
104-
initPubSubForTenant(tenant);
148+
vertx.executeBlocking(promise -> subscribeToTenantTopics(tenant));
105149
}
106150
});
107151
}
108152

153+
private Future<Map<String, Map<String, String>>> findMissingPubSubResources(final List<String> tenants) {
154+
final PubSubBasedAdminClientManager pubSubBasedAdminClientManager = adminClientManagerFactory
155+
.createAdminClientManager();
156+
final Future<Set<String>> topicsFuture = pubSubBasedAdminClientManager.listTopics();
157+
final Future<Set<Subscription>> subscriptionsFuture = pubSubBasedAdminClientManager.listSubscriptions();
158+
final Map<String, String> missingTopics = new HashMap<>();
159+
final Map<String, String> missingSubscriptions = new HashMap<>();
160+
final Map<String, String> faultySubscriptions = new HashMap<>();
161+
return CompositeFuture.join(topicsFuture, subscriptionsFuture).map(i -> {
162+
final Set<String> topicSet = topicsFuture.result();
163+
final Set<Subscription> subscriptionSet = subscriptionsFuture.result();
164+
final Map<String, Subscription> subscriptionMap = new HashMap<>();
165+
subscriptionSet.forEach(s -> subscriptionMap.put(s.getName(), s));
166+
for (String tenantEndpoint : PubSubConstants.getTenantEndpoints()) {
167+
for (String tenant : tenants) {
168+
final String topic = String
169+
.valueOf(TopicName.of(projectId, PubSubMessageHelper.getTopicName(tenantEndpoint, tenant)));
170+
addMissingTopicToMap(topicSet, topic, missingTopics);
171+
final String subscription = String.valueOf(
172+
SubscriptionName.of(projectId, PubSubMessageHelper.getTopicName(tenantEndpoint, tenant)));
173+
addMissingOrFaultySubscription(subscriptionMap, missingSubscriptions, faultySubscriptions,
174+
subscription, topic);
175+
if (PubSubConstants.getEndpointsWithAdditionalSubscription().contains(tenantEndpoint)) {
176+
final String apiSubscription = String.valueOf(SubscriptionName.of(projectId,
177+
PubSubMessageHelper.getTopicName(
178+
String.format(PubSubConstants.COMMUNICATION_API_SUBSCRIPTION_NAME,
179+
tenantEndpoint),
180+
tenant)));
181+
addMissingOrFaultySubscription(subscriptionMap, missingSubscriptions, faultySubscriptions,
182+
apiSubscription, topic);
183+
}
184+
}
185+
}
186+
return Map.of("missingTopics", missingTopics, "missingSubscriptions", missingSubscriptions,
187+
"faultySubscriptions", faultySubscriptions);
188+
}).onFailure(thr -> log.error("Cannot find missing Pub/Sub resources", thr))
189+
.onComplete(i -> pubSubBasedAdminClientManager.closeAdminClients());
190+
}
191+
192+
private void addMissingTopicToMap(final Set<String> topicSet, final String topic,
193+
final Map<String, String> missingTopics) {
194+
if (!topicSet.contains(topic)) {
195+
missingTopics.put(topic, "");
196+
}
197+
}
198+
199+
private void addMissingOrFaultySubscription(final Map<String, Subscription> subscriptionMap,
200+
final Map<String, String> missingSubscriptions, final Map<String, String> faultySubscriptions,
201+
final String subscription, final String topic) {
202+
if (!subscriptionMap.containsKey(subscription)) {
203+
missingSubscriptions.put(subscription, topic);
204+
return;
205+
}
206+
final String deletedTopic = "_deleted-topic_";
207+
if (subscriptionMap.get(subscription) != null
208+
&& StringUtils.equals(subscriptionMap.get(subscription).getTopic(), deletedTopic)) {
209+
faultySubscriptions.put(subscription, topic);
210+
}
211+
}
212+
109213
/**
110214
* Handle incoming tenant CREATE notifications.
111215
*
@@ -165,12 +269,17 @@ private Future<Void> createPubSubResourceForTenant(final String tenantId,
165269
final PubSubResourceType pubSubResourceType,
166270
final PubSubBasedAdminClientManager pubSubBasedAdminClientManager) {
167271
final List<Future> futureList = new ArrayList<>();
168-
final List<String> topics = PubSubConstants.getTenantTopics();
169-
for (String topic : topics) {
272+
for (String tenantEndpoint : PubSubConstants.getTenantEndpoints()) {
170273
if (pubSubResourceType == PubSubResourceType.TOPIC) {
171-
futureList.add(pubSubBasedAdminClientManager.getOrCreateTopic(topic, tenantId));
274+
futureList.add(pubSubBasedAdminClientManager.getOrCreateTopic(tenantEndpoint, tenantId));
172275
} else {
173-
futureList.add(pubSubBasedAdminClientManager.getOrCreateSubscription(topic, tenantId));
276+
futureList.add(pubSubBasedAdminClientManager.getOrCreateSubscription(tenantEndpoint, tenantEndpoint,
277+
tenantId));
278+
if (PubSubConstants.getEndpointsWithAdditionalSubscription().contains(tenantEndpoint)) {
279+
futureList.add(pubSubBasedAdminClientManager.getOrCreateSubscription(tenantEndpoint,
280+
String.format(PubSubConstants.COMMUNICATION_API_SUBSCRIPTION_NAME, tenantEndpoint),
281+
tenantId));
282+
}
174283
}
175284
}
176285
return CompositeFuture.join(futureList)
@@ -198,18 +307,31 @@ private void subscribeToTenantTopics(final String tenant) {
198307
}
199308

200309
private void cleanupPubSubResources(final String tenant) {
201-
final List<String> pubSubTopicsToDelete = PubSubConstants.getTenantTopics().stream()
202-
.map(id -> TopicName.of(projectId, "%s.%s".formatted(tenant, id)).toString()).toList();
203-
final List<String> pubSubSubscriptionsToDelete = PubSubConstants.getTenantTopics().stream()
204-
.map(id -> SubscriptionName.of(projectId, "%s.%s".formatted(tenant, id)).toString()).toList();
310+
final List<String> pubSubTopicsToDelete = PubSubConstants.getTenantEndpoints().stream().map(
311+
endpoint -> String.valueOf(TopicName.of(projectId, PubSubMessageHelper.getTopicName(endpoint, tenant))))
312+
.toList();
313+
final List<String> pubSubSubscriptionsToDelete = Stream.concat(PubSubConstants.getTenantEndpoints().stream()
314+
.map(endpoint -> String
315+
.valueOf(SubscriptionName.of(projectId, PubSubMessageHelper.getTopicName(endpoint, tenant)))),
316+
PubSubConstants.getEndpointsWithAdditionalSubscription().stream()
317+
.map(endpoint -> String.valueOf(SubscriptionName.of(projectId,
318+
PubSubMessageHelper.getTopicName(
319+
String.format(PubSubConstants.COMMUNICATION_API_SUBSCRIPTION_NAME, endpoint),
320+
tenant)))))
321+
.toList();
322+
internalMessaging.closeSubscribersForTenant(tenant);
205323
PubSubMessageHelper.getCredentialsProvider()
206324
.ifPresentOrElse(provider -> {
207325
final PubSubBasedAdminClientManager pubSubBasedAdminClientManager = adminClientManagerFactory
208326
.createAdminClientManager();
209-
pubSubBasedAdminClientManager.deleteTopics(pubSubTopicsToDelete);
210-
pubSubBasedAdminClientManager.deleteSubscriptions(pubSubSubscriptionsToDelete);
211-
log.info("All topics and subscriptions for tenant {} were deleted successfully.", tenant);
212-
pubSubBasedAdminClientManager.closeAdminClients();
327+
CompositeFuture.join(pubSubBasedAdminClientManager.deleteTopics(pubSubTopicsToDelete),
328+
pubSubBasedAdminClientManager.deleteSubscriptions(pubSubSubscriptionsToDelete))
329+
.onSuccess(compFuture -> log.info(
330+
"All topics and subscriptions of tenant {} were deleted successfully.", tenant))
331+
.onFailure(throwable -> log.warn(
332+
"Some topics or subscriptions of tenant {} could not be deleted.", tenant,
333+
throwable))
334+
.onComplete(compFuture -> pubSubBasedAdminClientManager.closeAdminClients());
213335
}, () -> log.error("credentials provider is empty"));
214336
}
215337

0 commit comments

Comments
 (0)