Fix in ServerCnx to prevent using recycled commands
@@ -208,7 +208,7 @@ public class ServerCnx extends PulsarHandler {
@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);
}
@@ -253,9 +253,9 @@ public class ServerCnx extends PulsarHandler {
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 {
@@ -549,7 +549,7 @@ public class ServerCnx extends PulsarHandler {
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);
@@ -567,7 +567,7 @@ public class ServerCnx extends PulsarHandler {
if (service.isAuthorizationEnabled()) {
authorizationFuture = service.getAuthorizationService().canConsumeAsync(topicName,
originalPrincipal != null ? originalPrincipal : authRole, authenticationData,
- subscribe.getSubscription());
+ subscription);
} else {
authorizationFuture = CompletableFuture.completedFuture(true);
}
@@ -994,13 +994,13 @@ public class ServerCnx extends PulsarHandler {
@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;
}
@@ -1010,7 +1010,7 @@ public class ServerCnx extends PulsarHandler {
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,
@@ -1018,12 +1018,12 @@ public class ServerCnx extends PulsarHandler {
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"));
}
}
