poorbarcode commented on code in PR #21946:
URL: https://github.com/apache/pulsar/pull/21946#discussion_r1475764104
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java:
##########
@@ -188,14 +222,34 @@ protected CompletableFuture<Boolean> isLocalTopicActive()
{
}, brokerService.executor());
}
+ public synchronized CompletableFuture<Void> disconnect(boolean
failIfHasBacklog) {
+ if (failIfHasBacklog && getNumberOfEntriesInBacklog() > 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(), getNumberOfEntriesInBacklog());
+ if (!tryChangeStatusToTerminating()) {
Review Comment:
Added. And I moved `synchronized` after changing the status to `Stopping`
successfully.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java:
##########
@@ -163,16 +175,38 @@ public synchronized void startProducer() {
}
- protected void checkTopicActiveAndRetryStartProducer() {
- isLocalTopicActive().thenAccept(isTopicActive -> {
- if (isTopicActive) {
- startProducer();
+ protected void scheduleCheckTopicActiveAndStartProducer(final long
waitTimeMs) {
Review Comment:
Done
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
