You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/05/14 15:46:47 UTC
(pulsar) branch branch-4.0 updated (b63f685c9a7 -> 1e3649cd425)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from b63f685c9a7 [cleanup][test] Remove unused parameter from deleteNamespaceWithRetry method in MockedPulsarServiceBaseTest (#24283)
new a66d6b786d3 [fix][client] Fix producer publishing getting stuck after message with incompatible schema is discarded (#24282)
new 1c0c1dc1448 [fix][io] Fix SyntaxWarning in Pulsar Python functions (#24297)
new 1e3649cd425 [fix][test] Fix more Netty ByteBuf leaks in tests (#24299)
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
...yKeyDispatcherMultipleConsumersClassicTest.java | 476 ++++++++++++---------
...ntStickyKeyDispatcherMultipleConsumersTest.java | 396 +++++++++--------
.../java/org/apache/pulsar/schema/SchemaTest.java | 20 +-
.../apache/pulsar/client/impl/ProducerImpl.java | 15 +-
.../impl/AcknowledgementsGroupingTrackerTest.java | 45 +-
.../client/impl/BinaryProtoLookupServiceTest.java | 15 +-
.../apache/pulsar/client/impl/ClientCnxTest.java | 55 +--
.../pulsar/client/impl/ClientTestFixtures.java | 92 +++-
.../instance/src/main/python/contextimpl.py | 2 +-
9 files changed, 650 insertions(+), 466 deletions(-)
(pulsar) 01/03: [fix][client] Fix producer publishing getting stuck after message with incompatible schema is discarded (#24282)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a66d6b786d32b733b3e8ecdb4d993a4a08197d0f
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Mon May 12 19:15:25 2025 +0800
[fix][client] Fix producer publishing getting stuck after message with incompatible schema is discarded (#24282)
Fixes #24262
Main Issue: #24262
### Motivation
The issue is a regression of https://github.com/apache/pulsar/pull/24178. The flow of the issue occurring is as follows
- Publish msg 1, which has an incompatible schema
- Publish msg 2, which has an incompatible schema
- The first message's schema failed to register because it is incompatible
- The message was discarded.
- Issue: it did not trigger a schema registration of the following messages
### Modifications
Fix the issue
(cherry picked from commit df2c619b14d277d8df82ce9a0ab1ab146cf51b7f)
---
.../java/org/apache/pulsar/schema/SchemaTest.java | 20 +++++++++++++++-----
.../org/apache/pulsar/client/impl/ProducerImpl.java | 15 +++++++++++----
2 files changed, 26 insertions(+), 9 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java
index 711e8ba5ad7..07c626a549d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java
@@ -1528,8 +1528,7 @@ public class SchemaTest extends MockedPulsarServiceBaseTest {
producer.close();
}
- // This test fails consistently, disabling until it is fixed. Issue https://github.com/apache/pulsar/issues/24262
- @Test(enabled = false)
+ @Test
public void testPendingQueueSizeIfIncompatible() throws Exception {
final String namespace = BrokerTestUtil.newUniqueName(PUBLIC_TENANT + "/ns");
admin.namespaces().createNamespace(namespace, Sets.newHashSet(CLUSTER_NAME));
@@ -1538,17 +1537,28 @@ public class SchemaTest extends MockedPulsarServiceBaseTest {
admin.topics().createNonPartitionedTopic(topic);
ProducerImpl producer = (ProducerImpl) pulsarClient.newProducer(Schema.AUTO_PRODUCE_BYTES())
- .maxPendingMessages(50).enableBatching(false).topic(topic).create();
- producer.newMessage(Schema.STRING).value("msg").sendAsync();
+ .maxPendingMessages(1000).enableBatching(false).topic(topic).create();
+ producer.newMessage(Schema.STRING).value("msg-1").sendAsync();
AtomicReference<CompletableFuture<MessageId>> latestSend = new AtomicReference<>();
for (int i = 0; i < 100; i++) {
- latestSend.set(producer.newMessage(Schema.BOOL).value(false).sendAsync());
+ final String msg = "msg-with-broken-schema-" + i;
+ latestSend.set(producer.newMessage(Schema.BOOL).value(false).sendAsync().thenApply(v -> {
+ log.info("send complete {}", msg);
+ return null;
+ }).exceptionally(ex -> {
+ log.error("failed to send {}", msg, ex);
+ return null;
+ }));
}
+ // Verify: msgs with broken schema will be discarded.
Awaitility.await().untilAsserted(() -> {
assertTrue(latestSend.get().isDone());
assertEquals(producer.getPendingQueueSize(), 0);
});
+ // Verify: msgs with compatible schema can be sent successfully.
+ producer.newMessage(Schema.STRING).value("msg-2").sendAsync();
+
// cleanup.
producer.close();
admin.topics().delete(topic, false);
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
index d1e89fba4b6..50a6f9aa81b 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -2456,6 +2456,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
* 3-1-1. If {@link #pauseSendingToPreservePublishOrderOnSchemaRegFailure} is true pause all following
* publishing to avoid out-of-order issue.
* 3-1-2. Otherwise, discard the failed message anc continuously publishing the following messages.
+ * Additionally, the following messages may need schema registration also.
* 3-2. The new schema registration failed due to other error, retry registering.
* Note: Since the current method accesses & modifies {@link #pendingMessages}, you should acquire a lock on
* {@link ProducerImpl} before calling method.
@@ -2474,6 +2475,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
Iterator<OpSendMsg> msgIterator = pendingMessages.iterator();
MessageImpl loopStartAt = latestMsgAttemptedRegisteredSchema;
OpSendMsg loopEndDueToSchemaRegisterNeeded = null;
+ boolean pausedSendingToPreservePublishOrderOnSchemaRegFailure = false;
while (msgIterator.hasNext()) {
OpSendMsg op = msgIterator.next();
if (loopStartAt != null) {
@@ -2518,6 +2520,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
+ " 2) Unload topic on target cluster. Schema details: {}",
topic, producerName, SchemaUtils.jsonifySchemaInfo(msgSchemaInfo, false));
loopEndDueToSchemaRegisterNeeded = op;
+ pausedSendingToPreservePublishOrderOnSchemaRegFailure = true;
break;
}
// Event 3-1-2.
@@ -2573,7 +2576,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
}
cnx.ctx().flush();
- // "Event 1-1" or "Event 3-1-1" or "Event 3-2".
+ // "Event 1-1" or "Event 3-1-1" or "Event 3-1-2" or "Event 3-2".
if (loopEndDueToSchemaRegisterNeeded != null) {
if (compareAndSetState(State.Connecting, State.Ready)) {
// "Event 1-1" happens after "Event 3-1-1".
@@ -2581,15 +2584,19 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
// after users changed the compatibility strategy to make the schema is compatible.
tryRegisterSchema(cnx, loopEndDueToSchemaRegisterNeeded.msg, loopEndDueToSchemaRegisterNeeded.callback,
expectedEpoch);
- } else if (!failedIncompatibleSchema && compareAndSetState(State.RegisteringSchema, State.Ready)) {
- // "Event 2-1" or "Event 3-2".
+ } else if (pausedSendingToPreservePublishOrderOnSchemaRegFailure) {
+ // Nothing to do if the event is "Event 3-1-1", just keep stuck.
+ return;
+ } else if (compareAndSetState(State.RegisteringSchema, State.Ready)) {
+ // "Event 2-1" or "Event 3-1-2" or "Event 3-2".
// "pendingMessages" has more messages to register new schema.
// This operation will not be conflict with another schema registration because both operations are
// attempt to acquire the same lock "ProducerImpl.this".
tryRegisterSchema(cnx, loopEndDueToSchemaRegisterNeeded.msg, loopEndDueToSchemaRegisterNeeded.callback,
expectedEpoch);
}
- // Nothing to do if the event is "Event 3-1-1", just keep stuck.
+ // Schema registration will trigger a new "recoverProcessOpSendMsgFrom", so return here. If failed to switch
+ // state, it means another task will trigger a new "recoverProcessOpSendMsgFrom".
return;
} else if (latestMsgAttemptedRegisteredSchema != null) {
// Event 2-2 or "Event 3-1-2".
(pulsar) 03/03: [fix][test] Fix more Netty ByteBuf leaks in tests (#24299)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 1e3649cd425cbac5841844933eb189f33e4bd501
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed May 14 10:00:26 2025 +0300
[fix][test] Fix more Netty ByteBuf leaks in tests (#24299)
(cherry picked from commit 174245d322574ef422f08a4c249db4e649010c83)
---
...yKeyDispatcherMultipleConsumersClassicTest.java | 476 ++++++++++++---------
...ntStickyKeyDispatcherMultipleConsumersTest.java | 396 +++++++++--------
.../impl/AcknowledgementsGroupingTrackerTest.java | 45 +-
.../client/impl/BinaryProtoLookupServiceTest.java | 15 +-
.../apache/pulsar/client/impl/ClientCnxTest.java | 55 +--
.../pulsar/client/impl/ClientTestFixtures.java | 92 +++-
6 files changed, 623 insertions(+), 456 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassicTest.java
index af42fc3dca4..286206fa361 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassicTest.java
@@ -27,10 +27,8 @@ import static org.mockito.Mockito.anyInt;
import static org.mockito.Mockito.anyList;
import static org.mockito.Mockito.anyLong;
import static org.mockito.Mockito.anySet;
-import static org.mockito.Mockito.argThat;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -40,21 +38,27 @@ import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
-import io.netty.channel.ChannelPromise;
import io.netty.channel.EventLoopGroup;
+import io.netty.util.concurrent.EventExecutor;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.SucceededFuture;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.Queue;
+import java.util.Objects;
import java.util.Set;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.TreeSet;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import org.apache.bookkeeper.common.util.OrderedExecutor;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
@@ -64,9 +68,12 @@ import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.EntryAndMetadata;
import org.apache.pulsar.broker.service.EntryBatchIndexesAcks;
import org.apache.pulsar.broker.service.EntryBatchSizes;
import org.apache.pulsar.broker.service.RedeliveryTracker;
+import org.apache.pulsar.broker.service.StickyKeyConsumerSelector;
+import org.apache.pulsar.broker.service.TransportCnx;
import org.apache.pulsar.broker.service.plugin.EntryFilterProvider;
import org.apache.pulsar.common.api.proto.KeySharedMeta;
import org.apache.pulsar.common.api.proto.KeySharedMode;
@@ -91,13 +98,14 @@ public class PersistentStickyKeyDispatcherMultipleConsumersClassicTest {
private PersistentTopic topicMock;
private PersistentSubscription subscriptionMock;
private ServiceConfiguration configMock;
- private ChannelPromise channelMock;
+ private Future<Void> succeededFuture;
private OrderedExecutor orderedExecutor;
private PersistentStickyKeyDispatcherMultipleConsumersClassic persistentDispatcher;
final String topicName = "persistent://public/default/testTopic";
final String subscriptionName = "testSubscription";
+ private AtomicInteger consumerMockAvailablePermits;
@BeforeMethod
public void setup() throws Exception {
@@ -128,7 +136,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumersClassicTest {
EventLoopGroup eventLoopGroup = mock(EventLoopGroup.class);
doReturn(eventLoopGroup).when(brokerMock).executor();
doAnswer(invocation -> {
- orderedExecutor.execute(((Runnable)invocation.getArguments()[0]));
+ orderedExecutor.execute(invocation.getArgument(0, Runnable.class));
return null;
}).when(eventLoopGroup).execute(any(Runnable.class));
@@ -141,12 +149,36 @@ public class PersistentStickyKeyDispatcherMultipleConsumersClassicTest {
doReturn(null).when(cursorMock).getLastIndividualDeletedRange();
doReturn(subscriptionName).when(cursorMock).getName();
- consumerMock = mock(Consumer.class);
- channelMock = mock(ChannelPromise.class);
+ consumerMock = createMockConsumer();
+ EventExecutor eventExecutor = mock(EventExecutor.class);
+ doAnswer(invocation -> {
+ orderedExecutor.execute(invocation.getArgument(0, Runnable.class));
+ return null;
+ }).when(eventExecutor).execute(any(Runnable.class));
+ doReturn(false).when(eventExecutor).inEventLoop();
+ succeededFuture = new SucceededFuture<>(eventExecutor, null);
doReturn("consumer1").when(consumerMock).consumerName();
- doReturn(1000).when(consumerMock).getAvailablePermits();
+ consumerMockAvailablePermits = new AtomicInteger(1000);
+ doAnswer(invocation -> consumerMockAvailablePermits.get()).when(consumerMock).getAvailablePermits();
doReturn(true).when(consumerMock).isWritable();
- doReturn(channelMock).when(consumerMock).sendMessages(
+ mockSendMessages(consumerMock, null);
+
+ subscriptionMock = mock(PersistentSubscription.class);
+ when(subscriptionMock.getTopic()).thenReturn(topicMock);
+ persistentDispatcher = new PersistentStickyKeyDispatcherMultipleConsumersClassic(
+ topicMock, cursorMock, subscriptionMock, configMock,
+ new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT));
+ }
+
+ private void mockSendMessages(Consumer consumerMock, java.util.function.Consumer<List<Entry>> entryConsumer) {
+ doAnswer(invocation -> {
+ List<Entry> entries = invocation.getArgument(0);
+ if (entryConsumer != null) {
+ entryConsumer.accept(entries);
+ }
+ entries.stream().filter(Objects::nonNull).forEach(Entry::release);
+ return succeededFuture;
+ }).when(consumerMock).sendMessages(
anyList(),
any(EntryBatchSizes.class),
any(EntryBatchIndexesAcks.class),
@@ -155,12 +187,16 @@ public class PersistentStickyKeyDispatcherMultipleConsumersClassicTest {
anyLong(),
any(RedeliveryTracker.class)
);
+ }
- subscriptionMock = mock(PersistentSubscription.class);
- when(subscriptionMock.getTopic()).thenReturn(topicMock);
- persistentDispatcher = new PersistentStickyKeyDispatcherMultipleConsumersClassic(
- topicMock, cursorMock, subscriptionMock, configMock,
- new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT));
+ protected static Consumer createMockConsumer() {
+ Consumer consumerMock = mock(Consumer.class);
+ TransportCnx transportCnx = mock(TransportCnx.class);
+ doReturn(transportCnx).when(consumerMock).cnx();
+ doReturn(true).when(transportCnx).isActive();
+ doReturn(100).when(consumerMock).getMaxUnackedMessages();
+ doReturn(1).when(consumerMock).getAvgMessagesPerEntry();
+ return consumerMock;
}
@AfterMethod(alwaysRun = true)
@@ -177,8 +213,8 @@ public class PersistentStickyKeyDispatcherMultipleConsumersClassicTest {
@Test(timeOut = 10000)
public void testAddConsumerWhenClosed() throws Exception {
persistentDispatcher.close().get();
- Consumer consumer = mock(Consumer.class);
- persistentDispatcher.addConsumer(consumer);
+ Consumer consumer = createMockConsumer();
+ persistentDispatcher.addConsumer(consumer).join();
verify(consumer, times(1)).disconnect();
assertEquals(0, persistentDispatcher.getConsumers().size());
assertTrue(persistentDispatcher.getSelector().getConsumerKeyHashRanges().isEmpty());
@@ -191,19 +227,19 @@ public class PersistentStickyKeyDispatcherMultipleConsumersClassicTest {
topicMock, cursorMock, subscriptionMock, configMock,
new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT));
- Consumer consumer0 = mock(Consumer.class);
+ Consumer consumer0 = createMockConsumer();
when(consumer0.consumerName()).thenReturn("c0-1");
- Consumer consumer1 = mock(Consumer.class);
+ Consumer consumer1 = createMockConsumer();
when(consumer1.consumerName()).thenReturn("c1");
- Consumer consumer2 = mock(Consumer.class);
+ Consumer consumer2 = createMockConsumer();
when(consumer2.consumerName()).thenReturn("c2");
- Consumer consumer3 = mock(Consumer.class);
+ Consumer consumer3 = createMockConsumer();
when(consumer3.consumerName()).thenReturn("c3");
- Consumer consumer4 = mock(Consumer.class);
+ Consumer consumer4 = createMockConsumer();
when(consumer4.consumerName()).thenReturn("c4");
- Consumer consumer5 = mock(Consumer.class);
+ Consumer consumer5 = createMockConsumer();
when(consumer5.consumerName()).thenReturn("c5");
- Consumer consumer6 = mock(Consumer.class);
+ Consumer consumer6 = createMockConsumer();
when(consumer6.consumerName()).thenReturn("c6");
when(cursorMock.getNumberOfEntriesSinceFirstNotAckedMessage()).thenReturn(100L);
@@ -266,7 +302,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumersClassicTest {
@Test
public void testSendMarkerMessage() {
try {
- persistentDispatcher.addConsumer(consumerMock);
+ persistentDispatcher.addConsumer(consumerMock).join();
persistentDispatcher.consumerFlow(consumerMock, 1000);
} catch (Exception e) {
fail("Failed to add mock consumer", e);
@@ -275,14 +311,16 @@ public class PersistentStickyKeyDispatcherMultipleConsumersClassicTest {
List<Entry> entries = new ArrayList<>();
ByteBuf markerMessage = Markers.newReplicatedSubscriptionsSnapshotRequest("testSnapshotId", "testSourceCluster");
entries.add(EntryImpl.create(1, 1, markerMessage));
- entries.add(EntryImpl.create(1, 2, createMessage("message1", 1)));
- entries.add(EntryImpl.create(1, 3, createMessage("message2", 2)));
- entries.add(EntryImpl.create(1, 4, createMessage("message3", 3)));
- entries.add(EntryImpl.create(1, 5, createMessage("message4", 4)));
- entries.add(EntryImpl.create(1, 6, createMessage("message5", 5)));
+ markerMessage.release();
+ entries.add(createEntry(1, 2, "message1", 1));
+ entries.add(createEntry(1, 3, "message2", 2));
+ entries.add(createEntry(1, 4, "message3", 3));
+ entries.add(createEntry(1, 5, "message4", 4));
+ entries.add(createEntry(1, 6, "message5", 5));
try {
- persistentDispatcher.readEntriesComplete(entries, PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Normal);
+ persistentDispatcher.readEntriesComplete(copyEntries(entries),
+ PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Normal);
} catch (Exception e) {
fail("Failed to readEntriesComplete.", e);
}
@@ -302,6 +340,13 @@ public class PersistentStickyKeyDispatcherMultipleConsumersClassicTest {
List<Integer> allTotalMessagesCaptor = totalMessagesCaptor.getAllValues();
Assert.assertEquals(allTotalMessagesCaptor.get(0).intValue(), 5);
});
+
+ entries.forEach(Entry::release);
+ }
+
+ private static List<Entry> copyEntries(List<Entry> entries) {
+ return entries.stream().map(entry -> EntryImpl.create((EntryImpl) entry))
+ .collect(Collectors.toList());
}
@Test(timeOut = 10000)
@@ -315,231 +360,244 @@ public class PersistentStickyKeyDispatcherMultipleConsumersClassicTest {
.setStart(0)
.setEnd(9);
- Consumer consumerMock = mock(Consumer.class);
+ Consumer consumerMock = createMockConsumer();
doReturn(keySharedMeta).when(consumerMock).getKeySharedMeta();
- persistentDispatcher.addConsumer(consumerMock);
+ mockSendMessages(consumerMock, null);
+ persistentDispatcher.addConsumer(consumerMock).join();
persistentDispatcher.consumerFlow(consumerMock, 1000);
} catch (Exception e) {
fail("Failed to add mock consumer", e);
}
List<Entry> entries = new ArrayList<>();
- entries.add(EntryImpl.create(1, 1, createMessage("message1", 1)));
- entries.add(EntryImpl.create(1, 2, createMessage("message2", 2)));
+ entries.add(createEntry(1, 1, "message1", 1));
+ entries.add(createEntry(1, 2, "message2", 2));
try {
//Should success,see issue #8960
- persistentDispatcher.readEntriesComplete(entries, PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Normal);
+ persistentDispatcher.readEntriesComplete(copyEntries(entries),
+ PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Normal);
} catch (Exception e) {
fail("Failed to readEntriesComplete.", e);
}
+
+ entries.forEach(Entry::release);
}
@Test
- public void testSkipRedeliverTemporally() {
- final Consumer slowConsumerMock = mock(Consumer.class);
- final ChannelPromise slowChannelMock = mock(ChannelPromise.class);
- // add entries to redeliver and read target
+ public void testSkipRedeliverTemporally() throws InterruptedException {
+ // add first consumer
+ persistentDispatcher.addConsumer(consumerMock).join();
+ // add slow consumer
+ final Consumer slowConsumerMock = createMockConsumer();
+ doReturn("consumer2").when(slowConsumerMock).consumerName();
+ AtomicInteger slowConsumerAvailablePermits = new AtomicInteger(0);
+ doAnswer(invocation -> {
+ return slowConsumerAvailablePermits.get();
+ }).when(slowConsumerMock).getAvailablePermits();
+ persistentDispatcher.addConsumer(slowConsumerMock).join();
+
+ StickyKeyConsumerSelector selector = persistentDispatcher.getSelector();
+ String keyForConsumer = generateKeyForConsumer(selector, consumerMock);
+ String keyForSlowConsumer = generateKeyForConsumer(selector, slowConsumerMock);
+
+ Set<Position> alreadySent = new ConcurrentSkipListSet<>();
+
+ final List<Entry> allEntries = new ArrayList<>();
+ allEntries.add(createEntry(1, 1, "message1", 1, keyForSlowConsumer));
+ allEntries.add(createEntry(1, 2, "message2", 2, keyForSlowConsumer));
+ allEntries.add(createEntry(1, 3, "message3", 3, keyForConsumer));
+
+ // add first entry to redeliver initially
final List<Entry> redeliverEntries = new ArrayList<>();
- redeliverEntries.add(EntryImpl.create(1, 1, createMessage("message1", 1, "key1")));
- final List<Entry> readEntries = new ArrayList<>();
- readEntries.add(EntryImpl.create(1, 2, createMessage("message2", 2, "key1")));
- readEntries.add(EntryImpl.create(1, 3, createMessage("message3", 3, "key2")));
+ redeliverEntries.add(allEntries.get(0));
try {
- Field totalAvailablePermitsField = PersistentDispatcherMultipleConsumersClassic.class.getDeclaredField("totalAvailablePermits");
+ Field totalAvailablePermitsField =
+ PersistentDispatcherMultipleConsumersClassic.class.getDeclaredField("totalAvailablePermits");
totalAvailablePermitsField.setAccessible(true);
totalAvailablePermitsField.set(persistentDispatcher, 1000);
-
- doAnswer(invocationOnMock -> {
- ((PersistentStickyKeyDispatcherMultipleConsumersClassic) invocationOnMock.getArgument(2))
- .readEntriesComplete(readEntries, PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Normal);
- return null;
- }).when(cursorMock).asyncReadEntriesOrWait(
- anyInt(), anyLong(), any(PersistentStickyKeyDispatcherMultipleConsumersClassic.class),
- eq(PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Normal), any());
} catch (Exception e) {
fail("Failed to set to field", e);
}
- // Create 2Consumers
- try {
- doReturn("consumer2").when(slowConsumerMock).consumerName();
- // Change slowConsumer availablePermits to 0 and back to normal
- when(slowConsumerMock.getAvailablePermits())
- .thenReturn(0)
- .thenReturn(1);
- doReturn(true).when(slowConsumerMock).isWritable();
- doReturn(slowChannelMock).when(slowConsumerMock).sendMessages(
- anyList(),
- any(EntryBatchSizes.class),
- any(EntryBatchIndexesAcks.class),
- anyInt(),
- anyLong(),
- anyLong(),
- any(RedeliveryTracker.class)
- );
-
- persistentDispatcher.addConsumer(consumerMock);
- persistentDispatcher.addConsumer(slowConsumerMock);
- } catch (Exception e) {
- fail("Failed to add mock consumer", e);
- }
+ // Mock Cursor#asyncReplayEntries
+ doAnswer(invocationOnMock -> {
+ Set<Position> positionsArg = invocationOnMock.getArgument(0);
+ Set<Position> positions = new TreeSet<>(positionsArg);
+ List<Entry> entries = allEntries.stream()
+ .filter(entry -> entry.getLedgerId() != -1 && positions.contains(entry.getPosition()))
+ .toList();
+ AsyncCallbacks.ReadEntriesCallback callback = invocationOnMock.getArgument(1);
+ Object ctx = invocationOnMock.getArgument(2);
+ callback.readEntriesComplete(copyEntries(entries), ctx);
+ return Collections.emptySet();
+ }).when(cursorMock).asyncReplayEntries(anySet(), any(), any(), anyBoolean());
- // run PersistentStickyKeyDispatcherMultipleConsumers#sendMessagesToConsumers
- // run readMoreEntries internally (and skip internally)
- // Change slowConsumer availablePermits to 1
- // run PersistentStickyKeyDispatcherMultipleConsumers#sendMessagesToConsumers internally
- // and then stop to dispatch to slowConsumer
- if (persistentDispatcher.sendMessagesToConsumers(PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Normal,
- redeliverEntries, true)) {
- persistentDispatcher.readMoreEntriesAsync();
- }
+ doAnswer(invocationOnMock -> {
+ int maxEntries = invocationOnMock.getArgument(0);
+ AsyncCallbacks.ReadEntriesCallback callback = invocationOnMock.getArgument(2);
+ List<Entry> entries = allEntries.stream()
+ .filter(entry -> entry.getLedgerId() != -1 && !alreadySent.contains(entry.getPosition()))
+ .limit(maxEntries)
+ .toList();
+ Object ctx = invocationOnMock.getArgument(3);
+ callback.readEntriesComplete(copyEntries(entries), ctx);
+ return null;
+ }).when(cursorMock).asyncReadEntriesOrWait(anyInt(), anyLong(), any(), any(), any());
+
+ doReturn(true).when(slowConsumerMock).isWritable();
+ CountDownLatch message3Sent = new CountDownLatch(1);
+ mockSendMessages(consumerMock, entries -> {
+ entries.forEach(entry -> {
+ alreadySent.add(entry.getPosition());
+ });
+ boolean message3Found = entries.stream()
+ .anyMatch(entry -> entry.getLedgerId() == 1 && entry.getEntryId() == 3);
+ if (message3Found) {
+ message3Sent.countDown();
+ }
+ });
+ CountDownLatch slowConsumerMessagesSent = new CountDownLatch(2);
+ mockSendMessages(slowConsumerMock, entries -> {
+ entries.forEach(entry -> {
+ alreadySent.add(entry.getPosition());
+ slowConsumerMessagesSent.countDown();
+ });
+ });
- Awaitility.await().untilAsserted(() -> {
- verify(consumerMock, times(1)).sendMessages(
- argThat(arg -> {
- assertEquals(arg.size(), 1);
- Entry entry = arg.get(0);
- assertEquals(entry.getLedgerId(), 1);
- assertEquals(entry.getEntryId(), 3);
- return true;
- }),
- any(EntryBatchSizes.class),
- any(EntryBatchIndexesAcks.class),
- anyInt(),
- anyLong(),
- anyLong(),
- any(RedeliveryTracker.class)
- );
+ // add entries to redeliver
+ redeliverEntries.forEach(entry -> {
+ // calculate hash
+ EntryAndMetadata entryAndMetadata = EntryAndMetadata.create(entry);
+ int stickyKeyHash = selector.makeStickyKeyHash(entryAndMetadata.getStickyKey());
+ // add to redeliver
+ persistentDispatcher.addMessageToReplay(entry.getLedgerId(), entry.getEntryId(), stickyKeyHash);
});
- verify(slowConsumerMock, times(0)).sendMessages(
- anyList(),
- any(EntryBatchSizes.class),
- any(EntryBatchIndexesAcks.class),
- anyInt(),
- anyLong(),
- anyLong(),
- any(RedeliveryTracker.class)
- );
+
+ // trigger readMoreEntries, will handle redelivery logic and skip slow consumer
+ persistentDispatcher.readMoreEntriesAsync();
+
+ assertTrue(message3Sent.await(5, TimeUnit.SECONDS));
+
+ // verify that slow consumer messages are not sent before message3 to "consumer"
+ assertEquals(slowConsumerMessagesSent.getCount(), 2);
+
+ // set permits to 2
+ slowConsumerAvailablePermits.set(2);
+
+ // now wait for slow consumer messages since there are permits
+ assertTrue(slowConsumerMessagesSent.await(5, TimeUnit.SECONDS));
+
+ allEntries.forEach(Entry::release);
}
@Test(timeOut = 30000)
public void testMessageRedelivery() throws Exception {
- final Queue<Position> actualEntriesToConsumer1 = new ConcurrentLinkedQueue<>();
- final Queue<Position> actualEntriesToConsumer2 = new ConcurrentLinkedQueue<>();
+ final List<Position> actualEntriesToConsumer1 = new CopyOnWriteArrayList<>();
+ final List<Position> actualEntriesToConsumer2 = new CopyOnWriteArrayList<>();
- final Queue<Position> expectedEntriesToConsumer1 = new ConcurrentLinkedQueue<>();
- expectedEntriesToConsumer1.add(PositionFactory.create(1, 1));
- final Queue<Position> expectedEntriesToConsumer2 = new ConcurrentLinkedQueue<>();
- expectedEntriesToConsumer2.add(PositionFactory.create(1, 2));
- expectedEntriesToConsumer2.add(PositionFactory.create(1, 3));
+ final List<Position> expectedEntriesToConsumer1 = new CopyOnWriteArrayList<>();
+ final List<Position> expectedEntriesToConsumer2 = new CopyOnWriteArrayList<>();
- final AtomicInteger remainingEntriesNum = new AtomicInteger(
- expectedEntriesToConsumer1.size() + expectedEntriesToConsumer2.size());
+ final CountDownLatch remainingEntriesNum = new CountDownLatch(3);
- // Messages with key1 are routed to consumer1 and messages with key2 are routed to consumer2
- final List<Entry> allEntries = new ArrayList<>();
- allEntries.add(EntryImpl.create(1, 1, createMessage("message1", 1, "key2")));
- allEntries.add(EntryImpl.create(1, 2, createMessage("message2", 2, "key1")));
- allEntries.add(EntryImpl.create(1, 3, createMessage("message3", 3, "key1")));
- allEntries.forEach(entry -> ((EntryImpl) entry).retain());
-
- final List<Entry> redeliverEntries = new ArrayList<>();
- redeliverEntries.add(allEntries.get(0)); // message1
- final List<Entry> readEntries = new ArrayList<>();
- readEntries.add(allEntries.get(2)); // message3
-
- final Consumer consumer1 = mock(Consumer.class);
+ final Consumer consumer1 = createMockConsumer();
doReturn("consumer1").when(consumer1).consumerName();
// Change availablePermits of consumer1 to 0 and then back to normal
when(consumer1.getAvailablePermits()).thenReturn(0).thenReturn(10);
doReturn(true).when(consumer1).isWritable();
- doAnswer(invocationOnMock -> {
- @SuppressWarnings("unchecked")
- List<Entry> entries = (List<Entry>) invocationOnMock.getArgument(0);
+ mockSendMessages(consumer1, entries -> {
for (Entry entry : entries) {
- remainingEntriesNum.decrementAndGet();
actualEntriesToConsumer1.add(entry.getPosition());
+ remainingEntriesNum.countDown();
}
- return channelMock;
- }).when(consumer1).sendMessages(anyList(), any(EntryBatchSizes.class), any(EntryBatchIndexesAcks.class),
- anyInt(), anyLong(), anyLong(), any(RedeliveryTracker.class));
+ });
- final Consumer consumer2 = mock(Consumer.class);
+ final Consumer consumer2 = createMockConsumer();
doReturn("consumer2").when(consumer2).consumerName();
when(consumer2.getAvailablePermits()).thenReturn(10);
doReturn(true).when(consumer2).isWritable();
- doAnswer(invocationOnMock -> {
- @SuppressWarnings("unchecked")
- List<Entry> entries = (List<Entry>) invocationOnMock.getArgument(0);
+ mockSendMessages(consumer2, entries -> {
for (Entry entry : entries) {
- remainingEntriesNum.decrementAndGet();
actualEntriesToConsumer2.add(entry.getPosition());
+ remainingEntriesNum.countDown();
}
- return channelMock;
- }).when(consumer2).sendMessages(anyList(), any(EntryBatchSizes.class), any(EntryBatchIndexesAcks.class),
- anyInt(), anyLong(), anyLong(), any(RedeliveryTracker.class));
+ });
- persistentDispatcher.addConsumer(consumer1);
- persistentDispatcher.addConsumer(consumer2);
+ persistentDispatcher.addConsumer(consumer1).join();
+ persistentDispatcher.addConsumer(consumer2).join();
final Field totalAvailablePermitsField = PersistentDispatcherMultipleConsumersClassic.class
.getDeclaredField("totalAvailablePermits");
totalAvailablePermitsField.setAccessible(true);
totalAvailablePermitsField.set(persistentDispatcher, 1000);
- final Field redeliveryMessagesField = PersistentDispatcherMultipleConsumersClassic.class
- .getDeclaredField("redeliveryMessages");
- redeliveryMessagesField.setAccessible(true);
- MessageRedeliveryController redeliveryMessages = (MessageRedeliveryController) redeliveryMessagesField
- .get(persistentDispatcher);
- redeliveryMessages.add(allEntries.get(0).getLedgerId(), allEntries.get(0).getEntryId(),
- persistentDispatcher.getStickyKeyHash(allEntries.get(0))); // message1
- redeliveryMessages.add(allEntries.get(1).getLedgerId(), allEntries.get(1).getEntryId(),
- persistentDispatcher.getStickyKeyHash(allEntries.get(1))); // message2
+ StickyKeyConsumerSelector selector = persistentDispatcher.getSelector();
+
+ String keyForConsumer1 = generateKeyForConsumer(selector, consumer1);
+ String keyForConsumer2 = generateKeyForConsumer(selector, consumer2);
+
+ // Messages with key1 are routed to consumer1 and messages with key2 are routed to consumer2
+ final List<Entry> allEntries = new ArrayList<>();
+ allEntries.add(createEntry(1, 1, "message1", 1, keyForConsumer1));
+ allEntries.add(createEntry(1, 2, "message2", 2, keyForConsumer1));
+ allEntries.add(createEntry(1, 3, "message3", 3, keyForConsumer2));
+
+ // add first entry to redeliver initially
+ final List<Entry> redeliverEntries = new ArrayList<>();
+ redeliverEntries.add(allEntries.get(0)); // message1
+
+ expectedEntriesToConsumer1.add(allEntries.get(0).getPosition());
+ expectedEntriesToConsumer1.add(allEntries.get(1).getPosition());
+ expectedEntriesToConsumer2.add(allEntries.get(2).getPosition());
// Mock Cursor#asyncReplayEntries
doAnswer(invocationOnMock -> {
- @SuppressWarnings("unchecked")
- Set<Position> positions = (Set<Position>) invocationOnMock.getArgument(0);
- List<Entry> entries = allEntries.stream().filter(entry -> positions.contains(entry.getPosition()))
+ Set<Position> positionsArg = invocationOnMock.getArgument(0);
+ Set<Position> positions = new TreeSet<>(positionsArg);
+ Set<Position> alreadyReceived = new TreeSet<>();
+ alreadyReceived.addAll(actualEntriesToConsumer1);
+ alreadyReceived.addAll(actualEntriesToConsumer2);
+ List<Entry> entries = allEntries.stream().filter(entry -> entry.getLedgerId() != -1
+ && positions.contains(entry.getPosition())
+ && !alreadyReceived.contains(entry.getPosition()))
.collect(Collectors.toList());
- if (!entries.isEmpty()) {
- ((PersistentStickyKeyDispatcherMultipleConsumersClassic) invocationOnMock.getArgument(1))
- .readEntriesComplete(entries, PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Replay);
- }
- return Collections.emptySet();
- }).when(cursorMock).asyncReplayEntries(anySet(), any(PersistentStickyKeyDispatcherMultipleConsumersClassic.class),
- eq(PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Replay), anyBoolean());
+ AsyncCallbacks.ReadEntriesCallback callback = invocationOnMock.getArgument(1);
+ Object ctx = invocationOnMock.getArgument(2);
+ callback.readEntriesComplete(copyEntries(entries), ctx);
+ return alreadyReceived;
+ }).when(cursorMock).asyncReplayEntries(anySet(), any(), any(), anyBoolean());
// Mock Cursor#asyncReadEntriesOrWait
- AtomicBoolean asyncReadEntriesOrWaitCalled = new AtomicBoolean();
doAnswer(invocationOnMock -> {
- if (asyncReadEntriesOrWaitCalled.compareAndSet(false, true)) {
- ((PersistentStickyKeyDispatcherMultipleConsumersClassic) invocationOnMock.getArgument(2))
- .readEntriesComplete(readEntries, PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Normal);
- } else {
- ((PersistentStickyKeyDispatcherMultipleConsumersClassic) invocationOnMock.getArgument(2))
- .readEntriesComplete(Collections.emptyList(), PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Normal);
- }
+ int maxEntries = invocationOnMock.getArgument(0);
+ Set<Position> alreadyReceived = new TreeSet<>();
+ alreadyReceived.addAll(actualEntriesToConsumer1);
+ alreadyReceived.addAll(actualEntriesToConsumer2);
+ List<Entry> entries = allEntries.stream()
+ .filter(entry -> entry.getLedgerId() != -1 && !alreadyReceived.contains(entry.getPosition()))
+ .limit(maxEntries)
+ .collect(Collectors.toList());
+ AsyncCallbacks.ReadEntriesCallback callback = invocationOnMock.getArgument(2);
+ Object ctx = invocationOnMock.getArgument(3);
+ callback.readEntriesComplete(copyEntries(entries), ctx);
return null;
- }).when(cursorMock).asyncReadEntriesOrWait(anyInt(), anyLong(),
- any(PersistentStickyKeyDispatcherMultipleConsumersClassic.class),
- eq(PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Normal), any());
-
- // (1) Run sendMessagesToConsumers
- // (2) Attempts to send message1 to consumer1 but skipped because availablePermits is 0
- // (3) Change availablePermits of consumer1 to 10
- // (4) Run readMoreEntries internally
- // (5) Run sendMessagesToConsumers internally
- // (6) Attempts to send message3 to consumer2 but skipped because redeliveryMessages contains message2
- persistentDispatcher.sendMessagesToConsumers(PersistentStickyKeyDispatcherMultipleConsumersClassic.ReadType.Replay,
- redeliverEntries, true);
- while (remainingEntriesNum.get() > 0) {
- // (7) Run readMoreEntries and resend message1 to consumer1 and message2-3 to consumer2
- persistentDispatcher.readMoreEntries();
- }
+ }).when(cursorMock).asyncReadEntriesOrWait(anyInt(), anyLong(), any(), any(), any());
+
+ // add entries to redeliver
+ redeliverEntries.forEach(entry -> {
+ // calculate hash
+ EntryAndMetadata entryAndMetadata = EntryAndMetadata.create(entry);
+ int stickyKeyHash = selector.makeStickyKeyHash(entryAndMetadata.getStickyKey());
+ // add to redeliver
+ persistentDispatcher.addMessageToReplay(entry.getLedgerId(), entry.getEntryId(), stickyKeyHash);
+ });
+
+ // trigger logic to read entries, includes redelivery logic
+ persistentDispatcher.readMoreEntriesAsync();
+
+ assertTrue(remainingEntriesNum.await(5, TimeUnit.SECONDS));
assertThat(actualEntriesToConsumer1).containsExactlyElementsOf(expectedEntriesToConsumer1);
assertThat(actualEntriesToConsumer2).containsExactlyElementsOf(expectedEntriesToConsumer2);
@@ -547,17 +605,39 @@ public class PersistentStickyKeyDispatcherMultipleConsumersClassicTest {
allEntries.forEach(entry -> entry.release());
}
- private ByteBuf createMessage(String message, int sequenceId) {
- return createMessage(message, sequenceId, "testKey");
+ private String generateKeyForConsumer(StickyKeyConsumerSelector selector, Consumer consumer) {
+ int i = 0;
+ while (!Thread.currentThread().isInterrupted()) {
+ String key = "key" + i++;
+ Consumer selectedConsumer = selector.select(key.getBytes(UTF_8));
+ if (selectedConsumer == consumer) {
+ return key;
+ }
+ }
+ return null;
+ }
+
+ private EntryImpl createEntry(long ledgerId, long entryId, String message, long sequenceId) {
+ return createEntry(ledgerId, entryId, message, sequenceId, "testKey");
+ }
+
+ private EntryImpl createEntry(long ledgerId, long entryId, String message, long sequenceId, String key) {
+ ByteBuf data = createMessage(message, sequenceId, key);
+ EntryImpl entry = EntryImpl.create(ledgerId, entryId, data);
+ data.release();
+ return entry;
}
- private ByteBuf createMessage(String message, int sequenceId, String key) {
+ private ByteBuf createMessage(String message, long sequenceId, String key) {
MessageMetadata messageMetadata = new MessageMetadata()
.setSequenceId(sequenceId)
.setProducerName("testProducer")
.setPartitionKey(key)
.setPartitionKeyB64Encoded(false)
.setPublishTime(System.currentTimeMillis());
- return serializeMetadataAndPayload(Commands.ChecksumType.Crc32c, messageMetadata, Unpooled.copiedBuffer(message.getBytes(UTF_8)));
+ ByteBuf payload = Unpooled.copiedBuffer(message.getBytes(UTF_8));
+ ByteBuf byteBuf = serializeMetadataAndPayload(Commands.ChecksumType.Crc32c, messageMetadata, payload);
+ payload.release();
+ return byteBuf;
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java
index aec161f237b..1e034d1442d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java
@@ -27,10 +27,8 @@ import static org.mockito.Mockito.anyInt;
import static org.mockito.Mockito.anyList;
import static org.mockito.Mockito.anyLong;
import static org.mockito.Mockito.anySet;
-import static org.mockito.Mockito.argThat;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -43,23 +41,26 @@ import com.google.common.collect.BoundType;
import com.google.common.collect.Range;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
-import io.netty.channel.ChannelPromise;
import io.netty.channel.EventLoopGroup;
+import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.Future;
-import io.netty.util.concurrent.GenericFutureListener;
+import io.netty.util.concurrent.SucceededFuture;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import java.util.Queue;
+import java.util.Objects;
import java.util.Set;
import java.util.TreeSet;
-import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import org.apache.bookkeeper.common.util.OrderedExecutor;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
@@ -70,9 +71,9 @@ import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.Consumer;
-import org.apache.pulsar.broker.service.EntryAndMetadata;
import org.apache.pulsar.broker.service.EntryBatchIndexesAcks;
import org.apache.pulsar.broker.service.EntryBatchSizes;
+import org.apache.pulsar.broker.service.PendingAcksMap;
import org.apache.pulsar.broker.service.RedeliveryTracker;
import org.apache.pulsar.broker.service.StickyKeyConsumerSelector;
import org.apache.pulsar.broker.service.TransportCnx;
@@ -102,7 +103,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
private PersistentTopic topicMock;
private PersistentSubscription subscriptionMock;
private ServiceConfiguration configMock;
- private ChannelPromise channelMock;
+ private Future<Void> succeededFuture;
private OrderedExecutor orderedExecutor;
private PersistentStickyKeyDispatcherMultipleConsumers persistentDispatcher;
@@ -144,7 +145,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
EventLoopGroup eventLoopGroup = mock(EventLoopGroup.class);
doReturn(eventLoopGroup).when(brokerMock).executor();
doAnswer(invocation -> {
- orderedExecutor.execute(((Runnable)invocation.getArguments()[0]));
+ orderedExecutor.execute(invocation.getArgument(0, Runnable.class));
return null;
}).when(eventLoopGroup).execute(any(Runnable.class));
@@ -197,12 +198,35 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
}).when(cursorMock).applyMaxSizeCap(anyInt(), anyLong());
consumerMock = createMockConsumer();
- channelMock = mock(ChannelPromise.class);
+ EventExecutor eventExecutor = mock(EventExecutor.class);
+ doAnswer(invocation -> {
+ orderedExecutor.execute(invocation.getArgument(0, Runnable.class));
+ return null;
+ }).when(eventExecutor).execute(any(Runnable.class));
+ doReturn(false).when(eventExecutor).inEventLoop();
+ succeededFuture = new SucceededFuture<>(eventExecutor, null);
doReturn("consumer1").when(consumerMock).consumerName();
consumerMockAvailablePermits = new AtomicInteger(1000);
doAnswer(invocation -> consumerMockAvailablePermits.get()).when(consumerMock).getAvailablePermits();
doReturn(true).when(consumerMock).isWritable();
- doReturn(channelMock).when(consumerMock).sendMessages(
+ mockSendMessages(consumerMock, null);
+
+ subscriptionMock = mock(PersistentSubscription.class);
+ when(subscriptionMock.getTopic()).thenReturn(topicMock);
+ persistentDispatcher = new PersistentStickyKeyDispatcherMultipleConsumers(
+ topicMock, cursorMock, subscriptionMock, configMock,
+ new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT));
+ }
+
+ private void mockSendMessages(Consumer consumerMock, java.util.function.Consumer<List<Entry>> entryConsumer) {
+ doAnswer(invocation -> {
+ List<Entry> entries = invocation.getArgument(0);
+ if (entryConsumer != null) {
+ entryConsumer.accept(entries);
+ }
+ entries.stream().filter(Objects::nonNull).forEach(Entry::release);
+ return succeededFuture;
+ }).when(consumerMock).sendMessages(
anyList(),
any(EntryBatchSizes.class),
any(EntryBatchIndexesAcks.class),
@@ -211,12 +235,6 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
anyLong(),
any(RedeliveryTracker.class)
);
-
- subscriptionMock = mock(PersistentSubscription.class);
- when(subscriptionMock.getTopic()).thenReturn(topicMock);
- persistentDispatcher = new PersistentStickyKeyDispatcherMultipleConsumers(
- topicMock, cursorMock, subscriptionMock, configMock,
- new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT));
}
protected static Consumer createMockConsumer() {
@@ -226,6 +244,8 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
doReturn(true).when(transportCnx).isActive();
doReturn(100).when(consumerMock).getMaxUnackedMessages();
doReturn(1).when(consumerMock).getAvgMessagesPerEntry();
+ PendingAcksMap pendingAcksMap = mock(PendingAcksMap.class);
+ doReturn(pendingAcksMap).when(consumerMock).getPendingAcks();
return consumerMock;
}
@@ -244,7 +264,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
public void testAddConsumerWhenClosed() throws Exception {
persistentDispatcher.close().get();
Consumer consumer = createMockConsumer();
- persistentDispatcher.addConsumer(consumer);
+ persistentDispatcher.addConsumer(consumer).join();
verify(consumer, times(1)).disconnect();
assertEquals(0, persistentDispatcher.getConsumers().size());
assertTrue(persistentDispatcher.getSelector().getConsumerKeyHashRanges().isEmpty());
@@ -253,7 +273,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
@Test
public void testSendMarkerMessage() {
try {
- persistentDispatcher.addConsumer(consumerMock);
+ persistentDispatcher.addConsumer(consumerMock).join();
persistentDispatcher.consumerFlow(consumerMock, 1000);
} catch (Exception e) {
fail("Failed to add mock consumer", e);
@@ -262,14 +282,16 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
List<Entry> entries = new ArrayList<>();
ByteBuf markerMessage = Markers.newReplicatedSubscriptionsSnapshotRequest("testSnapshotId", "testSourceCluster");
entries.add(EntryImpl.create(1, 1, markerMessage));
- entries.add(EntryImpl.create(1, 2, createMessage("message1", 1)));
- entries.add(EntryImpl.create(1, 3, createMessage("message2", 2)));
- entries.add(EntryImpl.create(1, 4, createMessage("message3", 3)));
- entries.add(EntryImpl.create(1, 5, createMessage("message4", 4)));
- entries.add(EntryImpl.create(1, 6, createMessage("message5", 5)));
+ markerMessage.release();
+ entries.add(createEntry(1, 2, "message1", 1));
+ entries.add(createEntry(1, 3, "message2", 2));
+ entries.add(createEntry(1, 4, "message3", 3));
+ entries.add(createEntry(1, 5, "message4", 4));
+ entries.add(createEntry(1, 6, "message5", 5));
try {
- persistentDispatcher.readEntriesComplete(entries, PersistentStickyKeyDispatcherMultipleConsumers.ReadType.Normal);
+ persistentDispatcher.readEntriesComplete(copyEntries(entries),
+ PersistentStickyKeyDispatcherMultipleConsumers.ReadType.Normal);
} catch (Exception e) {
fail("Failed to readEntriesComplete.", e);
}
@@ -289,6 +311,13 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
List<Integer> allTotalMessagesCaptor = totalMessagesCaptor.getAllValues();
Assert.assertEquals(allTotalMessagesCaptor.get(0).intValue(), 5);
});
+
+ entries.forEach(Entry::release);
+ }
+
+ private static List<Entry> copyEntries(List<Entry> entries) {
+ return entries.stream().map(entry -> EntryImpl.create((EntryImpl) entry))
+ .collect(Collectors.toList());
}
@Test(timeOut = 10000)
@@ -303,148 +332,168 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
Consumer consumerMock = createMockConsumer();
doReturn(keySharedMeta).when(consumerMock).getKeySharedMeta();
- persistentDispatcher.addConsumer(consumerMock);
+ mockSendMessages(consumerMock, null);
+ persistentDispatcher.addConsumer(consumerMock).join();
persistentDispatcher.consumerFlow(consumerMock, 1000);
} catch (Exception e) {
fail("Failed to add mock consumer", e);
}
List<Entry> entries = new ArrayList<>();
- entries.add(EntryImpl.create(1, 1, createMessage("message1", 1)));
- entries.add(EntryImpl.create(1, 2, createMessage("message2", 2)));
+ entries.add(createEntry(1, 1, "message1", 1));
+ entries.add(createEntry(1, 2, "message2", 2));
try {
//Should success,see issue #8960
- persistentDispatcher.readEntriesComplete(entries, PersistentStickyKeyDispatcherMultipleConsumers.ReadType.Normal);
+ persistentDispatcher.readEntriesComplete(copyEntries(entries),
+ PersistentStickyKeyDispatcherMultipleConsumers.ReadType.Normal);
} catch (Exception e) {
fail("Failed to readEntriesComplete.", e);
}
+
+ entries.forEach(Entry::release);
}
@Test
- public void testSkipRedeliverTemporally() {
+ public void testSkipRedeliverTemporally() throws InterruptedException {
+ // add first consumer
+ persistentDispatcher.addConsumer(consumerMock).join();
+ // add slow consumer
final Consumer slowConsumerMock = createMockConsumer();
- AtomicInteger slowConsumerPermits = new AtomicInteger(0);
- doAnswer(invocation -> slowConsumerPermits.get()).when(slowConsumerMock).getAvailablePermits();
+ doReturn("consumer2").when(slowConsumerMock).consumerName();
+ AtomicInteger slowConsumerAvailablePermits = new AtomicInteger(0);
+ doAnswer(invocation -> {
+ return slowConsumerAvailablePermits.get();
+ }).when(slowConsumerMock).getAvailablePermits();
+ persistentDispatcher.addConsumer(slowConsumerMock).join();
- final ChannelPromise slowChannelMock = mock(ChannelPromise.class);
- // add entries to redeliver and read target
+ StickyKeyConsumerSelector selector = persistentDispatcher.getSelector();
+ String keyForConsumer = generateKeyForConsumer(selector, consumerMock);
+ String keyForSlowConsumer = generateKeyForConsumer(selector, slowConsumerMock);
+
+ Set<Position> alreadySent = new ConcurrentSkipListSet<>();
+
+ final List<Entry> allEntries = new ArrayList<>();
+ allEntries.add(createEntry(1, 1, "message1", 1, keyForSlowConsumer));
+ allEntries.add(createEntry(1, 2, "message2", 2, keyForSlowConsumer));
+ allEntries.add(createEntry(1, 3, "message3", 3, keyForConsumer));
+
+ // add first entry to redeliver initially
final List<Entry> redeliverEntries = new ArrayList<>();
- redeliverEntries.add(EntryImpl.create(1, 1, createMessage("message1", 1, "key123")));
- final List<Entry> readEntries = new ArrayList<>();
- readEntries.add(EntryImpl.create(1, 2, createMessage("message2", 2, "key123")));
- readEntries.add(EntryImpl.create(1, 3, createMessage("message3", 3, "key222")));
+ redeliverEntries.add(allEntries.get(0));
try {
Field totalAvailablePermitsField = PersistentDispatcherMultipleConsumers.class.getDeclaredField("totalAvailablePermits");
totalAvailablePermitsField.setAccessible(true);
totalAvailablePermitsField.set(persistentDispatcher, 1000);
-
- doAnswer(invocationOnMock -> {
- ((PersistentStickyKeyDispatcherMultipleConsumers) invocationOnMock.getArgument(2))
- .readEntriesComplete(readEntries, PersistentStickyKeyDispatcherMultipleConsumers.ReadType.Normal);
- return null;
- }).when(cursorMock).asyncReadEntriesOrWait(
- anyInt(), anyLong(), any(PersistentStickyKeyDispatcherMultipleConsumers.class),
- eq(PersistentStickyKeyDispatcherMultipleConsumers.ReadType.Normal), any());
} catch (Exception e) {
fail("Failed to set to field", e);
}
- // Create 2Consumers
- try {
- doReturn("consumer2").when(slowConsumerMock).consumerName();
- doReturn(true).when(slowConsumerMock).isWritable();
- doReturn(slowChannelMock).when(slowConsumerMock).sendMessages(
- anyList(),
- any(EntryBatchSizes.class),
- any(EntryBatchIndexesAcks.class),
- anyInt(),
- anyLong(),
- anyLong(),
- any(RedeliveryTracker.class)
- );
+ // Mock Cursor#asyncReplayEntries
+ doAnswer(invocationOnMock -> {
+ Set<Position> positionsArg = invocationOnMock.getArgument(0);
+ Set<Position> positions = new TreeSet<>(positionsArg);
+ List<Entry> entries = allEntries.stream()
+ .filter(entry -> entry.getLedgerId() != -1 && positions.contains(entry.getPosition()))
+ .toList();
+ AsyncCallbacks.ReadEntriesCallback callback = invocationOnMock.getArgument(1);
+ Object ctx = invocationOnMock.getArgument(2);
+ callback.readEntriesComplete(copyEntries(entries), ctx);
+ return Collections.emptySet();
+ }).when(cursorMock).asyncReplayEntries(anySet(), any(), any(), anyBoolean());
- persistentDispatcher.addConsumer(consumerMock);
- persistentDispatcher.addConsumer(slowConsumerMock);
- } catch (Exception e) {
- fail("Failed to add mock consumer", e);
- }
+ doAnswer(invocationOnMock -> {
+ int maxEntries = invocationOnMock.getArgument(0);
+ AsyncCallbacks.ReadEntriesCallback callback = invocationOnMock.getArgument(2);
+ List<Entry> entries = allEntries.stream()
+ .filter(entry -> entry.getLedgerId() != -1 && !alreadySent.contains(entry.getPosition()))
+ .limit(maxEntries)
+ .toList();
+ Object ctx = invocationOnMock.getArgument(3);
+ callback.readEntriesComplete(copyEntries(entries), ctx);
+ return null;
+ }).when(cursorMock).asyncReadEntriesWithSkipOrWait(anyInt(), anyLong(), any(), any(), any(), any());
+
+ doReturn(true).when(slowConsumerMock).isWritable();
+ CountDownLatch message3Sent = new CountDownLatch(1);
+ mockSendMessages(consumerMock, entries -> {
+ entries.forEach(entry -> {
+ alreadySent.add(entry.getPosition());
+ });
+ boolean message3Found = entries.stream()
+ .anyMatch(entry -> entry.getLedgerId() == 1 && entry.getEntryId() == 3);
+ if (message3Found) {
+ message3Sent.countDown();
+ }
+ });
+ CountDownLatch slowConsumerMessagesSent = new CountDownLatch(2);
+ mockSendMessages(slowConsumerMock, entries -> {
+ entries.forEach(entry -> {
+ alreadySent.add(entry.getPosition());
+ slowConsumerMessagesSent.countDown();
+ });
+ });
- // run PersistentStickyKeyDispatcherMultipleConsumers#sendMessagesToConsumers
- // run readMoreEntries internally (and skip internally)
- // Change slowConsumer availablePermits to 1
- // run PersistentStickyKeyDispatcherMultipleConsumers#sendMessagesToConsumers internally
- // and then stop to dispatch to slowConsumer
- persistentDispatcher.readEntriesComplete(redeliverEntries,
- PersistentDispatcherMultipleConsumers.ReadType.Replay);
- verify(consumerMock, times(1)).sendMessages(
- argThat(arg -> {
- assertEquals(arg.size(), 1);
- Entry entry = arg.get(0);
- assertEquals(entry.getLedgerId(), 1);
- assertEquals(entry.getEntryId(), 1);
- return true;
- }),
- any(EntryBatchSizes.class),
- any(EntryBatchIndexesAcks.class),
- anyInt(),
- anyLong(),
- anyLong(),
- any(RedeliveryTracker.class)
- );
- verify(slowConsumerMock, times(0)).sendMessages(
- anyList(),
- any(EntryBatchSizes.class),
- any(EntryBatchIndexesAcks.class),
- anyInt(),
- anyLong(),
- anyLong(),
- any(RedeliveryTracker.class)
- );
+ // add entries to redeliver
+ redeliverEntries.forEach(entry -> {
+ // add extra retain since addEntryToReplay will release it
+ ((EntryImpl) entry).retain();
+ persistentDispatcher.addEntryToReplay(entry);
+ });
+
+ // trigger readMoreEntries, will handle redelivery logic and skip slow consumer
+ persistentDispatcher.readMoreEntriesAsync();
+
+ assertTrue(message3Sent.await(5, TimeUnit.SECONDS));
+
+ // verify that slow consumer messages are not sent before message3 to "consumer"
+ assertEquals(slowConsumerMessagesSent.getCount(), 2);
+
+ // set permits to 2
+ slowConsumerAvailablePermits.set(2);
+
+ // now wait for slow consumer messages since there are permits
+ assertTrue(slowConsumerMessagesSent.await(5, TimeUnit.SECONDS));
+
+ allEntries.forEach(Entry::release);
}
@Test(timeOut = 30000)
public void testMessageRedelivery() throws Exception {
- final Queue<Position> actualEntriesToConsumer1 = new ConcurrentLinkedQueue<>();
- final Queue<Position> actualEntriesToConsumer2 = new ConcurrentLinkedQueue<>();
+ final List<Position> actualEntriesToConsumer1 = new CopyOnWriteArrayList<>();
+ final List<Position> actualEntriesToConsumer2 = new CopyOnWriteArrayList<>();
- final Queue<Position> expectedEntriesToConsumer1 = new ConcurrentLinkedQueue<>();
- final Queue<Position> expectedEntriesToConsumer2 = new ConcurrentLinkedQueue<>();
+ final List<Position> expectedEntriesToConsumer1 = new CopyOnWriteArrayList<>();
+ final List<Position> expectedEntriesToConsumer2 = new CopyOnWriteArrayList<>();
- final AtomicInteger remainingEntriesNum = new AtomicInteger(0);
+ final CountDownLatch remainingEntriesNum = new CountDownLatch(3);
final Consumer consumer1 = createMockConsumer();
doReturn("consumer1").when(consumer1).consumerName();
// Change availablePermits of consumer1 to 0 and then back to normal
when(consumer1.getAvailablePermits()).thenReturn(0).thenReturn(10);
doReturn(true).when(consumer1).isWritable();
- doAnswer(invocationOnMock -> {
- List<Entry> entries = invocationOnMock.getArgument(0);
+ mockSendMessages(consumer1, entries -> {
for (Entry entry : entries) {
- remainingEntriesNum.decrementAndGet();
actualEntriesToConsumer1.add(entry.getPosition());
+ remainingEntriesNum.countDown();
}
- return channelMock;
- }).when(consumer1).sendMessages(anyList(), any(EntryBatchSizes.class), any(EntryBatchIndexesAcks.class),
- anyInt(), anyLong(), anyLong(), any(RedeliveryTracker.class));
+ });
final Consumer consumer2 = createMockConsumer();
doReturn("consumer2").when(consumer2).consumerName();
when(consumer2.getAvailablePermits()).thenReturn(10);
doReturn(true).when(consumer2).isWritable();
- doAnswer(invocationOnMock -> {
- List<Entry> entries = invocationOnMock.getArgument(0);
+ mockSendMessages(consumer2, entries -> {
for (Entry entry : entries) {
- remainingEntriesNum.decrementAndGet();
actualEntriesToConsumer2.add(entry.getPosition());
+ remainingEntriesNum.countDown();
}
- return channelMock;
- }).when(consumer2).sendMessages(anyList(), any(EntryBatchSizes.class), any(EntryBatchIndexesAcks.class),
- anyInt(), anyLong(), anyLong(), any(RedeliveryTracker.class));
+ });
- persistentDispatcher.addConsumer(consumer1);
- persistentDispatcher.addConsumer(consumer2);
+ persistentDispatcher.addConsumer(consumer1).join();
+ persistentDispatcher.addConsumer(consumer2).join();
final Field totalAvailablePermitsField = PersistentDispatcherMultipleConsumers.class
.getDeclaredField("totalAvailablePermits");
@@ -458,21 +507,13 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
// Messages with key1 are routed to consumer1 and messages with key2 are routed to consumer2
final List<Entry> allEntries = new ArrayList<>();
- allEntries.add(EntryAndMetadata.create(EntryImpl.create(1, 1, createMessage("message1", 1, keyForConsumer1))));
- allEntries.add(EntryAndMetadata.create(EntryImpl.create(1, 2, createMessage("message2", 2, keyForConsumer1))));
- allEntries.add(EntryAndMetadata.create(EntryImpl.create(1, 3, createMessage("message3", 3, keyForConsumer2))));
- allEntries.forEach(entry -> {
- EntryImpl entryImpl = (EntryImpl) ((EntryAndMetadata) entry).unwrap();
- entryImpl.retain();
- // initialize sticky key hash
- persistentDispatcher.getStickyKeyHash(entry);
- });
- remainingEntriesNum.set(allEntries.size());
+ allEntries.add(createEntry(1, 1, "message1", 1, keyForConsumer1));
+ allEntries.add(createEntry(1, 2, "message2", 2, keyForConsumer1));
+ allEntries.add(createEntry(1, 3, "message3", 3, keyForConsumer2));
+ // add first entry to redeliver initially
final List<Entry> redeliverEntries = new ArrayList<>();
redeliverEntries.add(allEntries.get(0)); // message1
- final List<Entry> readEntries = new ArrayList<>();
- readEntries.add(allEntries.get(2)); // message3
expectedEntriesToConsumer1.add(allEntries.get(0).getPosition());
expectedEntriesToConsumer1.add(allEntries.get(1).getPosition());
@@ -485,14 +526,15 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
Set<Position> alreadyReceived = new TreeSet<>();
alreadyReceived.addAll(actualEntriesToConsumer1);
alreadyReceived.addAll(actualEntriesToConsumer2);
- List<Entry> entries = allEntries.stream().filter(entry -> positions.contains(entry.getPosition())
+ List<Entry> entries = allEntries.stream().filter(entry -> entry.getLedgerId() != -1
+ && positions.contains(entry.getPosition())
&& !alreadyReceived.contains(entry.getPosition()))
.collect(Collectors.toList());
- PersistentStickyKeyDispatcherMultipleConsumers dispatcher = invocationOnMock.getArgument(1);
- dispatcher.readEntriesComplete(entries, PersistentStickyKeyDispatcherMultipleConsumers.ReadType.Replay);
+ AsyncCallbacks.ReadEntriesCallback callback = invocationOnMock.getArgument(1);
+ Object ctx = invocationOnMock.getArgument(2);
+ callback.readEntriesComplete(copyEntries(entries), ctx);
return alreadyReceived;
- }).when(cursorMock).asyncReplayEntries(anySet(), any(PersistentStickyKeyDispatcherMultipleConsumers.class),
- eq(PersistentStickyKeyDispatcherMultipleConsumers.ReadType.Replay), anyBoolean());
+ }).when(cursorMock).asyncReplayEntries(anySet(), any(), any(), anyBoolean());
// Mock Cursor#asyncReadEntriesOrWait
doAnswer(invocationOnMock -> {
@@ -501,32 +543,26 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
alreadyReceived.addAll(actualEntriesToConsumer1);
alreadyReceived.addAll(actualEntriesToConsumer2);
List<Entry> entries = allEntries.stream()
- .filter(entry -> !alreadyReceived.contains(entry.getPosition()))
- .limit(maxEntries).collect(Collectors.toList());
- PersistentStickyKeyDispatcherMultipleConsumers dispatcher = invocationOnMock.getArgument(2);
- dispatcher.readEntriesComplete(entries, PersistentStickyKeyDispatcherMultipleConsumers.ReadType.Normal);
+ .filter(entry -> entry.getLedgerId() != -1 && !alreadyReceived.contains(entry.getPosition()))
+ .limit(maxEntries)
+ .collect(Collectors.toList());
+ AsyncCallbacks.ReadEntriesCallback callback = invocationOnMock.getArgument(2);
+ Object ctx = invocationOnMock.getArgument(3);
+ callback.readEntriesComplete(copyEntries(entries), ctx);
return null;
- }).when(cursorMock).asyncReadEntriesWithSkipOrWait(anyInt(), anyLong(),
- any(PersistentStickyKeyDispatcherMultipleConsumers.class),
- eq(PersistentStickyKeyDispatcherMultipleConsumers.ReadType.Normal), any(), any());
-
- // (1) Run sendMessagesToConsumers
- // (2) Attempts to send message1 to consumer1 but skipped because availablePermits is 0
- // (3) Change availablePermits of consumer1 to 10
- // (4) Run readMoreEntries internally
- // (5) Run sendMessagesToConsumers internally
- // (6) Attempts to send message3 to consumer2 but skipped because redeliveryMessages contains message2
+ }).when(cursorMock).asyncReadEntriesWithSkipOrWait(anyInt(), anyLong(), any(), any(), any(), any());
+
+ // add entries to redeliver
redeliverEntries.forEach(entry -> {
- EntryImpl entryImpl = (EntryImpl) ((EntryAndMetadata) entry).unwrap();
- entryImpl.retain();
+ // add extra retain since addEntryToReplay will release it
+ ((EntryImpl) entry).retain();
persistentDispatcher.addEntryToReplay(entry);
});
- persistentDispatcher.sendMessagesToConsumers(PersistentStickyKeyDispatcherMultipleConsumers.ReadType.Replay,
- redeliverEntries, true);
- while (remainingEntriesNum.get() > 0) {
- // (7) Run readMoreEntries and resend message1 to consumer1 and message2-3 to consumer2
- persistentDispatcher.readMoreEntries();
- }
+
+ // trigger logic to read entries, includes redelivery logic
+ persistentDispatcher.readMoreEntries();
+
+ assertTrue(remainingEntriesNum.await(5, TimeUnit.SECONDS));
assertThat(actualEntriesToConsumer1).containsExactlyElementsOf(expectedEntriesToConsumer1);
assertThat(actualEntriesToConsumer2).containsExactlyElementsOf(expectedEntriesToConsumer2);
@@ -580,10 +616,10 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
// add a consumer without permits to trigger the retry behavior
consumerMockAvailablePermits.set(0);
- dispatcher.addConsumer(consumerMock);
+ dispatcher.addConsumer(consumerMock).join();
// call "readEntriesComplete" directly to test the retry behavior
- List<Entry> entries = List.of(EntryImpl.create(1, 1, createMessage("message1", 1)));
+ List<Entry> entries = List.of(createEntry(1, 1, "message1", 1));
dispatcher.readEntriesComplete(new ArrayList<>(entries), PersistentDispatcherMultipleConsumers.ReadType.Normal);
Awaitility.await().untilAsserted(() -> {
assertEquals(retryDelays.size(), 1);
@@ -591,7 +627,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
}
);
// test the second retry delay
- entries = List.of(EntryImpl.create(1, 1, createMessage("message1", 1)));
+ entries = List.of(createEntry(1, 1, "message1", 1));
dispatcher.readEntriesComplete(new ArrayList<>(entries), PersistentDispatcherMultipleConsumers.ReadType.Normal);
Awaitility.await().untilAsserted(() -> {
assertEquals(retryDelays.size(), 2);
@@ -601,7 +637,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
);
// verify the max retry delay
for (int i = 0; i < 100; i++) {
- entries = List.of(EntryImpl.create(1, 1, createMessage("message1", 1)));
+ entries = List.of(createEntry(1, 1, "message1", 1));
dispatcher.readEntriesComplete(new ArrayList<>(entries), PersistentDispatcherMultipleConsumers.ReadType.Normal);
}
Awaitility.await().untilAsserted(() -> {
@@ -612,14 +648,14 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
);
// unblock to check that the retry delay is reset
consumerMockAvailablePermits.set(1000);
- entries = List.of(EntryImpl.create(1, 2, createMessage("message2", 1, "key2")));
+ entries = List.of(createEntry(1, 2, "message2", 1, "key2"));
dispatcher.readEntriesComplete(new ArrayList<>(entries), PersistentDispatcherMultipleConsumers.ReadType.Normal);
// wait that the possibly async handling has completed
Awaitility.await().untilAsserted(() -> assertFalse(dispatcher.isSendInProgress()));
// now block again to check the next retry delay so verify it was reset
consumerMockAvailablePermits.set(0);
- entries = List.of(EntryImpl.create(1, 3, createMessage("message3", 1, "key3")));
+ entries = List.of(createEntry(1, 3, "message3", 1, "key3"));
dispatcher.readEntriesComplete(new ArrayList<>(entries), PersistentDispatcherMultipleConsumers.ReadType.Normal);
Awaitility.await().untilAsserted(() -> {
assertEquals(retryDelays.size(), 103);
@@ -663,10 +699,10 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
// add a consumer without permits to trigger the retry behavior
consumerMockAvailablePermits.set(0);
- dispatcher.addConsumer(consumerMock);
+ dispatcher.addConsumer(consumerMock).join();
// call "readEntriesComplete" directly to test the retry behavior
- List<Entry> entries = List.of(EntryImpl.create(1, 1, createMessage("message1", 1)));
+ List<Entry> entries = List.of(createEntry(1, 1, "message1", 1));
dispatcher.readEntriesComplete(new ArrayList<>(entries), PersistentDispatcherMultipleConsumers.ReadType.Normal);
Awaitility.await().untilAsserted(() -> {
assertEquals(retryDelays.size(), 1);
@@ -674,7 +710,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
}
);
// test the second retry delay
- entries = List.of(EntryImpl.create(1, 1, createMessage("message1", 1)));
+ entries = List.of(createEntry(1, 1, "message1", 1));
dispatcher.readEntriesComplete(new ArrayList<>(entries), PersistentDispatcherMultipleConsumers.ReadType.Normal);
Awaitility.await().untilAsserted(() -> {
assertEquals(retryDelays.size(), 2);
@@ -684,7 +720,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
);
// verify the max retry delay
for (int i = 0; i < 100; i++) {
- entries = List.of(EntryImpl.create(1, 1, createMessage("message1", 1)));
+ entries = List.of(createEntry(1, 1, "message1", 1));
dispatcher.readEntriesComplete(new ArrayList<>(entries), PersistentDispatcherMultipleConsumers.ReadType.Normal);
}
Awaitility.await().untilAsserted(() -> {
@@ -695,14 +731,14 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
);
// unblock to check that the retry delay is reset
consumerMockAvailablePermits.set(1000);
- entries = List.of(EntryImpl.create(1, 2, createMessage("message2", 1, "key2")));
+ entries = List.of(createEntry(1, 2, "message2", 1, "key2"));
dispatcher.readEntriesComplete(new ArrayList<>(entries), PersistentDispatcherMultipleConsumers.ReadType.Normal);
// wait that the possibly async handling has completed
Awaitility.await().untilAsserted(() -> assertFalse(dispatcher.isSendInProgress()));
// now block again to check the next retry delay so verify it was reset
consumerMockAvailablePermits.set(0);
- entries = List.of(EntryImpl.create(1, 3, createMessage("message3", 1, "key3")));
+ entries = List.of(createEntry(1, 3, "message3", 1, "key3"));
dispatcher.readEntriesComplete(new ArrayList<>(entries), PersistentDispatcherMultipleConsumers.ReadType.Normal);
Awaitility.await().untilAsserted(() -> {
assertEquals(retryDelays.size(), 103);
@@ -771,19 +807,11 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
};
}
- doAnswer(invocationOnMock -> {
- GenericFutureListener<Future<Void>> listener = invocationOnMock.getArgument(0);
- Future<Void> future = mock(Future.class);
- when(future.isDone()).thenReturn(true);
- listener.operationComplete(future);
- return channelMock;
- }).when(channelMock).addListener(any());
-
// add a consumer with permits
consumerMockAvailablePermits.set(1000);
- dispatcher.addConsumer(consumerMock);
+ dispatcher.addConsumer(consumerMock).join();
- List<Entry> entries = new ArrayList<>(List.of(EntryImpl.create(1, 1, createMessage("message1", 1))));
+ List<Entry> entries = new ArrayList<>(List.of(createEntry(1, 1, "message1", 1)));
dispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
Awaitility.await().untilAsserted(() -> {
assertEquals(reScheduleReadInMsCalled.get(), 0, "reScheduleReadInMs should not be called");
@@ -791,17 +819,27 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
});
}
- private ByteBuf createMessage(String message, int sequenceId) {
- return createMessage(message, sequenceId, "testKey");
+ private EntryImpl createEntry(long ledgerId, long entryId, String message, long sequenceId) {
+ return createEntry(ledgerId, entryId, message, sequenceId, "testKey");
+ }
+
+ private EntryImpl createEntry(long ledgerId, long entryId, String message, long sequenceId, String key) {
+ ByteBuf data = createMessage(message, sequenceId, key);
+ EntryImpl entry = EntryImpl.create(ledgerId, entryId, data);
+ data.release();
+ return entry;
}
- private ByteBuf createMessage(String message, int sequenceId, String key) {
+ private ByteBuf createMessage(String message, long sequenceId, String key) {
MessageMetadata messageMetadata = new MessageMetadata()
.setSequenceId(sequenceId)
.setProducerName("testProducer")
.setPartitionKey(key)
.setPartitionKeyB64Encoded(false)
.setPublishTime(System.currentTimeMillis());
- return serializeMetadataAndPayload(Commands.ChecksumType.Crc32c, messageMetadata, Unpooled.copiedBuffer(message.getBytes(UTF_8)));
+ ByteBuf payload = Unpooled.copiedBuffer(message.getBytes(UTF_8));
+ ByteBuf byteBuf = serializeMetadataAndPayload(Commands.ChecksumType.Crc32c, messageMetadata, payload);
+ payload.release();
+ return byteBuf;
}
}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java
index a62d9e74798..a2b70a92c2f 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.impl;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
@@ -37,6 +38,7 @@ import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.MessageIdAdv;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
@@ -44,8 +46,8 @@ import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
import org.apache.pulsar.client.util.TimedCompletableFuture;
import org.apache.pulsar.common.api.proto.CommandAck.AckType;
-import org.apache.pulsar.common.util.collections.ConcurrentBitSetRecyclable;
import org.apache.pulsar.common.api.proto.ProtocolVersion;
+import org.apache.pulsar.common.util.collections.ConcurrentBitSetRecyclable;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
@@ -56,6 +58,7 @@ public class AcknowledgementsGroupingTrackerTest {
private ClientCnx cnx;
private ConsumerImpl<?> consumer;
private EventLoopGroup eventLoopGroup;
+ private AtomicBoolean returnCnx = new AtomicBoolean(true);
@BeforeClass
public void setup() throws NoSuchFieldException, IllegalAccessException {
@@ -67,12 +70,12 @@ public class AcknowledgementsGroupingTrackerTest {
ConnectionPool connectionPool = mock(ConnectionPool.class);
when(client.getCnxPool()).thenReturn(connectionPool);
doReturn(client).when(consumer).getClient();
- doReturn(cnx).when(consumer).getClientCnx();
doReturn(new ConsumerStatsRecorderImpl()).when(consumer).getStats();
doReturn(UnAckedMessageTracker.UNACKED_MESSAGE_TRACKER_DISABLED)
.when(consumer).getUnAckedMessageTracker();
- ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
- when(cnx.ctx()).thenReturn(ctx);
+ ChannelHandlerContext ctx = ClientTestFixtures.mockChannelHandlerContext();
+ doAnswer(invocation -> returnCnx.get() ? cnx : null).when(consumer).getClientCnx();
+ doReturn(ctx).when(cnx).ctx();
}
@DataProvider(name = "isNeedReceipt")
@@ -130,8 +133,6 @@ public class AcknowledgementsGroupingTrackerTest {
tracker.addAcknowledgment(msg6, AckType.Individual, Collections.emptyMap());
assertTrue(tracker.isDuplicate(msg6));
- when(consumer.getClientCnx()).thenReturn(cnx);
-
tracker.flush();
assertTrue(tracker.isDuplicate(msg1));
@@ -190,8 +191,6 @@ public class AcknowledgementsGroupingTrackerTest {
tracker.addListAcknowledgment(Collections.singletonList(msg6), AckType.Individual, Collections.emptyMap());
assertTrue(tracker.isDuplicate(msg6));
- when(consumer.getClientCnx()).thenReturn(cnx);
-
tracker.flush();
assertTrue(tracker.isDuplicate(msg1));
@@ -218,12 +217,13 @@ public class AcknowledgementsGroupingTrackerTest {
assertFalse(tracker.isDuplicate(msg1));
- when(consumer.getClientCnx()).thenReturn(null);
-
- tracker.addAcknowledgment(msg1, AckType.Individual, Collections.emptyMap());
- assertFalse(tracker.isDuplicate(msg1));
-
- when(consumer.getClientCnx()).thenReturn(cnx);
+ returnCnx.set(false);
+ try {
+ tracker.addAcknowledgment(msg1, AckType.Individual, Collections.emptyMap());
+ assertFalse(tracker.isDuplicate(msg1));
+ } finally {
+ returnCnx.set(true);
+ }
tracker.flush();
assertFalse(tracker.isDuplicate(msg1));
@@ -247,12 +247,13 @@ public class AcknowledgementsGroupingTrackerTest {
assertFalse(tracker.isDuplicate(msg1));
- when(consumer.getClientCnx()).thenReturn(null);
-
- tracker.addListAcknowledgment(Collections.singletonList(msg1), AckType.Individual, Collections.emptyMap());
- assertTrue(tracker.isDuplicate(msg1));
-
- when(consumer.getClientCnx()).thenReturn(cnx);
+ returnCnx.set(false);
+ try {
+ tracker.addListAcknowledgment(Collections.singletonList(msg1), AckType.Individual, Collections.emptyMap());
+ assertTrue(tracker.isDuplicate(msg1));
+ } finally {
+ returnCnx.set(true);
+ }
tracker.flush();
assertFalse(tracker.isDuplicate(msg1));
@@ -312,8 +313,6 @@ public class AcknowledgementsGroupingTrackerTest {
tracker.addAcknowledgment(msg6, AckType.Individual, Collections.emptyMap());
assertTrue(tracker.isDuplicate(msg6));
- when(consumer.getClientCnx()).thenReturn(cnx);
-
tracker.flush();
assertTrue(tracker.isDuplicate(msg1));
@@ -374,8 +373,6 @@ public class AcknowledgementsGroupingTrackerTest {
tracker.addListAcknowledgment(Collections.singletonList(msg6), AckType.Individual, Collections.emptyMap());
assertTrue(tracker.isDuplicate(msg6));
- when(consumer.getClientCnx()).thenReturn(cnx);
-
tracker.flush();
assertTrue(tracker.isDuplicate(msg1));
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BinaryProtoLookupServiceTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BinaryProtoLookupServiceTest.java
index 11e00eefcfd..bd4b21b1ea2 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BinaryProtoLookupServiceTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BinaryProtoLookupServiceTest.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.client.impl;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.anyLong;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
@@ -36,6 +37,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.pulsar.client.api.PulsarClientException.LookupException;
import org.apache.pulsar.client.impl.BinaryProtoLookupService.LookupDataResult;
@@ -70,8 +72,17 @@ public class BinaryProtoLookupServiceTest {
CompletableFuture<LookupDataResult> lookupFuture2 = CompletableFuture.completedFuture(lookupResult2);
ClientCnx clientCnx = mock(ClientCnx.class);
- when(clientCnx.newLookup(any(ByteBuf.class), anyLong())).thenReturn(lookupFuture1, lookupFuture1,
- lookupFuture2);
+ AtomicInteger lookupInvocationCounter = new AtomicInteger();
+ doAnswer(invocation -> {
+ ByteBuf byteBuf = invocation.getArgument(0);
+ byteBuf.release();
+ int lookupInvocationCount = lookupInvocationCounter.incrementAndGet();
+ if (lookupInvocationCount < 3) {
+ return lookupFuture1;
+ } else {
+ return lookupFuture2;
+ }
+ }).when(clientCnx).newLookup(any(ByteBuf.class), anyLong());
CompletableFuture<ClientCnx> connectionFuture = CompletableFuture.completedFuture(clientCnx);
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java
index bc1d940c76b..e7986ec57a2 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java
@@ -27,8 +27,6 @@ import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import io.netty.buffer.ByteBuf;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.EventLoopGroup;
import io.netty.util.concurrent.DefaultThreadFactory;
@@ -65,14 +63,7 @@ public class ClientCnxTest {
conf.setOperationTimeoutMs(10);
conf.setKeepAliveIntervalSeconds(0);
ClientCnx cnx = new ClientCnx(InstrumentProvider.NOOP, conf, eventLoop);
-
- ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
- Channel channel = mock(Channel.class);
- when(ctx.channel()).thenReturn(channel);
- ChannelFuture listenerFuture = mock(ChannelFuture.class);
- when(listenerFuture.addListener(any())).thenReturn(listenerFuture);
- when(ctx.writeAndFlush(any())).thenReturn(listenerFuture);
-
+ ChannelHandlerContext ctx = ClientTestFixtures.mockChannelHandlerContext();
cnx.channelActive(ctx);
try {
@@ -91,13 +82,7 @@ public class ClientCnxTest {
conf.setOperationTimeoutMs(10_000);
conf.setKeepAliveIntervalSeconds(0);
ClientCnx cnx = new ClientCnx(InstrumentProvider.NOOP, conf, eventLoop);
-
- ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
- Channel channel = mock(Channel.class);
- when(ctx.channel()).thenReturn(channel);
- ChannelFuture listenerFuture = mock(ChannelFuture.class);
- when(listenerFuture.addListener(any())).thenReturn(listenerFuture);
- when(ctx.writeAndFlush(any())).thenReturn(listenerFuture);
+ ChannelHandlerContext ctx = ClientTestFixtures.mockChannelHandlerContext();
cnx.channelActive(ctx);
CountDownLatch countDownLatch = new CountDownLatch(1);
CompletableFuture<Exception> completableFuture = new CompletableFuture<>();
@@ -129,13 +114,7 @@ public class ClientCnxTest {
conf.setOperationTimeoutMs(10_000);
conf.setKeepAliveIntervalSeconds(0);
ClientCnx cnx = new ClientCnx(InstrumentProvider.NOOP, conf, eventLoop);
-
- ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
- Channel channel = mock(Channel.class);
- when(ctx.channel()).thenReturn(channel);
- ChannelFuture listenerFuture = mock(ChannelFuture.class);
- when(listenerFuture.addListener(any())).thenReturn(listenerFuture);
- when(ctx.writeAndFlush(any())).thenReturn(listenerFuture);
+ ChannelHandlerContext ctx = ClientTestFixtures.mockChannelHandlerContext();
cnx.channelActive(ctx);
cnx.state = ClientCnx.State.Ready;
CountDownLatch countDownLatch = new CountDownLatch(1);
@@ -172,13 +151,7 @@ public class ClientCnxTest {
conf.setOperationTimeoutMs(10_000);
conf.setKeepAliveIntervalSeconds(0);
ClientCnx cnx = new ClientCnx(InstrumentProvider.NOOP, conf, eventLoop);
-
- ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
- Channel channel = mock(Channel.class);
- when(ctx.channel()).thenReturn(channel);
- ChannelFuture listenerFuture = mock(ChannelFuture.class);
- when(listenerFuture.addListener(any())).thenReturn(listenerFuture);
- when(ctx.writeAndFlush(any())).thenReturn(listenerFuture);
+ ChannelHandlerContext ctx = ClientTestFixtures.mockChannelHandlerContext();
cnx.channelActive(ctx);
for (int i = 0; i < 5001; i++) {
cnx.newLookup(null, i);
@@ -199,9 +172,7 @@ public class ClientCnxTest {
conf.setOperationTimeoutMs(10);
ClientCnx cnx = new ClientCnx(InstrumentProvider.NOOP, conf, eventLoop);
- ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
- Channel channel = mock(Channel.class);
- when(ctx.channel()).thenReturn(channel);
+ ChannelHandlerContext ctx = ClientTestFixtures.mockChannelHandlerContext();
Field ctxField = PulsarHandler.class.getDeclaredField("ctx");
ctxField.setAccessible(true);
@@ -233,9 +204,7 @@ public class ClientCnxTest {
ClientConfigurationData conf = new ClientConfigurationData();
ClientCnx cnx = new ClientCnx(InstrumentProvider.NOOP, conf, eventLoop);
- ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
- Channel channel = mock(Channel.class);
- when(ctx.channel()).thenReturn(channel);
+ ChannelHandlerContext ctx = ClientTestFixtures.mockChannelHandlerContext();
Field ctxField = PulsarHandler.class.getDeclaredField("ctx");
ctxField.setAccessible(true);
@@ -248,10 +217,6 @@ public class ClientCnxTest {
cnxField.setAccessible(true);
cnxField.set(cnx, ClientCnx.State.SentConnectFrame);
- ChannelFuture listenerFuture = mock(ChannelFuture.class);
- when(listenerFuture.addListener(any())).thenReturn(listenerFuture);
- when(ctx.writeAndFlush(any())).thenReturn(listenerFuture);
-
ByteBuf getLastIdCmd = Commands.newGetLastMessageId(5, requestId);
CompletableFuture<?> future = cnx.sendGetLastMessageId(getLastIdCmd, requestId);
@@ -396,13 +361,7 @@ public class ClientCnxTest {
ClientConfigurationData conf = new ClientConfigurationData();
ClientCnx cnx = new ClientCnx(InstrumentProvider.NOOP, conf, eventLoop);
- ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
- Channel channel = mock(Channel.class);
- when(ctx.channel()).thenReturn(channel);
-
- ChannelFuture listenerFuture = mock(ChannelFuture.class);
- when(listenerFuture.addListener(any())).thenReturn(listenerFuture);
- when(ctx.writeAndFlush(any())).thenReturn(listenerFuture);
+ ChannelHandlerContext ctx = ClientTestFixtures.mockChannelHandlerContext();
Field ctxField = PulsarHandler.class.getDeclaredField("ctx");
ctxField.setAccessible(true);
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientTestFixtures.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientTestFixtures.java
index 915c3dcc05a..e9132fb9f45 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientTestFixtures.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientTestFixtures.java
@@ -22,12 +22,22 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.EventLoop;
+import io.netty.util.ReferenceCountUtil;
import io.netty.util.Timer;
+import io.netty.util.concurrent.EventExecutor;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.GenericFutureListener;
+import io.netty.util.concurrent.SucceededFuture;
import java.net.SocketAddress;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
@@ -75,11 +85,7 @@ class ClientTestFixtures {
}
static PulsarClientImpl mockClientCnx(PulsarClientImpl clientMock) {
- ClientCnx clientCnxMock = mock(ClientCnx.class, Mockito.RETURNS_DEEP_STUBS);
- when(clientCnxMock.ctx()).thenReturn(mock(ChannelHandlerContext.class));
- when(clientCnxMock.sendRequestWithId(any(), anyLong()))
- .thenReturn(CompletableFuture.completedFuture(mock(ProducerResponse.class)));
- when(clientCnxMock.channel().remoteAddress()).thenReturn(mock(SocketAddress.class));
+ ClientCnx clientCnxMock = mockClientCnx();
when(clientMock.getConnection(any())).thenReturn(CompletableFuture.completedFuture(clientCnxMock));
when(clientMock.getConnection(anyString())).thenReturn(CompletableFuture.completedFuture(clientCnxMock));
when(clientMock.getConnection(anyString(), anyInt()))
@@ -94,6 +100,82 @@ class ClientTestFixtures {
return clientMock;
}
+ public static ClientCnx mockClientCnx() {
+ ClientCnx clientCnxMock = mock(ClientCnx.class, Mockito.RETURNS_DEEP_STUBS);
+ ChannelHandlerContext ctx = mockChannelHandlerContext();
+ doReturn(ctx).when(clientCnxMock).ctx();
+ doAnswer(invocation -> {
+ ByteBuf buf = invocation.getArgument(0);
+ buf.release();
+ return CompletableFuture.completedFuture(mock(ProducerResponse.class));
+ }).when(clientCnxMock).sendRequestWithId(any(), anyLong());
+ when(clientCnxMock.channel().remoteAddress()).thenReturn(mock(SocketAddress.class));
+ return clientCnxMock;
+ }
+
+ /**
+ * Mock a ChannelHandlerContext where write and writeAndFlush are always successful.
+ * This might not be suitable for all tests.
+ *
+ * @return a mocked ChannelHandlerContext
+ */
+ public static ChannelHandlerContext mockChannelHandlerContext() {
+ ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
+
+ // return an empty channel mock from ctx.channel()
+ Channel channel = mock(Channel.class);
+ when(ctx.channel()).thenReturn(channel);
+
+ // handle write and writeAndFlush methods so that the input message is released
+
+ // create a listener future that is returned from write and writeAndFlush
+ // that immediately completes the listener in the calling thread as if it was successful
+ ChannelFuture listenerFuture = mock(ChannelFuture.class);
+ Future<Void> succeededFuture = createSucceededFuture();
+ doAnswer(invocation -> {
+ GenericFutureListener<Future<Void>> listener = invocation.getArgument(0);
+ listener.operationComplete(succeededFuture);
+ return listenerFuture;
+ }).when(listenerFuture).addListener(any());
+
+ // handle write and writeAndFlush methods so that the input message is released
+ doAnswer(invocation -> {
+ Object msg = invocation.getArgument(0);
+ ReferenceCountUtil.release(msg);
+ return listenerFuture;
+ }).when(ctx).write(any(), any());
+ doAnswer(invocation -> {
+ Object msg = invocation.getArgument(0);
+ ReferenceCountUtil.release(msg);
+ return listenerFuture;
+ }).when(ctx).writeAndFlush(any(), any());
+ doAnswer(invocation -> {
+ Object msg = invocation.getArgument(0);
+ ReferenceCountUtil.release(msg);
+ return listenerFuture;
+ }).when(ctx).writeAndFlush(any());
+
+ return ctx;
+ }
+
+ public static Future<Void> createSucceededFuture() {
+ EventExecutor eventExecutor = mockEventExecutor();
+ // create a succeeded future that is returned from the listener, listeners will run in the calling thread
+ // using the mocked EventExecutor
+ SucceededFuture<Void> succeededFuture = new SucceededFuture<>(eventExecutor, null);
+ return succeededFuture;
+ }
+
+ public static EventExecutor mockEventExecutor() {
+ // mock an EventExecutor that runs the listener in the calling thread
+ EventExecutor eventExecutor = mock(EventExecutor.class);
+ doAnswer(invocation -> {
+ invocation.getArgument(0, Runnable.class).run();
+ return null;
+ }).when(eventExecutor).execute(any(Runnable.class));
+ return eventExecutor;
+ }
+
static <T> CompletableFuture<T> createDelayedCompletedFuture(T result, int delayMillis) {
CompletableFuture<T> future = new CompletableFuture<>();
SCHEDULER.schedule(() -> future.complete(result), delayMillis, TimeUnit.MILLISECONDS);
(pulsar) 02/03: [fix][io] Fix SyntaxWarning in Pulsar Python functions (#24297)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 1c0c1dc14481522db4afed90287bc36e710fef76
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue May 13 19:33:29 2025 +0300
[fix][io] Fix SyntaxWarning in Pulsar Python functions (#24297)
(cherry picked from commit 272acfb483f3554155a121cafdbb6c4321539f9a)
---
pulsar-functions/instance/src/main/python/contextimpl.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pulsar-functions/instance/src/main/python/contextimpl.py b/pulsar-functions/instance/src/main/python/contextimpl.py
index bfe7b23927b..826ad65b2e0 100755
--- a/pulsar-functions/instance/src/main/python/contextimpl.py
+++ b/pulsar-functions/instance/src/main/python/contextimpl.py
@@ -210,7 +210,7 @@ class ContextImpl(pulsar.Context):
topic_consumer = self.consumers[topic]
else:
# if this topic is a partitioned topic
- m = re.search('(.+)-partition-(\d+)', topic)
+ m = re.search(r'(.+)-partition-(\d+)', topic)
if not m:
raise ValueError('Invalid topicname %s' % topic)
elif m.group(1) in self.consumers: