[Broker] Fix race condition in stopping replicator while it is starting
there was a chance for a race condition that resulted in the replicator not disconnecting
if disconnect was called when the replicator was starting.
@@ -194,15 +194,13 @@ public abstract class AbstractReplicator {
return CompletableFuture.completedFuture(null);
}
- if (producer != null && (STATE_UPDATER.compareAndSet(this, State.Starting, State.Stopping)
- || STATE_UPDATER.compareAndSet(this, State.Started, State.Stopping))) {
+ if (STATE_UPDATER.compareAndSet(this, State.Starting, State.Stopping)
+ || STATE_UPDATER.compareAndSet(this, State.Started, State.Stopping)) {
log.info("[{}][{} -> {}] Disconnect replicator at position {} with backlog {}", topicName, localCluster,
remoteCluster, getReplicatorReadPosition(), getNumberOfEntriesInBacklog());
- return closeProducerAsync();
}
- STATE_UPDATER.set(this, State.Stopped);
- return CompletableFuture.completedFuture(null);
+ return closeProducerAsync();
}
public CompletableFuture<Void> remove() {
