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 944148c5dc2af..871231f371ad5 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 @@ -790,23 +790,44 @@ protected void handleCommandScalableTopicLookup( return; } - // Create a DagWatchSession that will send the initial layout and watch for changes - var session = new org.apache.pulsar.broker.service.scalable.DagWatchSession( - sessionId, topicName, this, resources, service); - dagWatchSessions.put(sessionId, session); - - session.start() - .thenAcceptAsync(session::pushUpdate, ctx.executor()) - .exceptionally(ex -> { - Throwable cause = ex.getCause() != null ? ex.getCause() : ex; - log.warn().attr("topic", topicName).exception(cause) - .log("ScalableTopicLookup failed"); - dagWatchSessions.remove(sessionId); - session.close(); - ctx.executor().execute(() -> + isTopicOperationAllowed(topicName, TopicOperation.LOOKUP, authenticationData, originalAuthData) + .thenAccept(isAuthorized -> { + if (!isAuthorized) { + final String msg = "Client is not authorized to ScalableTopicLookup"; + log.warn() + .attr("principal", getPrincipal()) + .attr("topic", topicName) + .log(msg); ctx.writeAndFlush(Commands.newScalableTopicError(sessionId, - ServerError.TopicNotFound, cause.getMessage())) - ); + ServerError.AuthorizationError, msg)); + return; + } + // Create a DagWatchSession that will send the initial layout and watch for changes + var session = new org.apache.pulsar.broker.service.scalable.DagWatchSession( + sessionId, topicName, this, resources, service); + dagWatchSessions.put(sessionId, session); + + session.start() + .thenAcceptAsync(session::pushUpdate, ctx.executor()) + .exceptionally(ex -> { + Throwable cause = ex.getCause() != null ? ex.getCause() : ex; + log.warn().attr("topic", topicName).exception(cause) + .log("ScalableTopicLookup failed"); + dagWatchSessions.remove(sessionId); + session.close(); + ctx.executor().execute(() -> + ctx.writeAndFlush(Commands.newScalableTopicError(sessionId, + ServerError.TopicNotFound, cause.getMessage())) + ); + return null; + }); + }) + .exceptionally(ex -> { + logAuthException(remoteAddress, "scalable-topic-lookup", getPrincipal(), + Optional.of(topicName), ex); + ctx.writeAndFlush(Commands.newScalableTopicError(sessionId, + ServerError.AuthorizationError, + "Exception occurred while trying to authorize ScalableTopicLookup")); return null; }); } @@ -814,6 +835,11 @@ protected void handleCommandScalableTopicLookup( @Override protected void handleCommandScalableTopicClose( CommandScalableTopicClose commandScalableTopicClose) { + // No per-call authorization: the session is keyed in this connection's + // dagWatchSessions map (per-ServerCnx), authentication is enforced at connect, + // and the originating ScalableTopicLookup was authorized when the session was + // created. A close for an unknown sessionId is an idempotent no-op. Same + // pattern as handleCloseProducer / handleCloseConsumer. checkArgument(state == State.Connected); final long sessionId = commandScalableTopicClose.getSessionId(); @@ -880,23 +906,46 @@ protected void handleCommandScalableTopicSubscribe( return; } - scalableTopicService.registerConsumer(topicName, subscription, consumerName, consumerId, this) - .whenCompleteAsync((assignment, ex) -> { - if (ex != null) { - Throwable cause = ex.getCause() != null ? ex.getCause() : ex; - log.warn().attr("topic", topicName).attr("subscription", subscription) - .attr("consumerName", consumerName).exception(cause) - .log("ScalableTopicSubscribe failed"); + isTopicOperationAllowed(topicName, subscription, TopicOperation.CONSUME) + .thenAccept(isAuthorized -> { + if (!isAuthorized) { + final String msg = "Client is not authorized to ScalableTopicSubscribe"; + log.warn() + .attr("principal", getPrincipal()) + .attr("topic", topicName) + .attr("subscription", subscription) + .log(msg); getCommandSender().sendScalableTopicSubscribeError(requestId, - ServerError.UnknownError, cause.getMessage()); + ServerError.AuthorizationError, msg); return; } - // Record the registration so we can call onConsumerDisconnect on channelInactive. - scalableConsumerRegistrations.put(consumerId, - new ScalableConsumerRegistrationRef(topicName, subscription, consumerName)); - getCommandSender().sendScalableTopicSubscribeResponse(requestId, - org.apache.pulsar.broker.service.scalable.ConsumerSession.toProto(assignment)); - }, ctx.executor()); + scalableTopicService.registerConsumer(topicName, subscription, consumerName, + consumerId, this) + .whenCompleteAsync((assignment, ex) -> { + if (ex != null) { + Throwable cause = ex.getCause() != null ? ex.getCause() : ex; + log.warn().attr("topic", topicName).attr("subscription", subscription) + .attr("consumerName", consumerName).exception(cause) + .log("ScalableTopicSubscribe failed"); + getCommandSender().sendScalableTopicSubscribeError(requestId, + ServerError.UnknownError, cause.getMessage()); + return; + } + // Record the registration so we can call onConsumerDisconnect on channelInactive. + scalableConsumerRegistrations.put(consumerId, + new ScalableConsumerRegistrationRef(topicName, subscription, consumerName)); + getCommandSender().sendScalableTopicSubscribeResponse(requestId, + org.apache.pulsar.broker.service.scalable.ConsumerSession.toProto(assignment)); + }, ctx.executor()); + }) + .exceptionally(ex -> { + logAuthException(remoteAddress, "scalable-topic-subscribe", getPrincipal(), + Optional.of(topicName), ex); + getCommandSender().sendScalableTopicSubscribeError(requestId, + ServerError.AuthorizationError, + "Exception occurred while trying to authorize ScalableTopicSubscribe"); + return null; + }); } @Override diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ScalableTopicBinaryAuthZTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ScalableTopicBinaryAuthZTest.java new file mode 100644 index 0000000000000..bc0c680ea26dc --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ScalableTopicBinaryAuthZTest.java @@ -0,0 +1,309 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.service; + +import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgsRecordingInvocations; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.argThat; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelHandler; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.LengthFieldBasedFrameDecoder; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import org.apache.pulsar.broker.PulsarService; +import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.auth.MockAuthenticationProvider; +import org.apache.pulsar.broker.auth.MockAuthorizationProvider; +import org.apache.pulsar.broker.authentication.AuthenticationDataSubscription; +import org.apache.pulsar.broker.authentication.AuthenticationProvider; +import org.apache.pulsar.broker.authentication.AuthenticationService; +import org.apache.pulsar.broker.authorization.AuthorizationService; +import org.apache.pulsar.broker.namespace.NamespaceService; +import org.apache.pulsar.broker.service.scalable.ScalableTopicService; +import org.apache.pulsar.broker.service.utils.ClientChannelHelper; +import org.apache.pulsar.broker.testcontext.PulsarTestContext; +import org.apache.pulsar.common.api.proto.CommandConnected; +import org.apache.pulsar.common.api.proto.CommandScalableTopicSubscribeResponse; +import org.apache.pulsar.common.api.proto.CommandScalableTopicUpdate; +import org.apache.pulsar.common.api.proto.ScalableConsumerType; +import org.apache.pulsar.common.api.proto.ServerError; +import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.common.policies.data.TopicOperation; +import org.apache.pulsar.common.protocol.Commands; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +/** + * Authorization tests for the PIP-460/PIP-466 scalable-topic binary protocol commands. + * + *

Verifies that: + *

+ * + *

Tests drive raw protocol bytes through an {@link EmbeddedChannel} so the assertions exercise + * the wire-level handler paths in {@link ServerCnx} without depending on a client-side + * implementation of the new commands. + */ +@Test(groups = "broker") +public class ScalableTopicBinaryAuthZTest { + + private static final String AUTHORIZED_ROLE = "pass.pass"; + private static final String UNAUTHORIZED_ROLE = "pass.fail"; + private static final String TOPIC = "persistent://public/default/test-scalable-topic"; + private static final String SUBSCRIPTION = "test-scalable-sub"; + + private EmbeddedChannel channel; + private ServiceConfiguration svcConfig; + private ServerCnx serverCnx; + private PulsarTestContext pulsarTestContext; + private PulsarService pulsar; + private BrokerService brokerService; + private ClientChannelHelper clientChannelHelper; + private AuthorizationService authorizationService; + + @BeforeMethod(alwaysRun = true) + public void setup() throws Exception { + svcConfig = new ServiceConfiguration(); + svcConfig.setBrokerShutdownTimeoutMs(0L); + svcConfig.setLoadBalancerOverrideBrokerNicSpeedGbps(Optional.of(1.0d)); + svcConfig.setKeepAliveIntervalSeconds(60); + svcConfig.setBacklogQuotaCheckEnabled(false); + svcConfig.setClusterName("use"); + + pulsarTestContext = PulsarTestContext.builderForNonStartableContext() + .config(svcConfig) + .spyByDefault() + .build(); + pulsar = pulsarTestContext.getPulsarService(); + brokerService = pulsarTestContext.getBrokerService(); + + NamespaceService namespaceService = pulsar.getNamespaceService(); + when(namespaceService.getBundleAsync(any())).thenReturn(CompletableFuture.completedFuture(null)); + + clientChannelHelper = new ClientChannelHelper(); + } + + @AfterMethod(alwaysRun = true) + public void teardown() throws Exception { + if (serverCnx != null) { + serverCnx.close(); + } + if (channel != null) { + channel.close(); + } + if (pulsarTestContext != null) { + pulsarTestContext.close(); + pulsarTestContext = null; + } + } + + @Test + public void testScalableTopicLookupRequiresLookupPermission() throws Exception { + configureAuthAndConnect(UNAUTHORIZED_ROLE); + + long sessionId = 1L; + channel.writeInbound(Commands.newScalableTopicLookup(sessionId, TOPIC)); + + Object response = readResponse(); + assertTrue(response instanceof CommandScalableTopicUpdate, + "Expected CommandScalableTopicUpdate, got " + response); + CommandScalableTopicUpdate update = (CommandScalableTopicUpdate) response; + assertEquals(update.getSessionId(), sessionId); + assertTrue(update.hasError()); + assertEquals(update.getError(), ServerError.AuthorizationError); + + TopicName topicName = TopicName.get(TOPIC); + verify(authorizationService, times(1)) + .allowTopicOperationAsync(topicName, TopicOperation.LOOKUP, UNAUTHORIZED_ROLE, + serverCnx.getAuthData()); + } + + @Test + public void testScalableTopicLookupAllowsAuthorizedRole() throws Exception { + configureAuthAndConnect(AUTHORIZED_ROLE); + + long sessionId = 2L; + channel.writeInbound(Commands.newScalableTopicLookup(sessionId, TOPIC)); + + // The session may eventually emit a TopicNotFound error from the metadata-store watch (the + // topic does not exist in the test harness), but it must NOT emit AuthorizationError. + Object response = readResponse(); + assertTrue(response instanceof CommandScalableTopicUpdate, + "Expected CommandScalableTopicUpdate, got " + response); + CommandScalableTopicUpdate update = (CommandScalableTopicUpdate) response; + if (update.hasError()) { + assertEquals(update.getError(), ServerError.TopicNotFound, + "Authorized role must not be rejected with AuthorizationError"); + } + + TopicName topicName = TopicName.get(TOPIC); + verify(authorizationService, times(1)) + .allowTopicOperationAsync(topicName, TopicOperation.LOOKUP, AUTHORIZED_ROLE, + serverCnx.getAuthData()); + } + + @Test + public void testScalableTopicSubscribeRequiresConsumePermission() throws Exception { + configureAuthAndConnect(UNAUTHORIZED_ROLE); + // Pre-stub the scalable-topic service so authorization (not service availability) is what + // determines the response. The mock is never invoked because authorization fails first. + when(brokerService.getScalableTopicService()).thenReturn(mock(ScalableTopicService.class)); + + long requestId = 10L; + channel.writeInbound(Commands.newScalableTopicSubscribe(requestId, TOPIC, SUBSCRIPTION, + "test-consumer", 1L, ScalableConsumerType.STREAM)); + + Object response = readResponse(); + assertTrue(response instanceof CommandScalableTopicSubscribeResponse, + "Expected CommandScalableTopicSubscribeResponse, got " + response); + CommandScalableTopicSubscribeResponse subResp = (CommandScalableTopicSubscribeResponse) response; + assertEquals(subResp.getRequestId(), requestId); + assertTrue(subResp.hasError()); + assertEquals(subResp.getError(), ServerError.AuthorizationError); + + TopicName topicName = TopicName.get(TOPIC); + verify(authorizationService, times(1)).allowTopicOperationAsync( + eq(topicName), eq(TopicOperation.CONSUME), eq(UNAUTHORIZED_ROLE), + argThat(arg -> arg instanceof AuthenticationDataSubscription + && SUBSCRIPTION.equals(((AuthenticationDataSubscription) arg).getSubscription()))); + } + + @Test + public void testScalableTopicCloseDoesNotCallAuthorization() throws Exception { + configureAuthAndConnect(UNAUTHORIZED_ROLE); + + // Send a Close for an unknown sessionId. The handler must treat this as a no-op: + // no response, no AuthorizationError, and the AuthorizationService must not be consulted. + channel.writeInbound(Commands.newScalableTopicClose(99999L)); + channel.runPendingTasks(); + + assertTrue(channel.outboundMessages().isEmpty(), + "ScalableTopicClose for an unknown session must not emit any response"); + verify(authorizationService, never()).allowTopicOperationAsync(any(), any(), any(), any()); + } + + @Test + public void testScalableTopicLookupSkipsAuthzWhenFeatureDisabled() throws Exception { + svcConfig.setScalableTopicsEnabled(false); + configureAuthAndConnect(AUTHORIZED_ROLE); + + long sessionId = 3L; + channel.writeInbound(Commands.newScalableTopicLookup(sessionId, TOPIC)); + + Object response = readResponse(); + assertTrue(response instanceof CommandScalableTopicUpdate); + CommandScalableTopicUpdate update = (CommandScalableTopicUpdate) response; + assertEquals(update.getSessionId(), sessionId); + assertTrue(update.hasError()); + assertEquals(update.getError(), ServerError.NotAllowedError, + "Disabled feature must short-circuit before authorization"); + + verify(authorizationService, never()).allowTopicOperationAsync(any(), any(), any(), any()); + } + + @Test + public void testScalableTopicSubscribeSkipsAuthzWhenFeatureDisabled() throws Exception { + svcConfig.setScalableTopicsEnabled(false); + configureAuthAndConnect(AUTHORIZED_ROLE); + + long requestId = 11L; + channel.writeInbound(Commands.newScalableTopicSubscribe(requestId, TOPIC, SUBSCRIPTION, + "test-consumer", 1L, ScalableConsumerType.STREAM)); + + Object response = readResponse(); + assertTrue(response instanceof CommandScalableTopicSubscribeResponse); + CommandScalableTopicSubscribeResponse subResp = (CommandScalableTopicSubscribeResponse) response; + assertEquals(subResp.getRequestId(), requestId); + assertTrue(subResp.hasError()); + assertEquals(subResp.getError(), ServerError.NotAllowedError); + + verify(authorizationService, never()).allowTopicOperationAsync(any(), any(), any(), any()); + } + + private void configureAuthAndConnect(String role) throws Exception { + AuthenticationService authenticationService = mock(AuthenticationService.class); + AuthenticationProvider authenticationProvider = new MockAuthenticationProvider(); + String authMethodName = authenticationProvider.getAuthMethodName(); + when(brokerService.getAuthenticationService()).thenReturn(authenticationService); + when(authenticationService.getAuthenticationProvider(authMethodName)).thenReturn(authenticationProvider); + svcConfig.setAuthenticationEnabled(true); + + svcConfig.setAuthorizationProvider(MockAuthorizationProvider.class.getName()); + authorizationService = spyWithClassAndConstructorArgsRecordingInvocations( + AuthorizationService.class, svcConfig, pulsarTestContext.getPulsarResources()); + when(brokerService.getAuthorizationService()).thenReturn(authorizationService); + svcConfig.setAuthorizationEnabled(true); + + resetChannel(); + ByteBuf connect = Commands.newConnect(authMethodName, role, "test"); + channel.writeInbound(connect); + + Object response = readResponse(); + assertTrue(response instanceof CommandConnected, "Expected CommandConnected, got " + response); + } + + private void resetChannel() throws Exception { + if (channel != null && channel.isActive()) { + serverCnx.close(); + channel.close().get(); + } + serverCnx = new ServerCnx(pulsar); + serverCnx.setAuthRole(""); + channel = new EmbeddedChannel( + new LengthFieldBasedFrameDecoder(5 * 1024 * 1024, 0, 4, 0, 4), + (ChannelHandler) serverCnx); + } + + private Object readResponse() throws Exception { + long sleepMs = 10; + long iterations = TimeUnit.SECONDS.toMillis(10) / sleepMs; + for (int i = 0; i < iterations; i++) { + channel.runPendingTasks(); + if (!channel.outboundMessages().isEmpty()) { + Object outbound = channel.outboundMessages().remove(); + Object cmd = clientChannelHelper.getCommand(outbound); + if (cmd != null) { + return cmd; + } + } + Thread.sleep(sleepMs); + } + throw new AssertionError("Timed out waiting for response"); + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java index e37556a41a4e3..7de4c198f30df 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java @@ -138,6 +138,8 @@ import org.apache.pulsar.common.api.proto.CommandPartitionedTopicMetadataResponse; import org.apache.pulsar.common.api.proto.CommandPing; import org.apache.pulsar.common.api.proto.CommandProducerSuccess; +import org.apache.pulsar.common.api.proto.CommandScalableTopicSubscribeResponse; +import org.apache.pulsar.common.api.proto.CommandScalableTopicUpdate; import org.apache.pulsar.common.api.proto.CommandSendError; import org.apache.pulsar.common.api.proto.CommandSendReceipt; import org.apache.pulsar.common.api.proto.CommandSubscribe; @@ -147,6 +149,7 @@ import org.apache.pulsar.common.api.proto.CommandWatchTopicListSuccess; import org.apache.pulsar.common.api.proto.MessageMetadata; import org.apache.pulsar.common.api.proto.ProtocolVersion; +import org.apache.pulsar.common.api.proto.ScalableConsumerType; import org.apache.pulsar.common.api.proto.ServerError; import org.apache.pulsar.common.api.proto.Subscription; import org.apache.pulsar.common.api.proto.TxnAction; @@ -1505,6 +1508,89 @@ public void testVerifyAuthRoleAndAuthDataFromDirectConnectionBroker() throws Exc })); } + /** + * PIP-460/PIP-466: ScalableTopicLookup must require {@link TopicOperation#LOOKUP} on the parent + * topic, and ScalableTopicSubscribe must require {@link TopicOperation#CONSUME} with the + * subscription. ScalableTopicClose carries no authorization (the session is per-connection). + */ + @Test + public void testScalableTopicCommandsRequireTopicAuthorization() throws Exception { + AuthenticationService authenticationService = mock(AuthenticationService.class); + AuthenticationProvider authenticationProvider = new MockAuthenticationProvider(); + String authMethodName = authenticationProvider.getAuthMethodName(); + when(brokerService.getAuthenticationService()).thenReturn(authenticationService); + when(authenticationService.getAuthenticationProvider(authMethodName)).thenReturn(authenticationProvider); + svcConfig.setAuthenticationEnabled(true); + + svcConfig.setAuthorizationProvider("org.apache.pulsar.broker.auth.MockAuthorizationProvider"); + AuthorizationService authorizationService = + spyWithClassAndConstructorArgsRecordingInvocations(AuthorizationService.class, svcConfig, + pulsarTestContext.getPulsarResources()); + when(brokerService.getAuthorizationService()).thenReturn(authorizationService); + svcConfig.setAuthorizationEnabled(true); + + // Pre-stub the scalable-topic prerequisites so authorization (not service availability) + // is what determines the response. The mocks are never invoked because authorization fails first. + when(brokerService.getScalableTopicService()).thenReturn( + mock(org.apache.pulsar.broker.service.scalable.ScalableTopicService.class)); + + resetChannel(); + assertTrue(channel.isActive()); + assertEquals(serverCnx.getState(), State.Start); + + // Connect with a role that passes authentication and fails authorization in MockAuthorizationProvider. + String clientRole = "pass.fail"; + ByteBuf connect = Commands.newConnect(authMethodName, clientRole, "test"); + channel.writeInbound(connect); + Object connectResponse = getResponse(); + assertTrue(connectResponse instanceof CommandConnected); + + String topicStr = "persistent://public/default/test-scalable-topic"; + TopicName topicName = TopicName.get(topicStr); + + // ScalableTopicLookup -> CommandScalableTopicUpdate with AuthorizationError + long sessionId = 100L; + ByteBuf lookup = Commands.newScalableTopicLookup(sessionId, topicStr); + channel.writeInbound(lookup); + Object lookupResponse = getResponse(); + assertTrue(lookupResponse instanceof CommandScalableTopicUpdate); + CommandScalableTopicUpdate update = (CommandScalableTopicUpdate) lookupResponse; + assertEquals(update.getSessionId(), sessionId); + assertTrue(update.hasError()); + assertEquals(update.getError(), ServerError.AuthorizationError); + verify(authorizationService, times(1)) + .allowTopicOperationAsync(topicName, TopicOperation.LOOKUP, clientRole, serverCnx.getAuthData()); + + // ScalableTopicSubscribe -> CommandScalableTopicSubscribeResponse with AuthorizationError + String subscriptionName = "test-scalable-sub"; + long requestId = 200L; + ByteBuf subscribe = Commands.newScalableTopicSubscribe(requestId, topicStr, subscriptionName, + "test-consumer", 1L, ScalableConsumerType.STREAM); + channel.writeInbound(subscribe); + Object subscribeResponse = getResponse(); + assertTrue(subscribeResponse instanceof CommandScalableTopicSubscribeResponse); + CommandScalableTopicSubscribeResponse subResp = (CommandScalableTopicSubscribeResponse) subscribeResponse; + assertEquals(subResp.getRequestId(), requestId); + assertTrue(subResp.hasError()); + assertEquals(subResp.getError(), ServerError.AuthorizationError); + verify(authorizationService, times(1)).allowTopicOperationAsync( + eq(topicName), eq(TopicOperation.CONSUME), eq(clientRole), argThat(arg -> { + assertTrue(arg instanceof AuthenticationDataSubscription); + AuthenticationDataSubscription authData = (AuthenticationDataSubscription) arg; + assertEquals(authData.getSubscription(), subscriptionName); + return true; + })); + + // ScalableTopicClose for an unknown sessionId is an idempotent no-op: no response, no error. + ByteBuf close = Commands.newScalableTopicClose(99999L); + channel.writeInbound(close); + channel.runPendingTasks(); + assertTrue(channel.outboundMessages().isEmpty(), + "ScalableTopicClose for an unknown session must not emit any response"); + + channel.finish(); + } + @Test public void testRefreshOriginalPrincipalWithAuthDataForwardedFromProxy() throws Exception { AuthenticationService authenticationService = mock(AuthenticationService.class); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/utils/ClientChannelHelper.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/utils/ClientChannelHelper.java index d6210767ed1f3..32382cdb6fe4f 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/utils/ClientChannelHelper.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/utils/ClientChannelHelper.java @@ -45,6 +45,9 @@ import org.apache.pulsar.common.api.proto.CommandPong; import org.apache.pulsar.common.api.proto.CommandProducer; import org.apache.pulsar.common.api.proto.CommandProducerSuccess; +import org.apache.pulsar.common.api.proto.CommandScalableTopicAssignmentUpdate; +import org.apache.pulsar.common.api.proto.CommandScalableTopicSubscribeResponse; +import org.apache.pulsar.common.api.proto.CommandScalableTopicUpdate; import org.apache.pulsar.common.api.proto.CommandSend; import org.apache.pulsar.common.api.proto.CommandSendError; import org.apache.pulsar.common.api.proto.CommandSendReceipt; @@ -218,6 +221,23 @@ protected void handlePing(CommandPing ping) { protected void handlePong(CommandPong pong) { return; } + + @Override + protected void handleCommandScalableTopicUpdate(CommandScalableTopicUpdate update) { + queue.offer(new CommandScalableTopicUpdate().copyFrom(update)); + } + + @Override + protected void handleCommandScalableTopicSubscribeResponse( + CommandScalableTopicSubscribeResponse response) { + queue.offer(new CommandScalableTopicSubscribeResponse().copyFrom(response)); + } + + @Override + protected void handleCommandScalableTopicAssignmentUpdate( + CommandScalableTopicAssignmentUpdate update) { + queue.offer(new CommandScalableTopicAssignmentUpdate().copyFrom(update)); + } }; }