You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zi...@apache.org on 2024/12/30 03:11:14 UTC
(pulsar) branch branch-3.3 updated (84a4bc60f81 -> a592077ca0f)
This is an automated email from the ASF dual-hosted git repository.
zixuan pushed a change to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 84a4bc60f81 [fix][ml] Topic load timeout due to ml data ledger future never finishes (#23772)
new 109f04251f8 [improve][client] Make replicateSubscriptionState nullable (#23757)
new a592077ca0f [fix][broker] Fix enableReplicatedSubscriptions (#23781)
The 2 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:
.../apache/pulsar/broker/service/ServerCnx.java | 4 +-
.../pulsar/broker/service/SubscriptionOption.java | 2 +-
.../service/nonpersistent/NonPersistentTopic.java | 4 +-
.../service/persistent/PersistentSubscription.java | 19 +++--
.../broker/service/persistent/PersistentTopic.java | 19 ++---
...nableReplicatedSubscriptionsIsDisabledTest.java | 83 +++++++++++++++++++
.../client/api/ReplicateSubscriptionTest.java | 96 ++++++++++++++++++++++
.../apache/pulsar/client/impl/ConsumerImpl.java | 2 +-
.../impl/conf/ConsumerConfigurationData.java | 14 +++-
.../client/impl/ConsumerBuilderImplTest.java | 36 +++++++-
.../apache/pulsar/common/protocol/Commands.java | 8 +-
11 files changed, 260 insertions(+), 27 deletions(-)
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/client/api/EnableReplicatedSubscriptionsIsDisabledTest.java
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/client/api/ReplicateSubscriptionTest.java
(pulsar) 02/02: [fix][broker] Fix enableReplicatedSubscriptions (#23781)
Posted by zi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zixuan pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a592077ca0f3782b018484d2f62a3d654ff4f9fd
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Fri Dec 27 19:19:03 2024 +0800
[fix][broker] Fix enableReplicatedSubscriptions (#23781)
Signed-off-by: Zixuan Liu <no...@gmail.com>
(cherry picked from commit 6e3eaf5150907ecb12bfec3022c26467c16e4721)
---
.../broker/service/persistent/PersistentTopic.java | 6 +-
...nableReplicatedSubscriptionsIsDisabledTest.java | 83 ++++++++++++++++++++++
2 files changed, 87 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index af3e021a97c..f75cac8a0f3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -954,9 +954,11 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
return brokerService.checkTopicNsOwnership(getName()).thenCompose(__ -> {
- if (replicatedSubscriptionStateArg != null && replicatedSubscriptionStateArg
+ Boolean replicatedSubscriptionState = replicatedSubscriptionStateArg;
+ if (replicatedSubscriptionState != null && replicatedSubscriptionState
&& !brokerService.pulsar().getConfiguration().isEnableReplicatedSubscriptions()) {
log.warn("[{}] Replicated Subscription is disabled by broker.", getName());
+ replicatedSubscriptionState = false;
}
if (subType == SubType.Key_Shared
@@ -1025,7 +1027,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
CompletableFuture<? extends Subscription> subscriptionFuture = isDurable
? getDurableSubscription(subscriptionName, initialPosition, startMessageRollbackDurationSec,
- replicatedSubscriptionStateArg, subscriptionProperties)
+ replicatedSubscriptionState, subscriptionProperties)
: getNonDurableSubscription(subscriptionName, startMessageId, initialPosition,
startMessageRollbackDurationSec, readCompacted, subscriptionProperties);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/EnableReplicatedSubscriptionsIsDisabledTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/EnableReplicatedSubscriptionsIsDisabledTest.java
new file mode 100644
index 00000000000..d002261cee4
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/EnableReplicatedSubscriptionsIsDisabledTest.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import org.apache.pulsar.broker.service.Subscription;
+import org.apache.pulsar.broker.service.Topic;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.common.naming.TopicName;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-api")
+public class EnableReplicatedSubscriptionsIsDisabledTest extends ProducerConsumerBase {
+
+ @BeforeClass
+ @Override
+ protected void setup() throws Exception {
+ super.internalSetup();
+ super.producerBaseSetup();
+ }
+
+ @Override
+ protected void doInitConf() throws Exception {
+ super.doInitConf();
+ conf.setEnableReplicatedSubscriptions(false);
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @Test
+ public void testReplicateSubscriptionStateIsEnabled() throws Exception {
+ String topicName = TopicName.get("my-property/my-ns/testReplicateSubscriptionStateIsEnabled").toString();
+ String subName = "my-subscription";
+ @Cleanup
+ Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
+ .topic(topicName)
+ .subscriptionName(subName)
+ .replicateSubscriptionState(true)
+ .subscribe();
+ CompletableFuture<Optional<Topic>> topicIfExists = pulsar.getBrokerService().getTopicIfExists(topicName);
+ assertThat(topicIfExists)
+ .succeedsWithin(3, TimeUnit.SECONDS)
+ .matches(optionalTopic -> {
+ assertTrue(optionalTopic.isPresent());
+ Topic topicRef = optionalTopic.get();
+ Subscription subscription = topicRef.getSubscription(subName);
+ assertNotNull(subscription);
+ assertTrue(subscription instanceof PersistentSubscription);
+ PersistentSubscription persistentSubscription = (PersistentSubscription) subscription;
+ assertEquals(persistentSubscription.getReplicatedControlled(), Boolean.FALSE);
+ return true;
+ });
+ }
+}
(pulsar) 01/02: [improve][client] Make replicateSubscriptionState nullable (#23757)
Posted by zi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zixuan pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 109f04251f824405e88702eee1053a3663f1b742
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Fri Dec 20 20:58:03 2024 +0800
[improve][client] Make replicateSubscriptionState nullable (#23757)
Signed-off-by: Zixuan Liu <no...@gmail.com>
(cherry picked from commit 3fce3097c76a9c8cb64cf3d8d87f6e050e6cb3a5)
---
.../apache/pulsar/broker/service/ServerCnx.java | 4 +-
.../pulsar/broker/service/SubscriptionOption.java | 2 +-
.../service/nonpersistent/NonPersistentTopic.java | 4 +-
.../service/persistent/PersistentSubscription.java | 19 +++--
.../broker/service/persistent/PersistentTopic.java | 19 ++---
.../client/api/ReplicateSubscriptionTest.java | 96 ++++++++++++++++++++++
.../apache/pulsar/client/impl/ConsumerImpl.java | 2 +-
.../impl/conf/ConsumerConfigurationData.java | 14 +++-
.../client/impl/ConsumerBuilderImplTest.java | 36 +++++++-
.../apache/pulsar/common/protocol/Commands.java | 8 +-
10 files changed, 176 insertions(+), 28 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
index ca91981963b..8be363171df 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -1229,8 +1229,8 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
? subscribe.getStartMessageRollbackDurationSec()
: -1;
final SchemaData schema = subscribe.hasSchema() ? getSchema(subscribe.getSchema()) : null;
- final boolean isReplicated = subscribe.hasReplicateSubscriptionState()
- && subscribe.isReplicateSubscriptionState();
+ final Boolean isReplicated =
+ subscribe.hasReplicateSubscriptionState() ? subscribe.isReplicateSubscriptionState() : null;
final boolean forceTopicCreation = subscribe.isForceTopicCreation();
final KeySharedMeta keySharedMeta = subscribe.hasKeySharedMeta()
? new KeySharedMeta().copyFrom(subscribe.getKeySharedMeta())
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SubscriptionOption.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SubscriptionOption.java
index af56d023616..328e7618f8c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SubscriptionOption.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SubscriptionOption.java
@@ -46,7 +46,7 @@ public class SubscriptionOption {
private boolean readCompacted;
private CommandSubscribe.InitialPosition initialPosition;
private long startMessageRollbackDurationSec;
- private boolean replicatedSubscriptionStateArg;
+ private Boolean replicatedSubscriptionStateArg;
private KeySharedMeta keySharedMeta;
private Optional<Map<String, String>> subscriptionProperties;
private long consumerEpoch;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
index 9d2f3588dc9..95df15956a4 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
@@ -267,7 +267,7 @@ public class NonPersistentTopic extends AbstractTopic implements Topic, TopicPol
return internalSubscribe(option.getCnx(), option.getSubscriptionName(), option.getConsumerId(),
option.getSubType(), option.getPriorityLevel(), option.getConsumerName(),
option.getStartMessageId(), option.getMetadata(), option.isReadCompacted(),
- option.getStartMessageRollbackDurationSec(), option.isReplicatedSubscriptionStateArg(),
+ option.getStartMessageRollbackDurationSec(), option.getReplicatedSubscriptionStateArg(),
option.getKeySharedMeta(), option.getSubscriptionProperties().orElse(null),
option.getSchemaType());
}
@@ -290,7 +290,7 @@ public class NonPersistentTopic extends AbstractTopic implements Topic, TopicPol
String consumerName, MessageId startMessageId,
Map<String, String> metadata, boolean readCompacted,
long resetStartMessageBackInSec,
- boolean replicateSubscriptionState,
+ Boolean replicateSubscriptionState,
KeySharedMeta keySharedMeta,
Map<String, String> subscriptionProperties,
SchemaType schemaType) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
index fde6a0716eb..4ae289d7bae 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
@@ -135,9 +135,11 @@ public class PersistentSubscription extends AbstractSubscription {
private volatile Map<String, String> subscriptionProperties;
private volatile CompletableFuture<Void> fenceFuture;
private volatile CompletableFuture<Void> inProgressResetCursorFuture;
+ private volatile Boolean replicatedControlled;
- static Map<String, Long> getBaseCursorProperties(boolean isReplicated) {
- return isReplicated ? REPLICATED_SUBSCRIPTION_CURSOR_PROPERTIES : NON_REPLICATED_SUBSCRIPTION_CURSOR_PROPERTIES;
+ static Map<String, Long> getBaseCursorProperties(Boolean isReplicated) {
+ return isReplicated != null && isReplicated ? REPLICATED_SUBSCRIPTION_CURSOR_PROPERTIES :
+ NON_REPLICATED_SUBSCRIPTION_CURSOR_PROPERTIES;
}
static boolean isCursorFromReplicatedSubscription(ManagedCursor cursor) {
@@ -145,19 +147,21 @@ public class PersistentSubscription extends AbstractSubscription {
}
public PersistentSubscription(PersistentTopic topic, String subscriptionName, ManagedCursor cursor,
- boolean replicated) {
+ Boolean replicated) {
this(topic, subscriptionName, cursor, replicated, Collections.emptyMap());
}
public PersistentSubscription(PersistentTopic topic, String subscriptionName, ManagedCursor cursor,
- boolean replicated, Map<String, String> subscriptionProperties) {
+ Boolean replicated, Map<String, String> subscriptionProperties) {
this.topic = topic;
this.cursor = cursor;
this.topicName = topic.getName();
this.subName = subscriptionName;
this.fullName = MoreObjects.toStringHelper(this).add("topic", topicName).add("name", subName).toString();
this.expiryMonitor = new PersistentMessageExpiryMonitor(topic, subscriptionName, cursor, this);
- this.setReplicated(replicated);
+ if (replicated != null) {
+ this.setReplicated(replicated);
+ }
this.subscriptionProperties = MapUtils.isEmpty(subscriptionProperties)
? Collections.emptyMap() : Collections.unmodifiableMap(subscriptionProperties);
if (topic.getBrokerService().getPulsar().getConfig().isTransactionCoordinatorEnabled()
@@ -196,6 +200,7 @@ public class PersistentSubscription extends AbstractSubscription {
}
public boolean setReplicated(boolean replicated) {
+ replicatedControlled = replicated;
ServiceConfiguration config = topic.getBrokerService().getPulsar().getConfig();
if (!replicated || !config.isEnableReplicatedSubscriptions()) {
@@ -1515,4 +1520,8 @@ public class PersistentSubscription extends AbstractSubscription {
private static final Logger log = LoggerFactory.getLogger(PersistentSubscription.class);
+ @VisibleForTesting
+ public Boolean getReplicatedControlled() {
+ return replicatedControlled;
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 20bdbe4effc..af3e021a97c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -535,7 +535,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
} else {
final String subscriptionName = Codec.decode(cursor.getName());
subscriptions.put(subscriptionName, createPersistentSubscription(subscriptionName, cursor,
- PersistentSubscription.isCursorFromReplicatedSubscription(cursor),
+ PersistentSubscription.isCursorFromReplicatedSubscription(cursor) ? true : null,
cursor.getCursorProperties()));
// subscription-cursor gets activated by default: deactivate as there is no active subscription
// right now
@@ -606,7 +606,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
private PersistentSubscription createPersistentSubscription(String subscriptionName, ManagedCursor cursor,
- boolean replicated, Map<String, String> subscriptionProperties) {
+ Boolean replicated, Map<String, String> subscriptionProperties) {
requireNonNull(topicCompactionService);
if (isCompactionSubscription(subscriptionName)
&& topicCompactionService instanceof PulsarTopicCompactionService pulsarTopicCompactionService) {
@@ -931,7 +931,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
option.getSubType(), option.getPriorityLevel(), option.getConsumerName(), option.isDurable(),
option.getStartMessageId(), option.getMetadata(), option.isReadCompacted(),
option.getInitialPosition(), option.getStartMessageRollbackDurationSec(),
- option.isReplicatedSubscriptionStateArg(), option.getKeySharedMeta(),
+ option.getReplicatedSubscriptionStateArg(), option.getKeySharedMeta(),
option.getSubscriptionProperties().orElse(Collections.emptyMap()),
option.getConsumerEpoch(), option.getSchemaType());
}
@@ -943,7 +943,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
Map<String, String> metadata, boolean readCompacted,
InitialPosition initialPosition,
long startMessageRollbackDurationSec,
- boolean replicatedSubscriptionStateArg,
+ Boolean replicatedSubscriptionStateArg,
KeySharedMeta keySharedMeta,
Map<String, String> subscriptionProperties,
long consumerEpoch,
@@ -954,12 +954,9 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
return brokerService.checkTopicNsOwnership(getName()).thenCompose(__ -> {
- boolean replicatedSubscriptionState = replicatedSubscriptionStateArg;
-
- if (replicatedSubscriptionState
+ if (replicatedSubscriptionStateArg != null && replicatedSubscriptionStateArg
&& !brokerService.pulsar().getConfiguration().isEnableReplicatedSubscriptions()) {
log.warn("[{}] Replicated Subscription is disabled by broker.", getName());
- replicatedSubscriptionState = false;
}
if (subType == SubType.Key_Shared
@@ -1028,7 +1025,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
CompletableFuture<? extends Subscription> subscriptionFuture = isDurable
? getDurableSubscription(subscriptionName, initialPosition, startMessageRollbackDurationSec,
- replicatedSubscriptionState, subscriptionProperties)
+ replicatedSubscriptionStateArg, subscriptionProperties)
: getNonDurableSubscription(subscriptionName, startMessageId, initialPosition,
startMessageRollbackDurationSec, readCompacted, subscriptionProperties);
@@ -1125,7 +1122,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
private CompletableFuture<Subscription> getDurableSubscription(String subscriptionName,
InitialPosition initialPosition,
long startMessageRollbackDurationSec,
- boolean replicated,
+ Boolean replicated,
Map<String, String> subscriptionProperties) {
CompletableFuture<Subscription> subscriptionFuture = new CompletableFuture<>();
if (checkMaxSubscriptionsPerTopicExceed(subscriptionName)) {
@@ -1156,7 +1153,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
return;
}
}
- if (replicated && !subscription.isReplicated()) {
+ if (replicated != null && replicated && !subscription.isReplicated()) {
// Flip the subscription state
subscription.setReplicated(replicated);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ReplicateSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ReplicateSubscriptionTest.java
new file mode 100644
index 00000000000..327081bf1b9
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ReplicateSubscriptionTest.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import org.apache.pulsar.broker.service.Subscription;
+import org.apache.pulsar.broker.service.Topic;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
+import org.apache.pulsar.client.impl.ConsumerBuilderImpl;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+public class ReplicateSubscriptionTest extends ProducerConsumerBase {
+
+ @BeforeClass
+ @Override
+ protected void setup() throws Exception {
+ super.internalSetup();
+ super.producerBaseSetup();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @Override
+ protected void doInitConf() throws Exception {
+ super.doInitConf();
+ }
+
+ @DataProvider
+ public Object[] replicateSubscriptionState() {
+ return new Object[]{
+ Boolean.TRUE,
+ Boolean.FALSE,
+ null
+ };
+ }
+
+ @Test(dataProvider = "replicateSubscriptionState")
+ public void testReplicateSubscriptionState(Boolean replicateSubscriptionState)
+ throws Exception {
+ String topic = "persistent://my-property/my-ns/" + System.nanoTime();
+ String subName = "sub-" + System.nanoTime();
+ ConsumerBuilder<String> consumerBuilder = pulsarClient.newConsumer(Schema.STRING)
+ .topic(topic)
+ .subscriptionName(subName);
+ if (replicateSubscriptionState != null) {
+ consumerBuilder.replicateSubscriptionState(replicateSubscriptionState);
+ }
+ ConsumerBuilderImpl consumerBuilderImpl = (ConsumerBuilderImpl) consumerBuilder;
+ assertEquals(consumerBuilderImpl.getConf().getReplicateSubscriptionState(), replicateSubscriptionState);
+ @Cleanup
+ Consumer<String> ignored = consumerBuilder.subscribe();
+ CompletableFuture<Optional<Topic>> topicIfExists = pulsar.getBrokerService().getTopicIfExists(topic);
+ assertThat(topicIfExists)
+ .succeedsWithin(3, TimeUnit.SECONDS)
+ .matches(optionalTopic -> {
+ assertTrue(optionalTopic.isPresent());
+ Topic topicRef = optionalTopic.get();
+ Subscription subscription = topicRef.getSubscription(subName);
+ assertNotNull(subscription);
+ assertTrue(subscription instanceof PersistentSubscription);
+ PersistentSubscription persistentSubscription = (PersistentSubscription) subscription;
+ assertEquals(persistentSubscription.getReplicatedControlled(), replicateSubscriptionState);
+ return true;
+ });
+ }
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index df5d2f7184f..8292fc35755 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -899,7 +899,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
synchronized (this) {
ByteBuf request = Commands.newSubscribe(topic, subscription, consumerId, requestId, getSubType(),
priorityLevel, consumerName, isDurable, startMessageIdData, metadata, readCompacted,
- conf.isReplicateSubscriptionState(),
+ conf.getReplicateSubscriptionState(),
InitialPosition.valueOf(subscriptionInitialPosition.getValue()),
startMessageRollbackDuration, si, createTopicIfDoesNotExist, conf.getKeySharedPolicy(),
// Use the current epoch to subscribe.
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java
index 3ae0e977d13..80a85940240 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.client.impl.conf;
import static com.google.common.base.Preconditions.checkArgument;
import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.collect.Sets;
import io.swagger.annotations.ApiModelProperty;
import java.io.Serializable;
@@ -378,7 +379,8 @@ public class ConsumerConfigurationData<T> implements Serializable, Cloneable {
value = "If `replicateSubscriptionState` is enabled, a subscription state is replicated to geo-replicated"
+ " clusters."
)
- private boolean replicateSubscriptionState = false;
+ @JsonProperty(access = JsonProperty.Access.READ_WRITE)
+ private Boolean replicateSubscriptionState;
private boolean resetIncludeHead = false;
@@ -434,4 +436,14 @@ public class ConsumerConfigurationData<T> implements Serializable, Cloneable {
throw new RuntimeException("Failed to clone ConsumerConfigurationData");
}
}
+
+ /**
+ * Backward compatibility with the old `replicateSubscriptionState` field.
+ * @deprecated Using {@link #getReplicateSubscriptionState()} instead.
+ */
+ @JsonIgnore
+ @Deprecated
+ public boolean isReplicateSubscriptionState() {
+ return replicateSubscriptionState != null && replicateSubscriptionState;
+ }
}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerBuilderImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerBuilderImplTest.java
index e4b7b4d1ec8..c103712d400 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerBuilderImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerBuilderImplTest.java
@@ -504,7 +504,7 @@ public class ConsumerBuilderImplTest {
assertTrue(configurationData.isRetryEnable());
assertFalse(configurationData.isAutoUpdatePartitions());
assertEquals(configurationData.getAutoUpdatePartitionsIntervalSeconds(), 2);
- assertTrue(configurationData.isReplicateSubscriptionState());
+ assertEquals(configurationData.getReplicateSubscriptionState(), Boolean.TRUE);
assertTrue(configurationData.isResetIncludeHead());
assertTrue(configurationData.isBatchIndexAckEnabled());
assertTrue(configurationData.isAckReceiptEnabled());
@@ -564,7 +564,7 @@ public class ConsumerBuilderImplTest {
assertFalse(configurationData.isRetryEnable());
assertTrue(configurationData.isAutoUpdatePartitions());
assertEquals(configurationData.getAutoUpdatePartitionsIntervalSeconds(), 60);
- assertFalse(configurationData.isReplicateSubscriptionState());
+ assertNull(configurationData.getReplicateSubscriptionState());
assertFalse(configurationData.isResetIncludeHead());
assertFalse(configurationData.isBatchIndexAckEnabled());
assertFalse(configurationData.isAckReceiptEnabled());
@@ -584,6 +584,38 @@ public class ConsumerBuilderImplTest {
assertNull(configurationData.getPayloadProcessor());
}
+ @Test
+ public void testReplicateSubscriptionState() {
+ ConsumerBuilderImpl<byte[]> consumerBuilder = createConsumerBuilder();
+ assertNull(consumerBuilder.getConf().getReplicateSubscriptionState());
+
+ consumerBuilder.replicateSubscriptionState(true);
+ assertEquals(consumerBuilder.getConf().getReplicateSubscriptionState(), Boolean.TRUE);
+
+ consumerBuilder.replicateSubscriptionState(false);
+ assertEquals(consumerBuilder.getConf().getReplicateSubscriptionState(), Boolean.FALSE);
+
+ Map<String, Object> conf = new HashMap<>();
+ consumerBuilder = createConsumerBuilder();
+ consumerBuilder.loadConf(conf);
+ assertNull(consumerBuilder.getConf().getReplicateSubscriptionState());
+
+ conf.put("replicateSubscriptionState", true);
+ consumerBuilder = createConsumerBuilder();
+ consumerBuilder.loadConf(conf);
+ assertEquals(consumerBuilder.getConf().getReplicateSubscriptionState(), Boolean.TRUE);
+
+ conf.put("replicateSubscriptionState", false);
+ consumerBuilder = createConsumerBuilder();
+ consumerBuilder.loadConf(conf);
+ assertEquals(consumerBuilder.getConf().getReplicateSubscriptionState(), Boolean.FALSE);
+
+ conf.put("replicateSubscriptionState", null);
+ consumerBuilder = createConsumerBuilder();
+ consumerBuilder.loadConf(conf);
+ assertNull(consumerBuilder.getConf().getReplicateSubscriptionState());
+ }
+
private ConsumerBuilderImpl<byte[]> createConsumerBuilder() {
ConsumerBuilderImpl<byte[]> consumerBuilder = new ConsumerBuilderImpl<>(null, Schema.BYTES);
Map<String, String> properties = new HashMap<>();
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
index 21604e2c944..2147aa8a871 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
@@ -583,7 +583,7 @@ public class Commands {
public static ByteBuf newSubscribe(String topic, String subscription, long consumerId, long requestId,
SubType subType, int priorityLevel, String consumerName, boolean isDurable, MessageIdData startMessageId,
- Map<String, String> metadata, boolean readCompacted, boolean isReplicated,
+ Map<String, String> metadata, boolean readCompacted, Boolean isReplicated,
InitialPosition subscriptionInitialPosition, long startMessageRollbackDurationInSec, SchemaInfo schemaInfo,
boolean createTopicIfDoesNotExist) {
return newSubscribe(topic, subscription, consumerId, requestId, subType, priorityLevel, consumerName,
@@ -594,7 +594,7 @@ public class Commands {
public static ByteBuf newSubscribe(String topic, String subscription, long consumerId, long requestId,
SubType subType, int priorityLevel, String consumerName, boolean isDurable, MessageIdData startMessageId,
- Map<String, String> metadata, boolean readCompacted, boolean isReplicated,
+ Map<String, String> metadata, boolean readCompacted, Boolean isReplicated,
InitialPosition subscriptionInitialPosition, long startMessageRollbackDurationInSec,
SchemaInfo schemaInfo, boolean createTopicIfDoesNotExist, KeySharedPolicy keySharedPolicy,
Map<String, String> subscriptionProperties, long consumerEpoch) {
@@ -610,9 +610,11 @@ public class Commands {
.setDurable(isDurable)
.setReadCompacted(readCompacted)
.setInitialPosition(subscriptionInitialPosition)
- .setReplicateSubscriptionState(isReplicated)
.setForceTopicCreation(createTopicIfDoesNotExist)
.setConsumerEpoch(consumerEpoch);
+ if (isReplicated != null) {
+ subscribe.setReplicateSubscriptionState(isReplicated);
+ }
if (subscriptionProperties != null && !subscriptionProperties.isEmpty()) {
List<KeyValue> keyValues = new ArrayList<>();