poorbarcode commented on code in PR #21946:
URL: https://github.com/apache/pulsar/pull/21946#discussion_r1475763765
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java:
##########
@@ -211,35 +265,67 @@ protected synchronized CompletableFuture<Void>
closeProducerAsync() {
});
}
-
- public CompletableFuture<Void> disconnect() {
- return disconnect(false);
+ public synchronized CompletableFuture<Void> terminate() {
+ if (producer == null) {
Review Comment:
Added. And I moved `synchronized` after changing the status to `Terminating`
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) {
+ brokerService.executor().schedule(() -> {
+ if (state == State.Terminated) {
+ return;
}
- }).exceptionally(ex -> {
- log.warn("[{}] Stop retry to create producer due to topic load
fail. Replicator state: {}", replicatorId,
- STATE_UPDATER.get(this), ex);
- return null;
- });
+ CompletableFuture<Optional<Topic>> topicFuture =
brokerService.getTopics().get(localTopicName);
+ if (topicFuture == null) {
+ // Topic closed.
+ return;
+ }
+ topicFuture.thenAccept(optional -> {
+ if (optional.isEmpty()) {
+ // Topic closed.
+ return;
+ }
+ if (optional.get() != localTopic) {
+ // Topic closed and created a new one, current replicator
is outdated.
+ return;
+ }
+ // TODO check isClosing or Deleting.
Review Comment:
Removed.
##########
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) {
+ brokerService.executor().schedule(() -> {
+ if (state == State.Terminated) {
Review Comment:
> And it's better to add an info level log here so that we can know why the
producer creation is skipped.
Done
> And it looks like we also have 4 more case below to skip the producer
creation. Is it better to have a log for each of them? We can have a method to
print the log with the skipped reason.
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]
