Skip to content

Commit

Permalink
Cleanup already deleted namespace topics. (apache#12597)
Browse files Browse the repository at this point in the history
Cherry pick from apache#7473.

apache#7473 has fix the `Cleanup already deleted namespace topics` issue, but with apache#8129 involved, changes have been 
changed back.

### Motivation
We are having frequent issues when user removes cluster from the global namespace where broker from removed-cluster fails to unload topic and namespace bundle still loaded with the broker. It happens when broker from removed-cluster receives below error
```
17:38:52.199 [pulsar-io-22-28] ERROR org.apache.pulsar.broker.service.persistent.PersistentReplicator - [persistent://prop/global/ns/tp1][east -> west] Failed to close dispatch rate limiter: org.apache.pulsar.client.api.PulsarClientException: Producer was not registered on the connection
:
17:38:52.199 [pulsar-io-22-28] WARN  org.apache.pulsar.broker.service.AbstractReplicator - [persistent://prop/global/ns/tp1][east -> west]] Exception: 'org.apache.pulsar.client.api.PulsarClientException: Producer was not registered on the connection' occured while trying to close the producer. retrying again in 0.1 s
:
17:38:52.351 [pulsar-io-22-37] ERROR org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://prop/global/ns/tp1] Error closing topic
java.util.concurrent.CompletionException: org.apache.pulsar.client.api.PulsarClientException: Producer was not registered on the connection
        at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:331) ~[?:?]
```

### Modification
Source Broker should return explicit error-code when producer is already closed and dest-broker from removed-cluster should handle this error and clean up the replicator and topic gracefully.
  • Loading branch information
Technoboy- authored and eolivelli committed Nov 29, 2021
1 parent 6228f26 commit 9cf5dad
Showing 1 changed file with 4 additions and 5 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -1579,9 +1579,8 @@ protected void handleCloseProducer(CommandCloseProducer closeProducer) {

CompletableFuture<Producer> producerFuture = producers.get(producerId);
if (producerFuture == null) {
log.warn("[{}] Producer was not registered on the connection. producerId={}", remoteAddress, producerId);
commandSender.sendErrorResponse(requestId, ServerError.UnknownError,
"Producer was not registered on the connection");
log.info("[{}] Producer {} was not registered on the connection", remoteAddress, producerId);
ctx.writeAndFlush(Commands.newSuccess(requestId));
return;
}

Expand Down Expand Up @@ -1626,8 +1625,8 @@ protected void handleCloseConsumer(CommandCloseConsumer closeConsumer) {

CompletableFuture<Consumer> consumerFuture = consumers.get(consumerId);
if (consumerFuture == null) {
log.warn("[{}] Consumer was not registered on the connection: consumerId={}", remoteAddress, consumerId);
commandSender.sendErrorResponse(requestId, ServerError.MetadataError, "Consumer not found");
log.info("[{}] Consumer was not registered on the connection: {}", consumerId, remoteAddress);
ctx.writeAndFlush(Commands.newSuccess(requestId));
return;
}

Expand Down

0 comments on commit 9cf5dad

Please sign in to comment.