Fix in ServerCnx to prevent using recycled commands (#1264)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
index bdcc6e3..85cedde 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -209,7 +209,7 @@
@Override
protected void handleLookup(CommandLookupTopic lookup) {
final long requestId = lookup.getRequestId();
-
+ final boolean authoritative = lookup.getAuthoritative();
if (log.isDebugEnabled()) {
log.debug("[{}] Received Lookup from {} for {}", lookup.getTopic(), remoteAddress, requestId);
}
@@ -254,9 +254,9 @@
String finalOriginalPrincipal = originalPrincipal;
isProxyAuthorizedFuture.thenApply(isProxyAuthorized -> {
if (isProxyAuthorized) {
- lookupDestinationAsync(getBrokerService().pulsar(), topicName, lookup.getAuthoritative(),
+ lookupDestinationAsync(getBrokerService().pulsar(), topicName, authoritative,
finalOriginalPrincipal != null ? finalOriginalPrincipal : authRole, authenticationData,
- lookup.getRequestId()).handle((lookupResponse, ex) -> {
+ requestId).handle((lookupResponse, ex) -> {
if (ex == null) {
ctx.writeAndFlush(lookupResponse);
} else {
@@ -550,7 +550,7 @@
subscribe.getStartMessageId().getLedgerId(), subscribe.getStartMessageId().getEntryId(),
subscribe.getStartMessageId().getPartition(), subscribe.getStartMessageId().getBatchIndex())
: null;
-
+ final String subscription = subscribe.getSubscription();
final int priorityLevel = subscribe.hasPriorityLevel() ? subscribe.getPriorityLevel() : 0;
final boolean readCompacted = subscribe.getReadCompacted();
final Map<String, String> metadata = CommandUtils.metadataFromCommand(subscribe);
@@ -568,7 +568,7 @@
if (service.isAuthorizationEnabled()) {
authorizationFuture = service.getAuthorizationService().canConsumeAsync(topicName,
originalPrincipal != null ? originalPrincipal : authRole, authenticationData,
- subscribe.getSubscription());
+ subscription);
} else {
authorizationFuture = CompletableFuture.completedFuture(true);
}
@@ -995,13 +995,13 @@
@Override
protected void handleSeek(CommandSeek seek) {
checkArgument(state == State.Connected);
-
+ final long requestId = seek.getRequestId();
CompletableFuture<Consumer> consumerFuture = consumers.get(seek.getConsumerId());
// Currently only seeking on a message id is supported
if (!seek.hasMessageId()) {
ctx.writeAndFlush(
- Commands.newError(seek.getRequestId(), ServerError.MetadataError, "Message id was not present"));
+ Commands.newError(requestId, ServerError.MetadataError, "Message id was not present"));
return;
}
@@ -1011,7 +1011,7 @@
MessageIdData msgIdData = seek.getMessageId();
Position position = new PositionImpl(msgIdData.getLedgerId(), msgIdData.getEntryId());
- long requestId = seek.getRequestId();
+
subscription.resetCursor(position).thenRun(() -> {
log.info("[{}] [{}][{}] Reset subscription to message id {}", remoteAddress,
@@ -1019,12 +1019,12 @@
ctx.writeAndFlush(Commands.newSuccess(requestId));
}).exceptionally(ex -> {
log.warn("[{}][{}] Failed to reset subscription: {}", remoteAddress, subscription, ex.getMessage(), ex);
- ctx.writeAndFlush(Commands.newError(seek.getRequestId(), ServerError.UnknownError,
+ ctx.writeAndFlush(Commands.newError(requestId, ServerError.UnknownError,
"Error when resetting subscription: " + ex.getCause().getMessage()));
return null;
});
} else {
- ctx.writeAndFlush(Commands.newError(seek.getRequestId(), ServerError.MetadataError, "Consumer not found"));
+ ctx.writeAndFlush(Commands.newError(requestId, ServerError.MetadataError, "Consumer not found"));
}
}
