Skip to content

[fix] [broker] Remove the repl cursor if the creation fails and the replication is turned off #20025

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
poorbarcode wants to merge 8 commits into from
Closed

[fix] [broker] Remove the repl cursor if the creation fails and the replication is turned off #20025

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
9dd077e
[fix] [broker] Ignore and remove the replicator cursor when the remot…
poorbarcode Apr 6, 2023
903466c
fix flaky test
poorbarcode Apr 6, 2023
1ce4fb9
address comments
poorbarcode Apr 6, 2023
20f8a39
address comment
poorbarcode Apr 6, 2023
5e15522
code format
poorbarcode Apr 6, 2023
593e66c
add logs
poorbarcode Apr 6, 2023
dd5aafa
revert the changes which not effected of the flaky test
poorbarcode Apr 6, 2023
395f40c
revert the changes which not effected of the flaky test
poorbarcode Apr 6, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
74 changes: 46 additions & 28 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1706,37 +1706,55 @@ private CompletableFuture<Boolean> checkReplicationCluster(String remoteCluster)

protected CompletableFuture<Void> addReplicationCluster(String remoteCluster, ManagedCursor cursor,
String localCluster) {
return AbstractReplicator.validatePartitionedTopicAsync(PersistentTopic.this.getName(), brokerService)
.thenCompose(__ -> checkReplicationCluster(remoteCluster))
.thenCompose(clusterExists -> {
if (!clusterExists) {
log.warn("Remove the replicator because the cluster '{}' does not exist", remoteCluster);
return removeReplicator(remoteCluster).thenApply(__ -> null);
}
return brokerService.pulsar().getPulsarResources().getClusterResources()
.getClusterAsync(remoteCluster)
.thenApply(clusterData ->
brokerService.getReplicationClient(remoteCluster, clusterData));
})
.thenAccept(replicationClient -> {
if (replicationClient == null) {
return;
CompletableFuture<Void> replicationStartFuture = new CompletableFuture<>();
AbstractReplicator.validatePartitionedTopicAsync(PersistentTopic.this.getName(), brokerService)
.thenCompose(__ -> brokerService.pulsar().getPulsarResources().getClusterResources()
.getClusterAsync(remoteCluster)
.thenApply(clusterData ->
brokerService.getReplicationClient(remoteCluster, clusterData)))
.thenAccept(replicationClient -> {
Replicator replicator = replicators.computeIfAbsent(remoteCluster, r -> {
try {
return new GeoPersistentReplicator(PersistentTopic.this, cursor, localCluster,
remoteCluster, brokerService, (PulsarClientImpl) replicationClient);
} catch (PulsarServerException e) {
log.error("[{}] Replicator startup failed {}", topic, remoteCluster, e);
}
Replicator replicator = replicators.computeIfAbsent(remoteCluster, r -> {
try {
return new GeoPersistentReplicator(PersistentTopic.this, cursor, localCluster,
remoteCluster, brokerService, (PulsarClientImpl) replicationClient);
} catch (PulsarServerException e) {
log.error("[{}] Replicator startup failed {}", topic, remoteCluster, e);
return null;
});
// clean up replicator if startup is failed
if (replicator == null) {
replicators.removeNullValue(remoteCluster);
}
}).whenComplete((ignore, ex) -> {
if (ex == null){
replicationStartFuture.complete(null);
} else {
Copy link
Copy Markdown
Contributor

@BewareMyPower BewareMyPower Apr 6, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's better to print the ex warning here.

checkReplicationCluster(remoteCluster).thenCompose(clusterExists -> {
if (!clusterExists) {
log.warn("[{}] Start remove the replicator because the cluster '{}' does not exist",
topic, remoteCluster);
return removeReplicator(remoteCluster).whenComplete((ignore2, removeCursorEx) -> {
if (removeCursorEx != null) {
log.error("[{}] Remove the cursor of replicator[{}] is failed, please reload topic."
, topic, remoteCluster);
replicationStartFuture.failedFuture(removeCursorEx);
} else {
log.warn("[{}] Remove the cursor of replicator[{}] successfully",
topic, remoteCluster);
replicationStartFuture.complete(null);
}
});
} else {
// Start replication is failed.
log.error("[{}] The replicator startup failed {}", topic, remoteCluster, ex);
replicationStartFuture.completeExceptionally(ex);
return CompletableFuture.completedFuture(null);
}
return null;
});

// clean up replicator if startup is failed
if (replicator == null) {
replicators.removeNullValue(remoteCluster);
}
});
}
});
return replicationStartFuture;
}

CompletableFuture<Void> removeReplicator(String remoteCluster) {
Expand Down
17 changes: 13 additions & 4 deletions ...broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
Expand Down Expand Up @@ -558,7 +559,10 @@ public void testCreateTopicWithZombieReplicatorCursor(boolean topicLevelPolicy)
admin.tenants().updateTenant("prop", tenantInfo);

if (topicLevelPolicy) {
admin.topics().setReplicationClusters(topicName, Collections.singletonList(remoteCluster));
// setReplicationClusters may fail, so do retry.
Awaitility.await().ignoreExceptions().untilAsserted(() -> {
admin.topics().setReplicationClusters(topicName, Collections.singletonList(remoteCluster));
});
} else {
admin.namespaces().setNamespaceReplicationClustersAsync(
namespace, Collections.singleton(remoteCluster)).get();
Expand All @@ -584,8 +588,13 @@ public void testCreateTopicWithZombieReplicatorCursor(boolean topicLevelPolicy)
admin.clusters().deleteCluster(remoteCluster);
// Now the cluster and its related policy has been removed but the replicator cursor still exists

topic.initialize().get(3, TimeUnit.SECONDS);
Awaitility.await().atMost(3, TimeUnit.SECONDS)
.until(() -> !topic.getManagedLedger().getCursors().iterator().hasNext());
// Verify:
// 1. Topic can load success. If the topic loading by client is failed, it will retry,
// so we can do retry "initialize topic".
// 2. The repl cursor will be deleted.
Awaitility.await().atMost(10, TimeUnit.SECONDS).ignoreExceptions().untilAsserted(() -> {
topic.initialize().get(3, TimeUnit.SECONDS);
assertFalse(topic.getManagedLedger().getCursors().iterator().hasNext());
});
}
}