You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2023/05/24 02:15:00 UTC
[pulsar] branch branch-2.11 updated (5a66d6cc7bb -> 833f08a1f17)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a change to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 5a66d6cc7bb [fix][broker] Fix class name typo `PrecisPublishLimiter` to "Precise" (#20310)
new 0b80bad015d [improve][cli] Allow pulser-client consume create a replicated subscription (#20316)
new fec0fca92ba [fix][broker] fix consume stuck of shared streaming dispatcher (#18315)
new 833f08a1f17 [fix] [broker] In Key_Shared mode: remove unnecessary mechanisms of message skip to avoid unnecessary consumption stuck (#20335)
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:
.../PersistentDispatcherMultipleConsumers.java | 29 +++++++++----
...istentStickyKeyDispatcherMultipleConsumers.java | 37 ++---------------
...istentStreamingDispatcherMultipleConsumers.java | 24 +++++++++--
...ntStickyKeyDispatcherMultipleConsumersTest.java | 47 ++++++++++++----------
.../PersistentTopicStreamingDispatcherE2ETest.java | 6 +++
.../org/apache/pulsar/client/cli/CmdConsume.java | 5 ++-
6 files changed, 80 insertions(+), 68 deletions(-)
[pulsar] 03/03: [fix] [broker] In Key_Shared mode: remove unnecessary mechanisms of message skip to avoid unnecessary consumption stuck (#20335)
Posted by te...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 833f08a1f17849990cd4b43f3826a438274ca96e
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Sat May 20 03:38:24 2023 +0800
[fix] [broker] In Key_Shared mode: remove unnecessary mechanisms of message skip to avoid unnecessary consumption stuck (#20335)
### Motivation
- https://github.com/apache/pulsar/pull/7105 provide a mechanism to avoid a stuck consumer affecting the consumption of other consumers:
- if all consumers can not accept more messages, stop delivering messages to the client.
- if one consumer can not accept more messages, just read new messages and deliver them to other consumers.
- https://github.com/apache/pulsar/pull/7553 provide a mechanism to fix the issue of lost order of consumption: If the consumer cannot accept any more messages, skip the consumer for the next round of message delivery because there may be messages with the same key in the replay queue.
- https://github.com/apache/pulsar/pull/10762 provide a mechanism to fix the issue of lost order of consumption: If there have any messages with the same key in the replay queue, do not deliver the new messages to this consumer.
https://github.com/apache/pulsar/pull/10762 and https://github.com/apache/pulsar/pull/7553 do the same thing and https://github.com/apache/pulsar/pull/10762 is better than https://github.com/apache/pulsar/pull/7553 , so https://github.com/apache/pulsar/pull/7553 is unnecessary.
### Modifications
remove the mechanism provided by https://github.com/apache/pulsar/pull/7553 to avoid unnecessary consumption stuck.
---
...istentStickyKeyDispatcherMultipleConsumers.java | 24 +---------------------
1 file changed, 1 insertion(+), 23 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
index 452c061b7fc..2b711024926 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
@@ -70,17 +70,12 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
*/
private final LinkedHashMap<Consumer, PositionImpl> recentlyJoinedConsumers;
- private final Set<Consumer> stuckConsumers;
- private final Set<Consumer> nextStuckConsumers;
-
PersistentStickyKeyDispatcherMultipleConsumers(PersistentTopic topic, ManagedCursor cursor,
Subscription subscription, ServiceConfiguration conf, KeySharedMeta ksm) {
super(topic, cursor, subscription, ksm.isAllowOutOfOrderDelivery());
this.allowOutOfOrderDelivery = ksm.isAllowOutOfOrderDelivery();
this.recentlyJoinedConsumers = allowOutOfOrderDelivery ? null : new LinkedHashMap<>();
- this.stuckConsumers = new HashSet<>();
- this.nextStuckConsumers = new HashSet<>();
this.keySharedMode = ksm.getKeySharedMode();
switch (this.keySharedMode) {
case AUTO_SPLIT:
@@ -219,8 +214,6 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
}
}
- nextStuckConsumers.clear();
-
final Map<Consumer, List<Entry>> groupedEntries = localGroupedEntries.get();
groupedEntries.clear();
final Map<Consumer, Set<Integer>> consumerStickyKeyHashesMap = new HashMap<>();
@@ -322,14 +315,11 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
}
}
- stuckConsumers.clear();
-
if (totalMessagesSent == 0 && (recentlyJoinedConsumers == null || recentlyJoinedConsumers.isEmpty())) {
// This means, that all the messages we've just read cannot be dispatched right now.
// This condition can only happen when:
// 1. We have consumers ready to accept messages (otherwise the would not haven been triggered)
// 2. All keys in the current set of messages are routing to consumers that are currently busy
- // and stuck is not caused by stuckConsumers
//
// The solution here is to move on and read next batch of messages which might hopefully contain
// also keys meant for other consumers.
@@ -338,10 +328,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
// ahead in the stream while the new consumers are not ready to accept the new messages,
// therefore would be most likely only increase the distance between read-position and mark-delete
// position.
- if (!nextStuckConsumers.isEmpty()) {
- isDispatcherStuckOnReplays = true;
- stuckConsumers.addAll(nextStuckConsumers);
- }
+ isDispatcherStuckOnReplays = true;
return true;
} else if (currentThreadKeyNumber == 0) {
return true;
@@ -352,8 +339,6 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
private int getRestrictedMaxEntriesForConsumer(Consumer consumer, List<Entry> entries, int maxMessages,
ReadType readType, Set<Integer> stickyKeyHashes) {
if (maxMessages == 0) {
- // the consumer was stuck
- nextStuckConsumers.add(consumer);
return 0;
}
if (readType == ReadType.Normal && stickyKeyHashes != null
@@ -370,13 +355,6 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
// At this point, all the old messages were already consumed and this consumer
// is now ready to receive any message
if (maxReadPosition == null) {
- // stop to dispatch by stuckConsumers
- if (stuckConsumers.contains(consumer)) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] stop to dispatch by stuckConsumers, consumer: {}", name, consumer);
- }
- return 0;
- }
// The consumer has not recently joined, so we can send all messages
return maxMessages;
}
[pulsar] 02/03: [fix][broker] fix consume stuck of shared streaming dispatcher (#18315)
Posted by te...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit fec0fca92ba955b02be74bae9fb05f1df3519d0d
Author: Cong Zhao <zh...@apache.org>
AuthorDate: Fri Nov 18 12:18:34 2022 +0800
[fix][broker] fix consume stuck of shared streaming dispatcher (#18315)
---
.../PersistentDispatcherMultipleConsumers.java | 29 +++++++++----
...istentStickyKeyDispatcherMultipleConsumers.java | 13 +-----
...istentStreamingDispatcherMultipleConsumers.java | 24 +++++++++--
...ntStickyKeyDispatcherMultipleConsumersTest.java | 47 ++++++++++++----------
.../PersistentTopicStreamingDispatcherE2ETest.java | 6 +++
5 files changed, 75 insertions(+), 44 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
index e8bbb42e58d..72f53bce32b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
@@ -94,7 +94,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
protected volatile PositionImpl minReplayedPosition = null;
protected boolean shouldRewindBeforeReadingOrReplaying = false;
protected final String name;
- protected boolean sendInProgress;
+ private boolean sendInProgress = false;
protected static final AtomicIntegerFieldUpdater<PersistentDispatcherMultipleConsumers>
TOTAL_AVAILABLE_PERMITS_UPDATER =
AtomicIntegerFieldUpdater.newUpdater(PersistentDispatcherMultipleConsumers.class,
@@ -257,7 +257,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
}
public synchronized void readMoreEntries() {
- if (sendInProgress) {
+ if (isSendInProgress()) {
// we cannot read more entries while sending the previous batch
// otherwise we could re-read the same entries and send duplicates
return;
@@ -560,9 +560,9 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
if (serviceConfig.isDispatcherDispatchMessagesInSubscriptionThread()) {
// setting sendInProgress here, because sendMessagesToConsumers will be executed
// in a separate thread, and we want to prevent more reads
- sendInProgress = true;
+ acquireSendInProgress();
dispatchMessagesThread.execute(safeRun(() -> {
- if (sendMessagesToConsumers(readType, entries)) {
+ if (sendMessagesToConsumers(readType, entries, false)) {
updatePendingBytesToDispatch(-size);
readMoreEntries();
} else {
@@ -570,7 +570,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
}
}));
} else {
- if (sendMessagesToConsumers(readType, entries)) {
+ if (sendMessagesToConsumers(readType, entries, true)) {
updatePendingBytesToDispatch(-size);
readMoreEntriesAsync();
} else {
@@ -579,12 +579,27 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
}
}
- protected final synchronized boolean sendMessagesToConsumers(ReadType readType, List<Entry> entries) {
+ protected synchronized void acquireSendInProgress() {
sendInProgress = true;
+ }
+
+ protected synchronized void releaseSendInProgress() {
+ sendInProgress = false;
+ }
+
+ protected synchronized boolean isSendInProgress() {
+ return sendInProgress;
+ }
+
+ protected final synchronized boolean sendMessagesToConsumers(ReadType readType, List<Entry> entries,
+ boolean needAcquireSendInProgress) {
+ if (needAcquireSendInProgress) {
+ acquireSendInProgress();
+ }
try {
return trySendMessagesToConsumers(readType, entries);
} finally {
- sendInProgress = false;
+ releaseSendInProgress();
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
index 1e52d985520..452c061b7fc 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
@@ -31,7 +31,6 @@ import java.util.List;
import java.util.Map;
import java.util.NavigableSet;
import java.util.Set;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedCursor;
@@ -343,17 +342,9 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
isDispatcherStuckOnReplays = true;
stuckConsumers.addAll(nextStuckConsumers);
}
- // readMoreEntries should run regardless whether or not stuck is caused by
- // stuckConsumers for avoid stopping dispatch.
- sendInProgress = false;
- topic.getBrokerService().executor().execute(safeRun(this::readMoreEntries));
+ return true;
} else if (currentThreadKeyNumber == 0) {
- sendInProgress = false;
- topic.getBrokerService().executor().schedule(safeRun(() -> {
- synchronized (PersistentStickyKeyDispatcherMultipleConsumers.this) {
- readMoreEntries();
- }
- }), 100, TimeUnit.MILLISECONDS);
+ return true;
}
return false;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStreamingDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStreamingDispatcherMultipleConsumers.java
index 9937402ef35..6ee5641ee34 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStreamingDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStreamingDispatcherMultipleConsumers.java
@@ -45,6 +45,7 @@ import org.apache.pulsar.broker.service.streamingdispatch.StreamingEntryReader;
public class PersistentStreamingDispatcherMultipleConsumers extends PersistentDispatcherMultipleConsumers
implements StreamingDispatcher {
+ private int sendingTaskCounter = 0;
private final StreamingEntryReader streamingEntryReader = new StreamingEntryReader((ManagedCursorImpl) cursor,
this, topic);
@@ -101,16 +102,16 @@ public class PersistentStreamingDispatcherMultipleConsumers extends PersistentDi
if (serviceConfig.isDispatcherDispatchMessagesInSubscriptionThread()) {
// setting sendInProgress here, because sendMessagesToConsumers will be executed
// in a separate thread, and we want to prevent more reads
- sendInProgress = true;
+ acquireSendInProgress();
dispatchMessagesThread.execute(safeRun(() -> {
- if (sendMessagesToConsumers(readType, Lists.newArrayList(entry))) {
+ if (sendMessagesToConsumers(readType, Lists.newArrayList(entry), false)) {
readMoreEntries();
} else {
updatePendingBytesToDispatch(-size);
}
}));
} else {
- if (sendMessagesToConsumers(readType, Lists.newArrayList(entry))) {
+ if (sendMessagesToConsumers(readType, Lists.newArrayList(entry), true)) {
readMoreEntriesAsync();
} else {
updatePendingBytesToDispatch(-size);
@@ -159,9 +160,24 @@ public class PersistentStreamingDispatcherMultipleConsumers extends PersistentDi
}
}
+ @Override
+ protected synchronized void acquireSendInProgress() {
+ sendingTaskCounter++;
+ }
+
+ @Override
+ protected synchronized void releaseSendInProgress() {
+ sendingTaskCounter--;
+ }
+
+ @Override
+ protected synchronized boolean isSendInProgress() {
+ return sendingTaskCounter > 0;
+ }
+
@Override
public synchronized void readMoreEntries() {
- if (sendInProgress) {
+ if (isSendInProgress()) {
// we cannot read more entries while sending the previous batch
// otherwise we could re-read the same entries and send duplicates
return;
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 5ff988fdc46..f0d2cea57d0 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
@@ -40,6 +40,7 @@ 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 java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Collections;
@@ -50,7 +51,6 @@ import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
-import io.netty.channel.EventLoopGroup;
import org.apache.bookkeeper.common.util.OrderedExecutor;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.Position;
@@ -149,8 +149,8 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
subscriptionMock = mock(PersistentSubscription.class);
persistentDispatcher = new PersistentStickyKeyDispatcherMultipleConsumers(
- topicMock, cursorMock, subscriptionMock, configMock,
- new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT));
+ topicMock, cursorMock, subscriptionMock, configMock,
+ new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT));
}
public void cleanup() {
@@ -195,31 +195,31 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
}
Awaitility.await().untilAsserted(() -> {
- ArgumentCaptor<Integer> totalMessagesCaptor = ArgumentCaptor.forClass(Integer.class);
- verify(consumerMock, times(1)).sendMessages(
- anyList(),
- any(EntryBatchSizes.class),
- any(EntryBatchIndexesAcks.class),
- totalMessagesCaptor.capture(),
- anyLong(),
- anyLong(),
- any(RedeliveryTracker.class)
- );
-
- List<Integer> allTotalMessagesCaptor = totalMessagesCaptor.getAllValues();
- Assert.assertEquals(allTotalMessagesCaptor.get(0).intValue(), 5);
- });
+ ArgumentCaptor<Integer> totalMessagesCaptor = ArgumentCaptor.forClass(Integer.class);
+ verify(consumerMock, times(1)).sendMessages(
+ anyList(),
+ any(EntryBatchSizes.class),
+ any(EntryBatchIndexesAcks.class),
+ totalMessagesCaptor.capture(),
+ anyLong(),
+ anyLong(),
+ any(RedeliveryTracker.class)
+ );
+
+ List<Integer> allTotalMessagesCaptor = totalMessagesCaptor.getAllValues();
+ Assert.assertEquals(allTotalMessagesCaptor.get(0).intValue(), 5);
+ });
}
@Test(timeOut = 10000)
public void testSendMessage() {
KeySharedMeta keySharedMeta = new KeySharedMeta().setKeySharedMode(KeySharedMode.STICKY);
PersistentStickyKeyDispatcherMultipleConsumers persistentDispatcher = new PersistentStickyKeyDispatcherMultipleConsumers(
- topicMock, cursorMock, subscriptionMock, configMock, keySharedMeta);
+ topicMock, cursorMock, subscriptionMock, configMock, keySharedMeta);
try {
keySharedMeta.addHashRange()
- .setStart(0)
- .setEnd(9);
+ .setStart(0)
+ .setEnd(9);
Consumer consumerMock = mock(Consumer.class);
doReturn(keySharedMeta).when(consumerMock).getKeySharedMeta();
@@ -297,7 +297,10 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
// Change slowConsumer availablePermits to 1
// run PersistentStickyKeyDispatcherMultipleConsumers#sendMessagesToConsumers internally
// and then stop to dispatch to slowConsumer
- persistentDispatcher.sendMessagesToConsumers(PersistentStickyKeyDispatcherMultipleConsumers.ReadType.Normal, redeliverEntries);
+ if (persistentDispatcher.sendMessagesToConsumers(PersistentStickyKeyDispatcherMultipleConsumers.ReadType.Normal,
+ redeliverEntries, true)) {
+ persistentDispatcher.readMoreEntriesAsync();
+ }
Awaitility.await().untilAsserted(() -> {
verify(consumerMock, times(1)).sendMessages(
@@ -438,7 +441,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
// (5) Run sendMessagesToConsumers internally
// (6) Attempts to send message3 to consumer2 but skipped because redeliveryMessages contains message2
persistentDispatcher.sendMessagesToConsumers(PersistentStickyKeyDispatcherMultipleConsumers.ReadType.Replay,
- redeliverEntries);
+ redeliverEntries, true);
while (remainingEntriesNum.get() > 0) {
// (7) Run readMoreEntries and resend message1 to consumer1 and message2-3 to consumer2
persistentDispatcher.readMoreEntries();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicStreamingDispatcherE2ETest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicStreamingDispatcherE2ETest.java
index a6fe95e7818..7f94a457bb7 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicStreamingDispatcherE2ETest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicStreamingDispatcherE2ETest.java
@@ -33,4 +33,10 @@ public class PersistentTopicStreamingDispatcherE2ETest extends PersistentTopicE2
super.doInitConf();
conf.setStreamingDispatch(true);
}
+
+ @Override
+ @Test
+ public void testMessageRedelivery() throws Exception {
+ super.testMessageRedelivery();
+ }
}
[pulsar] 01/03: [improve][cli] Allow pulser-client consume create a replicated subscription (#20316)
Posted by te...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 0b80bad015d214507d513ba759d4c4458c835b8e
Author: Kim, Joo Hyuk <be...@gmail.com>
AuthorDate: Fri May 19 00:21:37 2023 +0900
[improve][cli] Allow pulser-client consume create a replicated subscription (#20316)
---
.../src/main/java/org/apache/pulsar/client/cli/CmdConsume.java | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdConsume.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdConsume.java
index 5850e557a95..23e31dc5c6c 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdConsume.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdConsume.java
@@ -142,6 +142,8 @@ public class CmdConsume {
private ClientBuilder clientBuilder;
private Authentication authentication;
private String serviceURL;
+ @Parameter(names = {"-rs", "--replicated" }, description = "Whether the subscription status should be replicated")
+ private boolean replicateSubscriptionState = false;
public CmdConsume() {
// Do nothing
@@ -304,7 +306,8 @@ public class CmdConsume {
.subscriptionType(subscriptionType)
.subscriptionMode(subscriptionMode)
.subscriptionInitialPosition(subscriptionInitialPosition)
- .poolMessages(poolMessages);
+ .poolMessages(poolMessages)
+ .replicateSubscriptionState(replicateSubscriptionState);
if (isRegex) {
builder.topicsPattern(Pattern.compile(topic));