Skip to content
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

[pulsar-broker] Cleanup already deleted namespace topics #7473

Merged
merged 6 commits into from
Jul 14, 2020
Merged

[pulsar-broker] Cleanup already deleted namespace topics #7473

Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
570132d
[pulsar-broker] Cleanup already deleted namespace topics
rdhabalia Jul 7, 2020
972ae3e
add error-codes
rdhabalia Jul 7, 2020
487462b
Revert "add error-codes"
rdhabalia Jul 8, 2020
c5dd73a
Revert "[pulsar-broker] Cleanup already deleted namespace topics"
rdhabalia Jul 8, 2020
0b257ef
send success if producer is already closed
rdhabalia Jul 8, 2020
22bfe2c
send success if consumer is already closed
rdhabalia Jul 8, 2020
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
9 changes: 4 additions & 5 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1358,9 +1358,8 @@ protected void handleCloseProducer(CommandCloseProducer closeProducer) {

CompletableFuture<Producer> producerFuture = producers.get(producerId);
if (producerFuture == null) {
log.warn("[{}] Producer {} was not registered on the connection", remoteAddress, producerId);
ctx.writeAndFlush(Commands.newError(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 @@ -1401,8 +1400,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);
ctx.writeAndFlush(Commands.newError(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