codelipenghui commented on code in PR #21946: URL: https://github.com/apache/pulsar/pull/21946#discussion_r1533581014
########## pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java: ########## @@ -188,58 +274,134 @@ protected CompletableFuture<Boolean> isLocalTopicActive() { }, brokerService.executor()); } - protected synchronized CompletableFuture<Void> closeProducerAsync() { - if (producer == null) { - STATE_UPDATER.set(this, State.Stopped); + /** + * @Deprecated This method only be used by {@link PersistentTopic#checkGC} now. + * TODO "PersistentReplicator.replicateEntries" may get a NullPointerException if this method and a + * "cursor.readComplete" execute concurrently. + */ + @Deprecated Review Comment: I think we can left a TODO there to refactor the code. ########## pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java: ########## @@ -188,58 +274,134 @@ protected CompletableFuture<Boolean> isLocalTopicActive() { }, brokerService.executor()); } - protected synchronized CompletableFuture<Void> closeProducerAsync() { - if (producer == null) { - STATE_UPDATER.set(this, State.Stopped); + /** + * @Deprecated This method only be used by {@link PersistentTopic#checkGC} now. + * TODO "PersistentReplicator.replicateEntries" may get a NullPointerException if this method and a + * "cursor.readComplete" execute concurrently. + */ + @Deprecated Review Comment: If it still be used in PersistentTopic#checkGC, I don't think we should deprecate this method. Do we have any other alternative for `PersistentTopic#checkGC`? ########## pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java: ########## @@ -64,10 +70,35 @@ public abstract class AbstractReplicator { protected static final AtomicReferenceFieldUpdater<AbstractReplicator, State> STATE_UPDATER = AtomicReferenceFieldUpdater.newUpdater(AbstractReplicator.class, State.class, "state"); - private volatile State state = State.Stopped; - - protected enum State { - Stopped, Starting, Started, Stopping + @VisibleForTesting + @Getter + protected volatile State state = State.Stopped; + + public enum State { + /** + * This enum has two mean meanings:Init, Stopped. + * Regarding the meaning "Stopped", only {@link PersistentTopic#checkGC} will call {@link #disconnect}, + * so this method only be used by {@link PersistentTopic#checkGC} now. + * TODO After improving the method {@link #disconnect)}, we should rename "Stopped" to "Init". + */ + // The internal producer is stopped. + Stopped, + // Trying to create a new internal producer. + Starting, + // The internal producer has started, and tries copy data. + Started, + /** + * @Deprecated Only {@link PersistentTopic#checkGC} will call {@link #disconnect}, so this method only be + * used by {@link PersistentTopic#checkGC} now. + * TODO After improving the method {@link #disconnect)}, this enum should be removed. + */ + @Deprecated + // The internal producer is trying to stop. + Stopping, Review Comment: We still use `Stopping` when disconnecting the replicator. Why do we need add `@Deprecated` annotation? BTW, is it better to change `Stopping`, `Stopped` to `Disconnecting`, `Disconnected`? Honestly, It's not easy to understand what is the difference between stop, terminate. But disconnecting will be more better in this case. ########## pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java: ########## @@ -188,58 +274,134 @@ protected CompletableFuture<Boolean> isLocalTopicActive() { }, brokerService.executor()); } - protected synchronized CompletableFuture<Void> closeProducerAsync() { - if (producer == null) { - STATE_UPDATER.set(this, State.Stopped); + /** + * @Deprecated This method only be used by {@link PersistentTopic#checkGC} now. + * TODO "PersistentReplicator.replicateEntries" may get a NullPointerException if this method and a + * "cursor.readComplete" execute concurrently. + */ + @Deprecated + public CompletableFuture<Void> disconnect(boolean failIfHasBacklog, boolean closeTheStartingProducer) { + long backlog = getNumberOfEntriesInBacklog(); + if (failIfHasBacklog && backlog > 0) { + CompletableFuture<Void> disconnectFuture = new CompletableFuture<>(); + disconnectFuture.completeExceptionally(new TopicBusyException("Cannot close a replicator with backlog")); + if (log.isDebugEnabled()) { + log.debug("[{}] Replicator disconnect failed since topic has backlog", replicatorId); + } + return disconnectFuture; + } + log.info("[{}] Disconnect replicator at position {} with backlog {}", replicatorId, + getReplicatorReadPosition(), backlog); + return closeProducerAsync(closeTheStartingProducer); + } + + /** + * @Deprecated This method only be used by {@link PersistentTopic#checkGC} now. + * TODO "PersistentReplicator.replicateEntries" may get a NullPointerException if this method and a + * "cursor.readComplete" execute concurrently. + */ + @Deprecated Review Comment: Same as the above comment. ########## pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java: ########## @@ -101,78 +132,133 @@ public AbstractReplicator(String localCluster, Topic localTopic, String remoteCl protected abstract String getProducerName(); - protected abstract void readEntries(org.apache.pulsar.client.api.Producer<byte[]> producer); + protected abstract void setProducerAndTriggerReadEntries(org.apache.pulsar.client.api.Producer<byte[]> producer); protected abstract Position getReplicatorReadPosition(); - protected abstract long getNumberOfEntriesInBacklog(); + public abstract long getNumberOfEntriesInBacklog(); protected abstract void disableReplicatorRead(); public String getRemoteCluster() { return remoteCluster; } - // This method needs to be synchronized with disconnects else if there is a disconnect followed by startProducer - // the end result can be disconnect. - public synchronized void startProducer() { - if (STATE_UPDATER.get(this) == State.Stopping) { - long waitTimeMs = backOff.next(); - if (log.isDebugEnabled()) { - log.debug( - "[{}] waiting for producer to close before attempting to reconnect, retrying in {} s", - replicatorId, waitTimeMs / 1000.0); - } - // BackOff before retrying - brokerService.executor().schedule(this::checkTopicActiveAndRetryStartProducer, waitTimeMs, - TimeUnit.MILLISECONDS); - return; - } - State state = STATE_UPDATER.get(this); - if (!STATE_UPDATER.compareAndSet(this, State.Stopped, State.Starting)) { - if (state == State.Started) { - // Already running + public void startProducer() { + // Guarantee only one task call "producerBuilder.createAsync()". + Pair<Boolean, State> setStartingRes = compareSetAndGetState(State.Stopped, State.Starting); + if (!setStartingRes.getLeft()) { + if (setStartingRes.getRight() == State.Starting) { + log.info("[{}] Skip the producer creation since other thread is doing starting, state : {}", + replicatorId, state); + } else if (setStartingRes.getRight() == State.Started) { + // Since the method "startProducer" will be called even if it is started, only print debug-level log. + if (log.isDebugEnabled()) { + log.debug("[{}] Replicator was already running. state: {}", replicatorId, state); + } + } else if (setStartingRes.getRight() == State.Stopping) { if (log.isDebugEnabled()) { - log.debug("[{}] Replicator was already running", replicatorId); + log.debug("[{}] Rep.producer is closing, delay to retry(wait the producer close success)." + + " state: {}", replicatorId, state); } + delayStartProducerAfterStopped(); } else { - log.info("[{}] Replicator already being started. Replicator state: {}", replicatorId, state); + /** {@link State.Terminating}, {@link State.Terminated}. **/ + log.info("[{}] Skip the producer creation since the replicator state is : {}", replicatorId, state); } - return; } log.info("[{}] Starting replicator", replicatorId); producerBuilder.createAsync().thenAccept(producer -> { - readEntries(producer); + setProducerAndTriggerReadEntries(producer); Review Comment: Actually, this method can't ensure the the replicator can be started successfully. It just close the producer internally and return. So we will not get any exception in this case. Do we need to handle this case?https://github.com/apache/pulsar/pull/21946/files#diff-16c47c755ed17a62606a6b9a6b1dc65d5fe0baed21f6e3541df264bec81b83f6R168-R171 ########## pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java: ########## @@ -451,7 +472,7 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) { } HAVE_PENDING_READ_UPDATER.set(this, FALSE); - brokerService.executor().schedule(this::readMoreEntries, waitTimeMillis, TimeUnit.MILLISECONDS); + brokerService.executor().schedule(() -> readMoreEntries(), waitTimeMillis, TimeUnit.MILLISECONDS); Review Comment: We don't need this change. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@pulsar.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org