Skip to content

Commit 2a9d4ac

Browse files
lhotarilhotari
authored
[improve][test] Support decorating topic, subscription, dispatcher, ManagedLedger and ManagedCursors instances in tests (#23892)
1 parent 223eea0 commit 2a9d4ac

File tree

7 files changed

+333
-74
lines changed

7 files changed

+333
-74
lines changed

‎managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java

+10-5
Original file line numberDiff line numberDiff line change
@@ -410,11 +410,8 @@ public void asyncOpen(final String name, final ManagedLedgerConfig config, final
410410
new EnsemblePlacementPolicyConfig(config.getBookKeeperEnsemblePlacementPolicyClassName(),
411411
config.getBookKeeperEnsemblePlacementPolicyProperties()))
412412
.thenAccept(bk -> {
413-
final ManagedLedgerImpl newledger = config.getShadowSource() == null
414-
? new ManagedLedgerImpl(this, bk, store, config, scheduledExecutor, name,
415-
mlOwnershipChecker)
416-
: new ShadowManagedLedgerImpl(this, bk, store, config, scheduledExecutor, name,
417-
mlOwnershipChecker);
413+
final ManagedLedgerImpl newledger =
414+
createManagedLedger(bk, store, name, config, mlOwnershipChecker);
418415
PendingInitializeManagedLedger pendingLedger = new PendingInitializeManagedLedger(newledger);
419416
pendingInitializeLedgers.put(name, pendingLedger);
420417
newledger.initialize(new ManagedLedgerInitializeLedgerCallback() {
@@ -472,6 +469,14 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) {
472469
});
473470
}
474471

472+
protected ManagedLedgerImpl createManagedLedger(BookKeeper bk, MetaStore store, String name,
473+
ManagedLedgerConfig config,
474+
Supplier<CompletableFuture<Boolean>> mlOwnershipChecker) {
475+
return config.getShadowSource() == null
476+
? new ManagedLedgerImpl(this, bk, store, config, scheduledExecutor, name, mlOwnershipChecker) :
477+
new ShadowManagedLedgerImpl(this, bk, store, config, scheduledExecutor, name, mlOwnershipChecker);
478+
}
479+
475480
@Override
476481
public void asyncOpenReadOnlyManagedLedger(String managedLedgerName,
477482
AsyncCallbacks.OpenReadOnlyManagedLedgerCallback callback,

‎managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java

+7-3
Original file line numberDiff line numberDiff line change
@@ -632,7 +632,7 @@ public void operationComplete(List<String> consumers, Stat s) {
632632
for (final String cursorName : consumers) {
633633
log.info("[{}] Loading cursor {}", name, cursorName);
634634
final ManagedCursorImpl cursor;
635-
cursor = new ManagedCursorImpl(bookKeeper, ManagedLedgerImpl.this, cursorName);
635+
cursor = createCursor(ManagedLedgerImpl.this.bookKeeper, cursorName);
636636

637637
cursor.recover(new VoidCallback() {
638638
@Override
@@ -663,7 +663,7 @@ public void operationFailed(ManagedLedgerException exception) {
663663
log.debug("[{}] Recovering cursor {} lazily", name, cursorName);
664664
}
665665
final ManagedCursorImpl cursor;
666-
cursor = new ManagedCursorImpl(bookKeeper, ManagedLedgerImpl.this, cursorName);
666+
cursor = createCursor(ManagedLedgerImpl.this.bookKeeper, cursorName);
667667
CompletableFuture<ManagedCursor> cursorRecoveryFuture = new CompletableFuture<>();
668668
uninitializedCursors.put(cursorName, cursorRecoveryFuture);
669669

@@ -1007,7 +1007,7 @@ public synchronized void asyncOpenCursor(final String cursorName, final InitialP
10071007
if (log.isDebugEnabled()) {
10081008
log.debug("[{}] Creating new cursor: {}", name, cursorName);
10091009
}
1010-
final ManagedCursorImpl cursor = new ManagedCursorImpl(bookKeeper, this, cursorName);
1010+
final ManagedCursorImpl cursor = createCursor(bookKeeper, cursorName);
10111011
CompletableFuture<ManagedCursor> cursorFuture = new CompletableFuture<>();
10121012
uninitializedCursors.put(cursorName, cursorFuture);
10131013
Position position = InitialPosition.Earliest == initialPosition ? getFirstPosition() : getLastPosition();
@@ -1039,6 +1039,10 @@ public void operationFailed(ManagedLedgerException exception) {
10391039
});
10401040
}
10411041

1042+
protected ManagedCursorImpl createCursor(BookKeeper bookKeeper, String cursorName) {
1043+
return new ManagedCursorImpl(bookKeeper, this, cursorName);
1044+
}
1045+
10421046
@Override
10431047
public synchronized void asyncDeleteCursor(final String consumerName, final DeleteCursorCallback callback,
10441048
final Object ctx) {

‎pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java

+12-2
Original file line numberDiff line numberDiff line change
@@ -116,8 +116,8 @@ public void initialize(ServiceConfiguration conf, MetadataStoreExtended metadata
116116

117117
try {
118118
this.managedLedgerFactory =
119-
new ManagedLedgerFactoryImpl(metadataStore, bkFactory, managedLedgerFactoryConfig, statsLogger,
120-
openTelemetry);
119+
createManagedLedgerFactory(metadataStore, openTelemetry, bkFactory, managedLedgerFactoryConfig,
120+
statsLogger);
121121
} catch (Exception e) {
122122
statsProvider.stop();
123123
defaultBkClient.close();
@@ -147,6 +147,16 @@ public BookKeeper getBookKeeperClient() {
147147
};
148148
}
149149

150+
protected ManagedLedgerFactoryImpl createManagedLedgerFactory(MetadataStoreExtended metadataStore,
151+
OpenTelemetry openTelemetry,
152+
BookkeeperFactoryForCustomEnsemblePlacementPolicy
153+
bkFactory,
154+
ManagedLedgerFactoryConfig managedLedgerFactoryConfig,
155+
StatsLogger statsLogger) throws Exception {
156+
return new ManagedLedgerFactoryImpl(metadataStore, bkFactory, managedLedgerFactoryConfig, statsLogger,
157+
openTelemetry);
158+
}
159+
150160
@Override
151161
public Collection<ManagedLedgerStorageClass> getStorageClasses() {
152162
return List.of(getDefaultStorageClass());

‎pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicFactory.java

+5
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
package org.apache.pulsar.broker.service;
2020

2121
import java.io.Closeable;
22+
import java.io.IOException;
2223
import org.apache.bookkeeper.mledger.ManagedLedger;
2324

2425
/**
@@ -28,4 +29,8 @@
2829
public interface TopicFactory extends Closeable {
2930

3031
<T extends Topic> T create(String topic, ManagedLedger ledger, BrokerService brokerService, Class<T> topicClazz);
32+
33+
default void close() throws IOException {
34+
// default implementation
35+
}
3136
}

‎pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

+76-63
Original file line numberDiff line numberDiff line change
@@ -250,70 +250,10 @@ private CompletableFuture<Void> addConsumerInternal(Consumer consumer) {
250250
}
251251

252252
if (dispatcher == null || !dispatcher.isConsumerConnected()) {
253-
Dispatcher previousDispatcher = null;
254-
switch (consumer.subType()) {
255-
case Exclusive:
256-
if (dispatcher == null || dispatcher.getType() != SubType.Exclusive) {
257-
previousDispatcher = dispatcher;
258-
dispatcher = new PersistentDispatcherSingleActiveConsumer(
259-
cursor, SubType.Exclusive, 0, topic, this);
260-
}
261-
break;
262-
case Shared:
263-
if (dispatcher == null || dispatcher.getType() != SubType.Shared) {
264-
previousDispatcher = dispatcher;
265-
if (config.isSubscriptionSharedUseClassicPersistentImplementation()) {
266-
dispatcher = new PersistentDispatcherMultipleConsumersClassic(topic, cursor, this);
267-
} else {
268-
dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursor, this);
269-
}
270-
}
271-
break;
272-
case Failover:
273-
int partitionIndex = TopicName.getPartitionIndex(topicName);
274-
if (partitionIndex < 0) {
275-
// For non partition topics, use a negative index so
276-
// dispatcher won't sort consumers before picking
277-
// an active consumer for the topic.
278-
partitionIndex = -1;
279-
}
280-
281-
if (dispatcher == null || dispatcher.getType() != SubType.Failover) {
282-
previousDispatcher = dispatcher;
283-
dispatcher = new PersistentDispatcherSingleActiveConsumer(cursor, SubType.Failover,
284-
partitionIndex, topic, this);
285-
}
286-
break;
287-
case Key_Shared:
288-
KeySharedMeta ksm = consumer.getKeySharedMeta();
289-
if (dispatcher == null || dispatcher.getType() != SubType.Key_Shared
290-
|| !((StickyKeyDispatcher) dispatcher)
291-
.hasSameKeySharedPolicy(ksm)) {
292-
previousDispatcher = dispatcher;
293-
if (config.isSubscriptionKeySharedUseClassicPersistentImplementation()) {
294-
dispatcher =
295-
new PersistentStickyKeyDispatcherMultipleConsumersClassic(topic, cursor,
296-
this,
297-
topic.getBrokerService().getPulsar().getConfiguration(), ksm);
298-
} else {
299-
dispatcher = new PersistentStickyKeyDispatcherMultipleConsumers(topic, cursor, this,
300-
topic.getBrokerService().getPulsar().getConfiguration(), ksm);
301-
}
302-
}
303-
break;
304-
default:
305-
return FutureUtil.failedFuture(
306-
new ServerMetadataException("Unsupported subscription type"));
307-
}
308-
309-
if (previousDispatcher != null) {
310-
previousDispatcher.close().thenRun(() -> {
311-
log.info("[{}][{}] Successfully closed previous dispatcher", topicName, subName);
312-
}).exceptionally(ex -> {
313-
log.error("[{}][{}] Failed to close previous dispatcher", topicName, subName, ex);
314-
return null;
315-
});
253+
if (consumer.subType() == null) {
254+
return FutureUtil.failedFuture(new ServerMetadataException("Unsupported subscription type"));
316255
}
256+
dispatcher = reuseOrCreateDispatcher(dispatcher, consumer);
317257
} else {
318258
Optional<CompletableFuture<Void>> compatibilityError =
319259
checkForConsumerCompatibilityErrorWithDispatcher(dispatcher, consumer);
@@ -327,6 +267,79 @@ private CompletableFuture<Void> addConsumerInternal(Consumer consumer) {
327267
});
328268
}
329269

270+
/**
271+
* Create a new dispatcher or reuse the existing one when it's compatible with the new consumer.
272+
* This protected method can be overridded for testing purpose for injecting test dispatcher instances with
273+
* special behaviors.
274+
* @param dispatcher the existing dispatcher
275+
* @param consumer the new consumer
276+
* @return the dispatcher to use, either the existing one or a new one
277+
*/
278+
protected Dispatcher reuseOrCreateDispatcher(Dispatcher dispatcher, Consumer consumer) {
279+
Dispatcher previousDispatcher = null;
280+
switch (consumer.subType()) {
281+
case Exclusive:
282+
if (dispatcher == null || dispatcher.getType() != SubType.Exclusive) {
283+
previousDispatcher = dispatcher;
284+
dispatcher = new PersistentDispatcherSingleActiveConsumer(
285+
cursor, SubType.Exclusive, 0, topic, this);
286+
}
287+
break;
288+
case Shared:
289+
if (dispatcher == null || dispatcher.getType() != SubType.Shared) {
290+
previousDispatcher = dispatcher;
291+
if (config.isSubscriptionSharedUseClassicPersistentImplementation()) {
292+
dispatcher = new PersistentDispatcherMultipleConsumersClassic(topic, cursor, this);
293+
} else {
294+
dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursor, this);
295+
}
296+
}
297+
break;
298+
case Failover:
299+
int partitionIndex = TopicName.getPartitionIndex(topicName);
300+
if (partitionIndex < 0) {
301+
// For non partition topics, use a negative index so
302+
// dispatcher won't sort consumers before picking
303+
// an active consumer for the topic.
304+
partitionIndex = -1;
305+
}
306+
307+
if (dispatcher == null || dispatcher.getType() != SubType.Failover) {
308+
previousDispatcher = dispatcher;
309+
dispatcher = new PersistentDispatcherSingleActiveConsumer(cursor, SubType.Failover,
310+
partitionIndex, topic, this);
311+
}
312+
break;
313+
case Key_Shared:
314+
KeySharedMeta ksm = consumer.getKeySharedMeta();
315+
if (dispatcher == null || dispatcher.getType() != SubType.Key_Shared
316+
|| !((StickyKeyDispatcher) dispatcher)
317+
.hasSameKeySharedPolicy(ksm)) {
318+
previousDispatcher = dispatcher;
319+
if (config.isSubscriptionKeySharedUseClassicPersistentImplementation()) {
320+
dispatcher =
321+
new PersistentStickyKeyDispatcherMultipleConsumersClassic(topic, cursor,
322+
this, config, ksm);
323+
} else {
324+
dispatcher = new PersistentStickyKeyDispatcherMultipleConsumers(topic, cursor, this,
325+
config, ksm);
326+
}
327+
}
328+
break;
329+
}
330+
331+
if (previousDispatcher != null) {
332+
previousDispatcher.close().thenRun(() -> {
333+
log.info("[{}][{}] Successfully closed previous dispatcher", topicName, subName);
334+
}).exceptionally(ex -> {
335+
log.error("[{}][{}] Failed to close previous dispatcher", topicName, subName, ex);
336+
return null;
337+
});
338+
}
339+
340+
return dispatcher;
341+
}
342+
330343
@Override
331344
public synchronized void removeConsumer(Consumer consumer, boolean isResetCursor) throws BrokerServiceException {
332345
cursor.updateLastActive();

‎pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

+11-1
Original file line numberDiff line numberDiff line change
@@ -586,7 +586,17 @@ public CompletableFuture<Void> unloadSubscription(@Nonnull String subName) {
586586
});
587587
}
588588

589-
private PersistentSubscription createPersistentSubscription(String subscriptionName, ManagedCursor cursor,
589+
590+
/**
591+
* Create a new subscription instance for the topic.
592+
* This protected method can be overridden in tests to return a special test implementation instance.
593+
* @param subscriptionName the name of the subscription
594+
* @param cursor the cursor to use for the subscription
595+
* @param replicated the subscription replication flag
596+
* @param subscriptionProperties the subscription properties
597+
* @return the subscription instance
598+
*/
599+
protected PersistentSubscription createPersistentSubscription(String subscriptionName, ManagedCursor cursor,
590600
Boolean replicated, Map<String, String> subscriptionProperties) {
591601
requireNonNull(topicCompactionService);
592602
if (isCompactionSubscription(subscriptionName)

0 commit comments

Comments
 (0)