[improve][broker] Enhance compatibility with consumer replicate subscription state

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java

@@ -1084,7 +1084,7 @@ public void setReplicatedSubscriptionStatus(
             @ApiParam(value = "Whether leader broker redirected this call to this broker. For internal use.")
             @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
             @ApiParam(value = "Whether to enable replicated subscription", required = true)
-            Boolean enabled) {
+            boolean enabled) {
         try {
             validateTopicName(tenant, cluster, namespace, encodedTopic);
             internalSetReplicatedSubscriptionStatus(asyncResponse, decode(encodedSubName), authoritative, enabled);

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java

@@ -4405,7 +4405,7 @@ public void setReplicatedSubscriptionStatus(
             @ApiParam(value = "Whether leader broker redirected this call to this broker. For internal use.")
             @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
             @ApiParam(value = "Whether to enable replicated subscription", required = true)
-            Boolean enabled) {
+            boolean enabled) {
         try {
             validateTopicName(tenant, namespace, encodedTopic);
             internalSetReplicatedSubscriptionStatus(asyncResponse, decode(encodedSubName), authoritative, enabled);

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

@@ -248,8 +248,6 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
 
     private final long connectionLivenessCheckTimeoutMillis;
 
-    private final boolean ignoreConsumerReplicateSubscriptionState;
-
     // Number of bytes pending to be published from a single specific IO thread.
     private static final FastThreadLocal<MutableLong> pendingBytesPerThread = new FastThreadLocal<MutableLong>() {
         @Override
@@ -318,9 +316,6 @@ public ServerCnx(PulsarService pulsar, String listenerName) {
         this.topicListService = new TopicListService(pulsar, this,
                 enableSubscriptionPatternEvaluation, maxSubscriptionPatternLength);
         this.brokerInterceptor = this.service != null ? this.service.getInterceptor() : null;
-        this.ignoreConsumerReplicateSubscriptionState =
-                Boolean.parseBoolean(
-                        conf.getProperties().getProperty("ignoreConsumerReplicateSubscriptionState", "false"));
     }
 
     @Override
@@ -1201,8 +1196,8 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
                 ? subscribe.getStartMessageRollbackDurationSec()
                 : -1;
         final SchemaData schema = subscribe.hasSchema() ? getSchema(subscribe.getSchema()) : null;
-        final Boolean isReplicated = ignoreConsumerReplicateSubscriptionState ? null :
-                (subscribe.hasReplicateSubscriptionState() ? subscribe.isReplicateSubscriptionState() : null);
+        final Boolean isReplicated =
+                subscribe.hasReplicateSubscriptionState() ? subscribe.isReplicateSubscriptionState() : null;
         final boolean forceTopicCreation = subscribe.isForceTopicCreation();
         final KeySharedMeta keySharedMeta = subscribe.hasKeySharedMeta()
               ? new KeySharedMeta().copyFrom(subscribe.getKeySharedMeta())

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

@@ -39,7 +39,6 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
-import lombok.Getter;
 import org.apache.bookkeeper.mledger.AsyncCallbacks;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.ClearBacklogCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCallback;
@@ -125,45 +124,27 @@ public class PersistentSubscription extends AbstractSubscription implements Subs
     // Map of properties that is used to mark this subscription as "replicated".
     // Since this is the only field at this point, we can just keep a static
     // instance of the map.
-    private static final Map<String, Long> REPLICATED_SUBSCRIPTION_WITH_TRUE_CURSOR_PROPERTIES = new TreeMap<>();
-    private static final Map<String, Long> REPLICATED_SUBSCRIPTION_WITH_FALSE_CURSOR_PROPERTIES = new TreeMap<>();
+    private static final Map<String, Long> REPLICATED_SUBSCRIPTION_CURSOR_PROPERTIES = new TreeMap<>();
     private static final Map<String, Long> NON_REPLICATED_SUBSCRIPTION_CURSOR_PROPERTIES = Collections.emptyMap();
 
     private volatile ReplicatedSubscriptionSnapshotCache replicatedSubscriptionSnapshotCache;
     private final PendingAckHandle pendingAckHandle;
     private volatile Map<String, String> subscriptionProperties;
     private volatile CompletableFuture<Void> fenceFuture;
     private volatile CompletableFuture<Void> inProgressResetCursorFuture;
-    @Getter
-    private volatile Boolean replicatedControlled;
 
     static {
-        REPLICATED_SUBSCRIPTION_WITH_TRUE_CURSOR_PROPERTIES.put(REPLICATED_SUBSCRIPTION_PROPERTY, 1L);
-        REPLICATED_SUBSCRIPTION_WITH_FALSE_CURSOR_PROPERTIES.put(REPLICATED_SUBSCRIPTION_PROPERTY, 0L);
+        REPLICATED_SUBSCRIPTION_CURSOR_PROPERTIES.put(REPLICATED_SUBSCRIPTION_PROPERTY, 1L);
     }
 
     static Map<String, Long> getBaseCursorProperties(Boolean isReplicated) {
-        if (isReplicated == null) {
-            return NON_REPLICATED_SUBSCRIPTION_CURSOR_PROPERTIES;
-        }
-        return isReplicated ? REPLICATED_SUBSCRIPTION_WITH_TRUE_CURSOR_PROPERTIES :
-                REPLICATED_SUBSCRIPTION_WITH_FALSE_CURSOR_PROPERTIES;
-    }
-
-    /**
-     * When consumer replicateSubscriptionState or admin.topcis
-     * .setReplicatedSubscriptionStatus() is set, return the value from the cursor, otherwise return null.
-     */
-    protected static Boolean isCursorFromReplicatedSubscription(ManagedCursor cursor) {
-        return getReplicatedSubscriptionConfiguration(cursor).orElse(null);
+        return isReplicated != null && isReplicated ? REPLICATED_SUBSCRIPTION_CURSOR_PROPERTIES :
+                NON_REPLICATED_SUBSCRIPTION_CURSOR_PROPERTIES;
     }
 
-    protected static Optional<Boolean> getReplicatedSubscriptionConfiguration(ManagedCursor cursor) {
+    public static boolean isCursorFromReplicatedSubscription(ManagedCursor cursor) {
         Long v = cursor.getProperties().get(REPLICATED_SUBSCRIPTION_PROPERTY);
-        if (v == null || (v < 0L || v > 1L)) {
-            return Optional.empty();
-        }
-        return Optional.of(v == 1L);
+        return v != null && v == 1L;
     }
 
     public PersistentSubscription(PersistentTopic topic, String subscriptionName, ManagedCursor cursor,
@@ -179,9 +160,7 @@ public PersistentSubscription(PersistentTopic topic, String subscriptionName, Ma
         this.subName = subscriptionName;
         this.fullName = MoreObjects.toStringHelper(this).add("topic", topicName).add("name", subName).toString();
         this.expiryMonitor = new PersistentMessageExpiryMonitor(topicName, subscriptionName, cursor, this);
-        if (replicated != null) {
-            this.setReplicated(replicated);
-        }
+        this.setReplicated(replicated);
         this.subscriptionProperties = MapUtils.isEmpty(subscriptionProperties)
                 ? Collections.emptyMap() : Collections.unmodifiableMap(subscriptionProperties);
         if (topic.getBrokerService().getPulsar().getConfig().isTransactionCoordinatorEnabled()
@@ -220,14 +199,18 @@ public boolean isReplicated() {
     }
 
     public boolean setReplicated(Boolean replicated) {
-        replicatedControlled = replicated;
         return setReplicated(replicated, true);
     }
 
-    public boolean setReplicated(Boolean replicated, boolean isPersistent) {
+    protected boolean setReplicated(Boolean replicated, boolean isPersistent) {
         ServiceConfiguration config = topic.getBrokerService().getPulsar().getConfig();
+        if (!config.isEnableReplicatedSubscriptions()) {
+            log.warn("[{}][{}] Failed set replicated subscription status to {}, please enable the "
+                    + "configuration enableReplicatedSubscriptions", topicName, subName, replicated);
+            return false;
+        }
 
-        if (replicated == null || !replicated || !config.isEnableReplicatedSubscriptions()) {
+        if (replicated == null || !replicated) {
             this.replicatedSubscriptionSnapshotCache = null;
         } else if (this.replicatedSubscriptionSnapshotCache == null) {
             this.replicatedSubscriptionSnapshotCache = new ReplicatedSubscriptionSnapshotCache(subName,
@@ -241,15 +224,8 @@ public boolean setReplicated(Boolean replicated, boolean isPersistent) {
         if (this.cursor == null) {
             return false;
         }
-
-        if (!config.isEnableReplicatedSubscriptions()) {
-            log.warn("[{}][{}] Failed set replicated subscription status to {}, please enable the "
-                    + "configuration enableReplicatedSubscriptions", topicName, subName, replicated);
-            return false;
-        }
-
-        if (replicated != null) {
-            return this.cursor.putProperty(REPLICATED_SUBSCRIPTION_PROPERTY, Boolean.TRUE.equals(replicated) ? 1L : 0L);
+        if (replicated != null && replicated) {
+            return this.cursor.putProperty(REPLICATED_SUBSCRIPTION_PROPERTY, 1L);
         } else {
             return this.cursor.removeProperty(REPLICATED_SUBSCRIPTION_PROPERTY);
         }
@@ -1471,9 +1447,4 @@ public PositionInPendingAckStats checkPositionInPendingAckState(PositionImpl pos
     }
 
     private static final Logger log = LoggerFactory.getLogger(PersistentSubscription.class);
-
-    @VisibleForTesting
-    public Boolean getReplicatedControlled() {
-        return replicatedControlled;
-    }
 }

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

@@ -897,13 +897,6 @@ private CompletableFuture<Consumer> internalSubscribe(final TransportCnx cnx, St
         }
 
         return brokerService.checkTopicNsOwnership(getName()).thenCompose(__ -> {
-            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
                     && !brokerService.pulsar().getConfiguration().isSubscriptionKeySharedEnable()) {
                 return FutureUtil.failedFuture(
@@ -970,7 +963,7 @@ private CompletableFuture<Consumer> internalSubscribe(final TransportCnx cnx, St
 
             CompletableFuture<? extends Subscription> subscriptionFuture = isDurable
                     ? getDurableSubscription(subscriptionName, initialPosition, startMessageRollbackDurationSec,
-                    replicatedSubscriptionState, subscriptionProperties)
+                    replicatedSubscriptionStateArg, subscriptionProperties)
                     : getNonDurableSubscription(subscriptionName, startMessageId, initialPosition,
                     startMessageRollbackDurationSec, readCompacted, subscriptionProperties);
 
@@ -1098,6 +1091,10 @@ public void openCursorComplete(ManagedCursor cursor, Object ctx) {
                         return;
                     }
                 }
+                if (replicated != null && replicated && !subscription.isReplicated()) {
+                    // Flip the subscription state
+                    subscription.setReplicated(replicated);
+                }
 
                 if (startMessageRollbackDurationSec > 0) {
                     resetSubscriptionCursor(subscription, subscriptionFuture, startMessageRollbackDurationSec);
@@ -3781,14 +3778,22 @@ public CompletableFuture<Void> addSchemaIfIdleOrCheckCompatible(SchemaData schem
         });
     }
 
+    private boolean replicateSubscriptionStateEnabledByTopicPolicies;
+
     public synchronized void checkReplicatedSubscriptionControllerState() {
-        Boolean replicatedSubscriptionStatus = topicPolicies.getReplicateSubscriptionState().get();
+        Boolean replicatedSubscriptionState = topicPolicies.getReplicateSubscriptionState().get();
         AtomicBoolean shouldBeEnabled = new AtomicBoolean(false);
         subscriptions.forEach((name, subscription) -> {
-            // If the subscription does not have a replicated flag configured, please apply the topic policies to the
-            // subscription.
-            if (subscription.getReplicatedControlled() == null) {
-                subscription.setReplicated(replicatedSubscriptionStatus, false);
+            if (Boolean.TRUE.equals(replicatedSubscriptionState)) {
+                if (!subscription.isReplicated() && subscription.setReplicated(true, false)) {
+                    replicateSubscriptionStateEnabledByTopicPolicies = true;
+                }
+            } else if (replicateSubscriptionStateEnabledByTopicPolicies) {
+                if (!PersistentSubscription.isCursorFromReplicatedSubscription(subscription.getCursor())) {
+                    if (subscription.isReplicated()) {
+                        subscription.setReplicated(false, false);
+                    }
+                }
             }
             if (subscription.isReplicated()) {
                 shouldBeEnabled.set(true);
@@ -3799,6 +3804,10 @@ public synchronized void checkReplicatedSubscriptionControllerState() {
             if (log.isDebugEnabled()) {
                 log.debug("[{}] There are no replicated subscriptions on the topic", topic);
             }
+            // When no replication subscriptions, set the flag to false.
+            if (replicateSubscriptionStateEnabledByTopicPolicies) {
+                replicateSubscriptionStateEnabledByTopicPolicies = false;
+            }
         }
 
         checkReplicatedSubscriptionControllerState(shouldBeEnabled.get());

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentSubscriptionTest.java

@@ -294,34 +294,22 @@ public void testGetReplicatedSubscriptionConfiguration() {
         assertThat(properties).containsEntry(PersistentSubscription.REPLICATED_SUBSCRIPTION_PROPERTY, 1L);
         ManagedCursor cursor = mock(ManagedCursor.class);
         doReturn(properties).when(cursor).getProperties();
-        Optional<Boolean> replicatedSubscriptionConfiguration =
-                PersistentSubscription.getReplicatedSubscriptionConfiguration(cursor);
-        assertThat(replicatedSubscriptionConfiguration).isNotEmpty().get().isEqualTo(Boolean.TRUE);
+        assertThat(PersistentSubscription.isCursorFromReplicatedSubscription(cursor)).isTrue();
 
         properties = new HashMap<>();
         properties.put(PersistentSubscription.REPLICATED_SUBSCRIPTION_PROPERTY, 10L);
         doReturn(properties).when(cursor).getProperties();
-        replicatedSubscriptionConfiguration =
-                PersistentSubscription.getReplicatedSubscriptionConfiguration(cursor);
-        assertThat(replicatedSubscriptionConfiguration).isEmpty();
+        assertThat(PersistentSubscription.isCursorFromReplicatedSubscription(cursor)).isFalse();
+
         properties = new HashMap<>();
         properties.put(PersistentSubscription.REPLICATED_SUBSCRIPTION_PROPERTY, -1L);
         doReturn(properties).when(cursor).getProperties();
-        replicatedSubscriptionConfiguration =
-                PersistentSubscription.getReplicatedSubscriptionConfiguration(cursor);
-        assertThat(replicatedSubscriptionConfiguration).isEmpty();
+        assertThat(PersistentSubscription.isCursorFromReplicatedSubscription(cursor)).isFalse();
 
         properties = PersistentSubscription.getBaseCursorProperties(false);
-        assertThat(properties).containsEntry(PersistentSubscription.REPLICATED_SUBSCRIPTION_PROPERTY, 0L);
-        doReturn(properties).when(cursor).getProperties();
-        replicatedSubscriptionConfiguration =
-                PersistentSubscription.getReplicatedSubscriptionConfiguration(cursor);
-        assertThat(replicatedSubscriptionConfiguration).isNotEmpty().get().isEqualTo(Boolean.FALSE);
+        assertThat(properties).doesNotContainKey(PersistentSubscription.REPLICATED_SUBSCRIPTION_PROPERTY);
 
         properties = PersistentSubscription.getBaseCursorProperties(null);
-        doReturn(properties).when(cursor).getProperties();
-        replicatedSubscriptionConfiguration =
-                PersistentSubscription.getReplicatedSubscriptionConfiguration(cursor);
-        assertThat(replicatedSubscriptionConfiguration).isEmpty();
+        assertThat(properties).doesNotContainKey(PersistentSubscription.REPLICATED_SUBSCRIPTION_PROPERTY);
     }
 }