Skip to content

Commit

Permalink
[fix][broker] Check replication cluster before starting the replicator
Browse files Browse the repository at this point in the history
Fixes apache#20010

### Motivation

`PersistentTopicTest.testCreateTopicWithZombieReplicatorCursor` is flaky
because the cursor could still be created again in `startReplicator`,
which could be called by:

```
onPoliciesUpdate
  checkReplicationAndRetryOnFailure
    checkReplication
```

### Modifications

- Call `checkReplicationCluster` before calling `startReplicator`.
- Support retrying `initialize` to see if retry works.
  • Loading branch information
BewareMyPower committed Apr 6, 2023
1 parent 8c50a6c commit 81e4d78
Show file tree
Hide file tree
Showing 2 changed files with 19 additions and 4 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -1542,7 +1542,13 @@ public CompletableFuture<Void> checkReplication() {
continue;
}
if (!replicators.containsKey(cluster)) {
futures.add(startReplicator(cluster));
futures.add(checkReplicationCluster(cluster).thenCompose(clusterExists -> {
if (clusterExists) {
return startReplicator(cluster);
} else {
return CompletableFuture.completedFuture(null);
}
}));
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
import java.io.ByteArrayOutputStream;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
Expand All @@ -52,6 +53,7 @@
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Supplier;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedger;
Expand Down Expand Up @@ -83,6 +85,7 @@
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

@Slf4j
@Test(groups = "broker")
public class PersistentTopicTest extends BrokerTestBase {

Expand Down Expand Up @@ -584,8 +587,14 @@ 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());
Awaitility.await().atMost(Duration.ofSeconds(10)).until(() -> {
log.info("Before initialize...");
try {
topic.initialize().get(3, TimeUnit.SECONDS);
} catch (ExecutionException e) {
log.warn("Failed to initialize: {}", e.getCause().getMessage());
}
return !topic.getManagedLedger().getCursors().iterator().hasNext();
});
}
}

0 comments on commit 81e4d78

Please sign in to comment.