Fix NPEs in PersistentReplicator
@@ -71,7 +71,7 @@ public class PersistentReplicator extends AbstractReplicator
private final PersistentTopic topic;
private final String replicatorName;
private final ManagedLedger ledger;
- protected ManagedCursor cursor;
+ protected volatile ManagedCursor cursor;
private Optional<DispatchRateLimiter> dispatchRateLimiter = Optional.empty();
@@ -190,12 +190,14 @@ public class PersistentReplicator extends AbstractReplicator
@Override
protected void disableReplicatorRead() {
+ if (this.cursor != null) {
// deactivate cursor after successfully close the producer
this.cursor.setInactive();
}
+ }
@Override
- protected synchronized CompletableFuture<Void> openCursorAsync() {
+ protected CompletableFuture<Void> openCursorAsync() {
log.info("[{}][{} -> {}] Starting open cursor for replicator", topicName, localCluster, remoteCluster);
if (cursor != null) {
log.info("[{}][{} -> {}] Using the exists cursor for replicator", topicName, localCluster, remoteCluster);
@@ -441,12 +443,14 @@ public class PersistentReplicator extends AbstractReplicator
}
public void updateCursorState() {
+ if (this.cursor != null) {
if (producer != null && producer.isConnected()) {
this.cursor.setActive();
} else {
this.cursor.setInactive();
}
}
+ }
private static final class ProducerSendCallback implements SendCallback {
private PersistentReplicator replicator;
@@ -688,7 +692,7 @@ public class PersistentReplicator extends AbstractReplicator
}
public ReplicatorStats getStats() {
- stats.replicationBacklog = cursor.getNumberOfEntriesInBacklog(false);
+ stats.replicationBacklog = cursor != null ? cursor.getNumberOfEntriesInBacklog(false) : 0;
stats.connected = producer != null && producer.isConnected();
stats.replicationDelayInSeconds = getReplicationDelayInSeconds();
