From c6f3485473ab415ed5749833e0284fc8b493142c Mon Sep 17 00:00:00 2001 From: Philipp Dolif Date: Sun, 13 Oct 2024 11:58:59 +0200 Subject: [PATCH 1/5] Refactor PersistentSubscriptionTest: Remove unused fields --- .../service/persistent/PersistentSubscriptionTest.java | 9 --------- 1 file changed, 9 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentSubscriptionTest.java index 309cd7b55ac0c..9a46a2919d12a 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentSubscriptionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentSubscriptionTest.java @@ -26,7 +26,6 @@ import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; -import io.netty.channel.EventLoopGroup; import java.lang.reflect.Field; import java.util.ArrayList; import java.util.Collections; @@ -35,7 +34,6 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; -import org.apache.bookkeeper.common.util.OrderedExecutor; import org.apache.bookkeeper.mledger.AsyncCallbacks; import org.apache.bookkeeper.mledger.ManagedLedger; import org.apache.bookkeeper.mledger.ManagedLedgerConfig; @@ -60,8 +58,6 @@ import org.apache.pulsar.common.policies.data.Policies; import org.apache.pulsar.transaction.common.exception.TransactionConflictException; import org.awaitility.Awaitility; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; @@ -83,11 +79,6 @@ public class PersistentSubscriptionTest { final TxnID txnID1 = new TxnID(1,1); final TxnID txnID2 = new TxnID(1,2); - private static final Logger log = LoggerFactory.getLogger(PersistentTopicTest.class); - - private OrderedExecutor executor; - private EventLoopGroup eventLoopGroup; - @BeforeMethod public void setup() throws Exception { pulsarTestContext = PulsarTestContext.builderForNonStartableContext() From 592c82b36861dd9aa2e139ba439bf319ab0f1cbc Mon Sep 17 00:00:00 2001 From: Philipp Dolif Date: Sun, 13 Oct 2024 12:20:37 +0200 Subject: [PATCH 2/5] [fix][broker] Key_Shared subscription: Reject consumers with incompatible policy --- .../NonPersistentSubscription.java | 13 ++++ .../persistent/PersistentSubscription.java | 13 ++++ .../broker/service/SubscriptionTestBase.java | 78 +++++++++++++++++++ .../NonPersistentSubscriptionTest.java | 64 +++++++++++++++ .../PersistentSubscriptionTest.java | 37 ++++++++- 5 files changed, 201 insertions(+), 4 deletions(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionTestBase.java create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscriptionTest.java diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java index e92eef5cb7bff..acf2a7c81b419 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java @@ -161,6 +161,19 @@ public synchronized CompletableFuture addConsumer(Consumer consumer) { } else { if (consumer.subType() != dispatcher.getType()) { return FutureUtil.failedFuture(new SubscriptionBusyException("Subscription is of different type")); + } else if (dispatcher.getType() == SubType.Key_Shared) { + KeySharedMeta dispatcherKsm = dispatcher.getConsumers().get(0).getKeySharedMeta(); + KeySharedMeta consumerKsm = consumer.getKeySharedMeta(); + if (dispatcherKsm.getKeySharedMode() != consumerKsm.getKeySharedMode()) { + return FutureUtil.failedFuture( + new SubscriptionBusyException("Subscription is of different key_shared mode")); + } + if (dispatcherKsm.isAllowOutOfOrderDelivery() != consumerKsm.isAllowOutOfOrderDelivery()) { + return FutureUtil.failedFuture( + new SubscriptionBusyException(dispatcherKsm.isAllowOutOfOrderDelivery() + ? "Subscription allows out of order delivery" : + "Subscription does not allow out of order delivery")); + } } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java index df1c23cbbcb30..e43104d7ce3aa 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java @@ -310,6 +310,19 @@ private CompletableFuture addConsumerInternal(Consumer consumer) { if (consumer.subType() != dispatcher.getType()) { return FutureUtil.failedFuture( new SubscriptionBusyException("Subscription is of different type")); + } else if (dispatcher.getType() == SubType.Key_Shared) { + KeySharedMeta dispatcherKsm = dispatcher.getConsumers().get(0).getKeySharedMeta(); + KeySharedMeta consumerKsm = consumer.getKeySharedMeta(); + if (dispatcherKsm.getKeySharedMode() != consumerKsm.getKeySharedMode()) { + return FutureUtil.failedFuture( + new SubscriptionBusyException("Subscription is of different key_shared mode")); + } + if (dispatcherKsm.isAllowOutOfOrderDelivery() != consumerKsm.isAllowOutOfOrderDelivery()) { + return FutureUtil.failedFuture( + new SubscriptionBusyException(dispatcherKsm.isAllowOutOfOrderDelivery() + ? "Subscription allows out of order delivery" : + "Subscription does not allow out of order delivery")); + } } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionTestBase.java new file mode 100644 index 0000000000000..caf48f101374d --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionTestBase.java @@ -0,0 +1,78 @@ +/* + * 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.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import org.apache.pulsar.broker.BrokerTestUtil; +import org.apache.pulsar.common.api.proto.CommandSubscribe; +import org.apache.pulsar.common.api.proto.KeySharedMeta; +import org.apache.pulsar.common.api.proto.KeySharedMode; +import org.testng.annotations.DataProvider; + +public abstract class SubscriptionTestBase { + + protected final String successTopicName = "persistent://prop/use/ns-abc/successTopic"; + protected final String subName = "subscriptionName"; + + @DataProvider(name = "incompatibleKeySharedPolicies") + public Object[][] incompatibleKeySharedPolicies() { + KeySharedMeta ksmSticky = new KeySharedMeta().setKeySharedMode(KeySharedMode.STICKY); + ksmSticky.addHashRange().setStart(0).setEnd(2); + + KeySharedMeta ksmStickyAllowOutOfOrder = new KeySharedMeta().setKeySharedMode(KeySharedMode.STICKY) + .setAllowOutOfOrderDelivery(true); + ksmStickyAllowOutOfOrder.addHashRange().setStart(3).setEnd(5); + + KeySharedMeta ksmAutoSplit = new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT); + KeySharedMeta ksmAutoSplitAllowOutOfOrder = new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT) + .setAllowOutOfOrderDelivery(true); + + String errorMessageDifferentMode = "Subscription is of different key_shared mode"; + String errorMessageOutOfOrderNotAllowed = "Subscription does not allow out of order delivery"; + String errorMessageOutOfOrderAllowed = "Subscription allows out of order delivery"; + + return new Object[][] { + { ksmAutoSplit, ksmSticky, errorMessageDifferentMode }, + { ksmAutoSplit, ksmStickyAllowOutOfOrder, errorMessageDifferentMode }, + { ksmAutoSplit, ksmAutoSplitAllowOutOfOrder, errorMessageOutOfOrderNotAllowed }, + + { ksmAutoSplitAllowOutOfOrder, ksmSticky, errorMessageDifferentMode }, + { ksmAutoSplitAllowOutOfOrder, ksmStickyAllowOutOfOrder, errorMessageDifferentMode }, + { ksmAutoSplitAllowOutOfOrder, ksmAutoSplit, errorMessageOutOfOrderAllowed }, + + { ksmSticky, ksmStickyAllowOutOfOrder, errorMessageOutOfOrderNotAllowed }, + { ksmSticky, ksmAutoSplit, errorMessageDifferentMode }, + { ksmSticky, ksmAutoSplitAllowOutOfOrder, errorMessageDifferentMode }, + + { ksmStickyAllowOutOfOrder, ksmSticky, errorMessageOutOfOrderAllowed }, + { ksmStickyAllowOutOfOrder, ksmAutoSplit, errorMessageDifferentMode }, + { ksmStickyAllowOutOfOrder, ksmAutoSplitAllowOutOfOrder, errorMessageDifferentMode } + }; + } + + protected Consumer createKeySharedMockConsumer(String name, KeySharedMeta ksm) { + Consumer consumer = BrokerTestUtil.createMockConsumer(name); + doReturn(CommandSubscribe.SubType.Key_Shared).when(consumer).subType(); + doReturn(ksm).when(consumer).getKeySharedMeta(); + doReturn(mock(PendingAcksMap.class)).when(consumer).getPendingAcks(); + return consumer; + } + +} \ No newline at end of file diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscriptionTest.java new file mode 100644 index 0000000000000..6ccecef2d3223 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscriptionTest.java @@ -0,0 +1,64 @@ +/* + * 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.nonpersistent; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; +import static org.testng.Assert.fail; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionBusyException; +import org.apache.pulsar.broker.service.Consumer; +import org.apache.pulsar.broker.service.SubscriptionTestBase; +import org.apache.pulsar.broker.testcontext.PulsarTestContext; +import org.apache.pulsar.common.api.proto.KeySharedMeta; +import org.testng.annotations.Test; + +@Test(groups = "broker") +public class NonPersistentSubscriptionTest extends SubscriptionTestBase { + + @Test(dataProvider = "incompatibleKeySharedPolicies") + public void testIncompatibleKeySharedPoliciesNotAllowed(KeySharedMeta consumer1Ksm, KeySharedMeta consumer2Ksm, + String expectedErrorMessage) throws Exception { + PulsarTestContext context = PulsarTestContext.builderForNonStartableContext().build(); + NonPersistentTopic topic = new NonPersistentTopic(successTopicName, context.getBrokerService()); + NonPersistentSubscription sub = new NonPersistentSubscription(topic, subName, Map.of()); + + // two consumers with incompatible key_shared policies + Consumer keySharedConsumerMock1 = createKeySharedMockConsumer("consumer-1", consumer1Ksm); + Consumer keySharedConsumerMock2 = createKeySharedMockConsumer("consumer-2", consumer2Ksm); + + // first consumer defines key_shared mode of subscription and whether out of order delivery is allowed + sub.addConsumer(keySharedConsumerMock1).get(5, TimeUnit.SECONDS); + + try { + // add second consumer with incompatible key_shared policy + sub.addConsumer(keySharedConsumerMock2).get(5, TimeUnit.SECONDS); + fail(SubscriptionBusyException.class.getSimpleName() + " not thrown"); + } catch (Exception e) { + // subscription throws exception when consumer with incompatible key_shared policy is added + Throwable cause = e.getCause(); + assertTrue(cause instanceof SubscriptionBusyException); + assertEquals(cause.getMessage(), expectedErrorMessage); + } + + context.close(); + } + +} \ No newline at end of file diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentSubscriptionTest.java index 9a46a2919d12a..53e78c5ad173d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentSubscriptionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentSubscriptionTest.java @@ -34,6 +34,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; import org.apache.bookkeeper.mledger.AsyncCallbacks; import org.apache.bookkeeper.mledger.ManagedLedger; import org.apache.bookkeeper.mledger.ManagedLedgerConfig; @@ -44,7 +45,9 @@ import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.commons.lang3.tuple.MutablePair; import org.apache.pulsar.broker.resources.NamespaceResources; +import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionBusyException; import org.apache.pulsar.broker.service.Consumer; +import org.apache.pulsar.broker.service.SubscriptionTestBase; import org.apache.pulsar.broker.testcontext.PulsarTestContext; import org.apache.pulsar.broker.transaction.buffer.impl.InMemTransactionBufferProvider; import org.apache.pulsar.broker.transaction.pendingack.PendingAckStore; @@ -54,6 +57,7 @@ import org.apache.pulsar.client.api.transaction.TxnID; import org.apache.pulsar.common.api.proto.CommandAck.AckType; import org.apache.pulsar.common.api.proto.CommandSubscribe; +import org.apache.pulsar.common.api.proto.KeySharedMeta; import org.apache.pulsar.common.api.proto.TxnAction; import org.apache.pulsar.common.policies.data.Policies; import org.apache.pulsar.transaction.common.exception.TransactionConflictException; @@ -63,7 +67,7 @@ import org.testng.annotations.Test; @Test(groups = "broker") -public class PersistentSubscriptionTest { +public class PersistentSubscriptionTest extends SubscriptionTestBase { private PulsarTestContext pulsarTestContext; private ManagedLedger ledgerMock; @@ -73,9 +77,6 @@ public class PersistentSubscriptionTest { private Consumer consumerMock; private ManagedLedgerConfig managedLedgerConfigMock; - final String successTopicName = "persistent://prop/use/ns-abc/successTopic"; - final String subName = "subscriptionName"; - final TxnID txnID1 = new TxnID(1,1); final TxnID txnID2 = new TxnID(1,2); @@ -226,6 +227,34 @@ public void testAcknowledgeUpdateCursorLastActive() throws Exception { assertTrue(persistentSubscription.cursor.getLastActive() > beforeAcknowledgeTimestamp); } + @Test(dataProvider = "incompatibleKeySharedPolicies") + public void testIncompatibleKeySharedPoliciesNotAllowed(KeySharedMeta consumer1Ksm, KeySharedMeta consumer2Ksm, + String expectedErrorMessage) throws Exception { + PulsarTestContext context = PulsarTestContext.builderForNonStartableContext().build(); + PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, context.getBrokerService()); + PersistentSubscription sub = new PersistentSubscription(topic, subName, cursorMock, false); + + // two consumers with incompatible key_shared policies + Consumer keySharedConsumerMock1 = createKeySharedMockConsumer("consumer-1", consumer1Ksm); + Consumer keySharedConsumerMock2 = createKeySharedMockConsumer("consumer-2", consumer2Ksm); + + // first consumer defines key_shared mode of subscription and whether out of order delivery is allowed + sub.addConsumer(keySharedConsumerMock1).get(5, TimeUnit.SECONDS); + + try { + // add second consumer with incompatible key_shared policy + sub.addConsumer(keySharedConsumerMock2).get(5, TimeUnit.SECONDS); + fail(SubscriptionBusyException.class.getSimpleName() + " not thrown"); + } catch (Exception e) { + // subscription throws exception when consumer with incompatible key_shared policy is added + Throwable cause = e.getCause(); + assertTrue(cause instanceof SubscriptionBusyException); + assertEquals(cause.getMessage(), expectedErrorMessage); + } + + context.close(); + } + public static class CustomTransactionPendingAckStoreProvider implements TransactionPendingAckStoreProvider { @Override public CompletableFuture newPendingAckStore(PersistentSubscription subscription) { From 82005e8399a114d4c7283c12afaa9a43fa8adde2 Mon Sep 17 00:00:00 2001 From: Philipp Dolif Date: Mon, 14 Oct 2024 19:01:35 +0200 Subject: [PATCH 3/5] Subscription: Update rejection error message for consumers with incompatible subscription type or key_shared policy --- .../NonPersistentSubscription.java | 23 ++++++++++----- .../persistent/PersistentSubscription.java | 24 ++++++++++------ .../broker/service/SubscriptionTestBase.java | 28 +++++++++++-------- 3 files changed, 49 insertions(+), 26 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java index acf2a7c81b419..5a5ad34448908 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java @@ -160,19 +160,28 @@ public synchronized CompletableFuture addConsumer(Consumer consumer) { } } else { if (consumer.subType() != dispatcher.getType()) { - return FutureUtil.failedFuture(new SubscriptionBusyException("Subscription is of different type")); + return FutureUtil.failedFuture(new SubscriptionBusyException( + String.format("Subscription is of different type. Active subscription type of '%s' " + + "is different than the connecting consumer's type '%s'.", + dispatcher.getType(), consumer.subType()))); } else if (dispatcher.getType() == SubType.Key_Shared) { KeySharedMeta dispatcherKsm = dispatcher.getConsumers().get(0).getKeySharedMeta(); KeySharedMeta consumerKsm = consumer.getKeySharedMeta(); if (dispatcherKsm.getKeySharedMode() != consumerKsm.getKeySharedMode()) { - return FutureUtil.failedFuture( - new SubscriptionBusyException("Subscription is of different key_shared mode")); + return FutureUtil.failedFuture(new SubscriptionBusyException( + String.format("Subscription is of different type. Active subscription key_shared " + + "mode of '%s' is different than the connecting consumer's " + + "key_shared mode '%s'.", + dispatcherKsm.getKeySharedMode(), consumerKsm.getKeySharedMode()))); } if (dispatcherKsm.isAllowOutOfOrderDelivery() != consumerKsm.isAllowOutOfOrderDelivery()) { - return FutureUtil.failedFuture( - new SubscriptionBusyException(dispatcherKsm.isAllowOutOfOrderDelivery() - ? "Subscription allows out of order delivery" : - "Subscription does not allow out of order delivery")); + return FutureUtil.failedFuture(new SubscriptionBusyException( + String.format("Subscription is of different type. %s", + dispatcherKsm.isAllowOutOfOrderDelivery() + ? "Active subscription allows out of order delivery while the " + + "connecting consumer does not allow it." : + "Active subscription does not allow out of order delivery while " + + "the connecting consumer allows it."))); } } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java index e43104d7ce3aa..16831c241a187 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java @@ -308,20 +308,28 @@ private CompletableFuture addConsumerInternal(Consumer consumer) { } } else { if (consumer.subType() != dispatcher.getType()) { - return FutureUtil.failedFuture( - new SubscriptionBusyException("Subscription is of different type")); + return FutureUtil.failedFuture(new SubscriptionBusyException( + String.format("Subscription is of different type. Active subscription type of '%s' " + + "is different than the connecting consumer's type '%s'.", + dispatcher.getType(), consumer.subType()))); } else if (dispatcher.getType() == SubType.Key_Shared) { KeySharedMeta dispatcherKsm = dispatcher.getConsumers().get(0).getKeySharedMeta(); KeySharedMeta consumerKsm = consumer.getKeySharedMeta(); if (dispatcherKsm.getKeySharedMode() != consumerKsm.getKeySharedMode()) { - return FutureUtil.failedFuture( - new SubscriptionBusyException("Subscription is of different key_shared mode")); + return FutureUtil.failedFuture(new SubscriptionBusyException( + String.format("Subscription is of different type. Active subscription key_shared " + + "mode of '%s' is different than the connecting consumer's " + + "key_shared mode '%s'.", + dispatcherKsm.getKeySharedMode(), consumerKsm.getKeySharedMode()))); } if (dispatcherKsm.isAllowOutOfOrderDelivery() != consumerKsm.isAllowOutOfOrderDelivery()) { - return FutureUtil.failedFuture( - new SubscriptionBusyException(dispatcherKsm.isAllowOutOfOrderDelivery() - ? "Subscription allows out of order delivery" : - "Subscription does not allow out of order delivery")); + return FutureUtil.failedFuture(new SubscriptionBusyException( + String.format("Subscription is of different type. %s", + dispatcherKsm.isAllowOutOfOrderDelivery() + ? "Active subscription allows out of order delivery while the " + + "connecting consumer does not allow it." : + "Active subscription does not allow out of order delivery while " + + "the connecting consumer allows it."))); } } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionTestBase.java index caf48f101374d..4bede327f1872 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionTestBase.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionTestBase.java @@ -44,26 +44,32 @@ public Object[][] incompatibleKeySharedPolicies() { KeySharedMeta ksmAutoSplitAllowOutOfOrder = new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT) .setAllowOutOfOrderDelivery(true); - String errorMessageDifferentMode = "Subscription is of different key_shared mode"; - String errorMessageOutOfOrderNotAllowed = "Subscription does not allow out of order delivery"; - String errorMessageOutOfOrderAllowed = "Subscription allows out of order delivery"; + String errorMessagePrefix = "Subscription is of different type. "; + String errorMessageSubscriptionModeSticky = errorMessagePrefix + "Active subscription key_shared mode of " + + "'STICKY' is different than the connecting consumer's key_shared mode 'AUTO_SPLIT'."; + String errorMessageSubscriptionModeAutoSplit = errorMessagePrefix + "Active subscription key_shared mode of " + + "'AUTO_SPLIT' is different than the connecting consumer's key_shared mode 'STICKY'."; + String errorMessageOutOfOrderNotAllowed = errorMessagePrefix + "Active subscription does not allow out of " + + "order delivery while the connecting consumer allows it."; + String errorMessageOutOfOrderAllowed = errorMessagePrefix + "Active subscription allows out of order delivery " + + "while the connecting consumer does not allow it."; return new Object[][] { - { ksmAutoSplit, ksmSticky, errorMessageDifferentMode }, - { ksmAutoSplit, ksmStickyAllowOutOfOrder, errorMessageDifferentMode }, + { ksmAutoSplit, ksmSticky, errorMessageSubscriptionModeAutoSplit }, + { ksmAutoSplit, ksmStickyAllowOutOfOrder, errorMessageSubscriptionModeAutoSplit }, { ksmAutoSplit, ksmAutoSplitAllowOutOfOrder, errorMessageOutOfOrderNotAllowed }, - { ksmAutoSplitAllowOutOfOrder, ksmSticky, errorMessageDifferentMode }, - { ksmAutoSplitAllowOutOfOrder, ksmStickyAllowOutOfOrder, errorMessageDifferentMode }, + { ksmAutoSplitAllowOutOfOrder, ksmSticky, errorMessageSubscriptionModeAutoSplit }, + { ksmAutoSplitAllowOutOfOrder, ksmStickyAllowOutOfOrder, errorMessageSubscriptionModeAutoSplit }, { ksmAutoSplitAllowOutOfOrder, ksmAutoSplit, errorMessageOutOfOrderAllowed }, { ksmSticky, ksmStickyAllowOutOfOrder, errorMessageOutOfOrderNotAllowed }, - { ksmSticky, ksmAutoSplit, errorMessageDifferentMode }, - { ksmSticky, ksmAutoSplitAllowOutOfOrder, errorMessageDifferentMode }, + { ksmSticky, ksmAutoSplit, errorMessageSubscriptionModeSticky }, + { ksmSticky, ksmAutoSplitAllowOutOfOrder, errorMessageSubscriptionModeSticky }, { ksmStickyAllowOutOfOrder, ksmSticky, errorMessageOutOfOrderAllowed }, - { ksmStickyAllowOutOfOrder, ksmAutoSplit, errorMessageDifferentMode }, - { ksmStickyAllowOutOfOrder, ksmAutoSplitAllowOutOfOrder, errorMessageDifferentMode } + { ksmStickyAllowOutOfOrder, ksmAutoSplit, errorMessageSubscriptionModeSticky }, + { ksmStickyAllowOutOfOrder, ksmAutoSplitAllowOutOfOrder, errorMessageSubscriptionModeSticky } }; } From 328f1bd1287ee8b4d60f2e0f13bb9bf92346ff70 Mon Sep 17 00:00:00 2001 From: Philipp Dolif Date: Tue, 15 Oct 2024 19:35:49 +0200 Subject: [PATCH 4/5] Refactoring: Move shared code for consumer/dispatcher compatibility error check to AbstractSubscription --- .../broker/service/AbstractSubscription.java | 46 +++++++++++++++++++ .../NonPersistentSubscription.java | 29 ++---------- .../persistent/PersistentSubscription.java | 28 ++--------- 3 files changed, 54 insertions(+), 49 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractSubscription.java index 10e6b79609721..ae108646532dc 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractSubscription.java @@ -19,8 +19,12 @@ package org.apache.pulsar.broker.service; import java.util.Optional; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.LongAdder; import java.util.function.ToLongFunction; +import org.apache.pulsar.common.api.proto.CommandSubscribe; +import org.apache.pulsar.common.api.proto.KeySharedMeta; +import org.apache.pulsar.common.util.FutureUtil; public abstract class AbstractSubscription implements Subscription { protected final LongAdder bytesOutFromRemovedConsumers = new LongAdder(); @@ -39,4 +43,46 @@ private long sumConsumers(ToLongFunction toCounter) { .map(dispatcher -> dispatcher.getConsumers().stream().mapToLong(toCounter).sum()) .orElse(0L); } + + /** + * Checks if the given consumer is compatible with the given dispatcher. They are incompatible if + *
    + *
  • the subscription type of the consumer differs from the subscription type of the dispatcher or
    • + *
  • if both the consumer and dispatcher are of {@link CommandSubscribe.SubType#Key_Shared} type but + * their policies differ ({@link org.apache.pulsar.common.api.proto.KeySharedMode#AUTO_SPLIT}/ + * {@link org.apache.pulsar.common.api.proto.KeySharedMode#STICKY} or allowOutOfOrderDelivery true/false).
    • + *
+ * @param dispatcher The dispatcher of the subscription + * @param consumer New consumer to be added to the subscription + * @return Optional containing failed future with {@link BrokerServiceException.SubscriptionBusyException} if + * consumer and dispatcher are incompatible or empty optional otherwise. + */ + protected Optional> checkForConsumerCompatibilityErrorWithDispatcher(Dispatcher dispatcher, + Consumer consumer) { + if (consumer.subType() != dispatcher.getType()) { + return Optional.of(FutureUtil.failedFuture(new BrokerServiceException.SubscriptionBusyException( + String.format("Subscription is of different type. Active subscription type of '%s' is different " + + "than the connecting consumer's type '%s'.", + dispatcher.getType(), consumer.subType())))); + } else if (dispatcher.getType() == CommandSubscribe.SubType.Key_Shared) { + KeySharedMeta dispatcherKsm = dispatcher.getConsumers().get(0).getKeySharedMeta(); + KeySharedMeta consumerKsm = consumer.getKeySharedMeta(); + if (dispatcherKsm.getKeySharedMode() != consumerKsm.getKeySharedMode()) { + return Optional.of(FutureUtil.failedFuture(new BrokerServiceException.SubscriptionBusyException( + String.format("Subscription is of different type. Active subscription key_shared mode of '%s' " + + "is different than the connecting consumer's key_shared mode '%s'.", + dispatcherKsm.getKeySharedMode(), consumerKsm.getKeySharedMode())))); + } + if (dispatcherKsm.isAllowOutOfOrderDelivery() != consumerKsm.isAllowOutOfOrderDelivery()) { + return Optional.of(FutureUtil.failedFuture(new BrokerServiceException.SubscriptionBusyException( + String.format("Subscription is of different type. %s", + dispatcherKsm.isAllowOutOfOrderDelivery() + ? "Active subscription allows out of order delivery while the connecting " + + "consumer does not allow it." : + "Active subscription does not allow out of order delivery while the connecting " + + "consumer allows it.")))); + } + } + return Optional.empty(); + } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java index 5a5ad34448908..55639868e0ac1 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java @@ -34,7 +34,6 @@ import org.apache.pulsar.broker.service.AnalyzeBacklogResult; import org.apache.pulsar.broker.service.BrokerServiceException; import org.apache.pulsar.broker.service.BrokerServiceException.ServerMetadataException; -import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionBusyException; import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionFencedException; import org.apache.pulsar.broker.service.Consumer; import org.apache.pulsar.broker.service.Dispatcher; @@ -159,30 +158,10 @@ public synchronized CompletableFuture addConsumer(Consumer consumer) { }); } } else { - if (consumer.subType() != dispatcher.getType()) { - return FutureUtil.failedFuture(new SubscriptionBusyException( - String.format("Subscription is of different type. Active subscription type of '%s' " - + "is different than the connecting consumer's type '%s'.", - dispatcher.getType(), consumer.subType()))); - } else if (dispatcher.getType() == SubType.Key_Shared) { - KeySharedMeta dispatcherKsm = dispatcher.getConsumers().get(0).getKeySharedMeta(); - KeySharedMeta consumerKsm = consumer.getKeySharedMeta(); - if (dispatcherKsm.getKeySharedMode() != consumerKsm.getKeySharedMode()) { - return FutureUtil.failedFuture(new SubscriptionBusyException( - String.format("Subscription is of different type. Active subscription key_shared " - + "mode of '%s' is different than the connecting consumer's " - + "key_shared mode '%s'.", - dispatcherKsm.getKeySharedMode(), consumerKsm.getKeySharedMode()))); - } - if (dispatcherKsm.isAllowOutOfOrderDelivery() != consumerKsm.isAllowOutOfOrderDelivery()) { - return FutureUtil.failedFuture(new SubscriptionBusyException( - String.format("Subscription is of different type. %s", - dispatcherKsm.isAllowOutOfOrderDelivery() - ? "Active subscription allows out of order delivery while the " - + "connecting consumer does not allow it." : - "Active subscription does not allow out of order delivery while " - + "the connecting consumer allows it."))); - } + Optional> compatibilityError = + checkForConsumerCompatibilityErrorWithDispatcher(dispatcher, consumer); + if (compatibilityError.isPresent()) { + return compatibilityError.get(); } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java index 16831c241a187..0096f398ada91 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java @@ -307,30 +307,10 @@ private CompletableFuture addConsumerInternal(Consumer consumer) { }); } } else { - if (consumer.subType() != dispatcher.getType()) { - return FutureUtil.failedFuture(new SubscriptionBusyException( - String.format("Subscription is of different type. Active subscription type of '%s' " - + "is different than the connecting consumer's type '%s'.", - dispatcher.getType(), consumer.subType()))); - } else if (dispatcher.getType() == SubType.Key_Shared) { - KeySharedMeta dispatcherKsm = dispatcher.getConsumers().get(0).getKeySharedMeta(); - KeySharedMeta consumerKsm = consumer.getKeySharedMeta(); - if (dispatcherKsm.getKeySharedMode() != consumerKsm.getKeySharedMode()) { - return FutureUtil.failedFuture(new SubscriptionBusyException( - String.format("Subscription is of different type. Active subscription key_shared " - + "mode of '%s' is different than the connecting consumer's " - + "key_shared mode '%s'.", - dispatcherKsm.getKeySharedMode(), consumerKsm.getKeySharedMode()))); - } - if (dispatcherKsm.isAllowOutOfOrderDelivery() != consumerKsm.isAllowOutOfOrderDelivery()) { - return FutureUtil.failedFuture(new SubscriptionBusyException( - String.format("Subscription is of different type. %s", - dispatcherKsm.isAllowOutOfOrderDelivery() - ? "Active subscription allows out of order delivery while the " - + "connecting consumer does not allow it." : - "Active subscription does not allow out of order delivery while " - + "the connecting consumer allows it."))); - } + Optional> compatibilityError = + checkForConsumerCompatibilityErrorWithDispatcher(dispatcher, consumer); + if (compatibilityError.isPresent()) { + return compatibilityError.get(); } } From c63f0c1387efba89dc03f7dcfdeb77ce786946a4 Mon Sep 17 00:00:00 2001 From: Philipp Dolif Date: Sun, 20 Oct 2024 19:41:18 +0200 Subject: [PATCH 5/5] Move subscription and consumer compatibility tests to separate test class --- ...SubscriptionConsumerCompatibilityTest.java | 175 ++++++++++++++++++ .../broker/service/SubscriptionTestBase.java | 84 --------- .../NonPersistentSubscriptionTest.java | 64 ------- .../PersistentSubscriptionTest.java | 37 +--- 4 files changed, 179 insertions(+), 181 deletions(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionConsumerCompatibilityTest.java delete mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionTestBase.java delete mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscriptionTest.java diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionConsumerCompatibilityTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionConsumerCompatibilityTest.java new file mode 100644 index 0000000000000..dc1d0170f9d72 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionConsumerCompatibilityTest.java @@ -0,0 +1,175 @@ +/* + * 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.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; +import static org.testng.Assert.fail; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.apache.bookkeeper.mledger.ManagedLedger; +import org.apache.bookkeeper.mledger.ManagedLedgerConfig; +import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl; +import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; +import org.apache.pulsar.broker.BrokerTestUtil; +import org.apache.pulsar.broker.service.nonpersistent.NonPersistentSubscription; +import org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic; +import org.apache.pulsar.broker.service.persistent.PersistentSubscription; +import org.apache.pulsar.broker.service.persistent.PersistentTopic; +import org.apache.pulsar.broker.testcontext.PulsarTestContext; +import org.apache.pulsar.common.api.proto.CommandSubscribe; +import org.apache.pulsar.common.api.proto.KeySharedMeta; +import org.apache.pulsar.common.api.proto.KeySharedMode; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +@Test(groups = "broker") +public class SubscriptionConsumerCompatibilityTest { + + private PulsarTestContext pulsarTestContext; + private ManagedLedger ledgerMock; + private ManagedCursorImpl cursorMock; + private final String successTopicName = "persistent://prop/use/ns-abc/successTopic"; + private final String subName = "subscriptionName"; + + @BeforeMethod + public void setup() throws Exception { + pulsarTestContext = PulsarTestContext.builderForNonStartableContext().build(); + + ledgerMock = mock(ManagedLedgerImpl.class); + ManagedLedgerConfig managedLedgerConfigMock = mock(ManagedLedgerConfig.class); + doReturn(managedLedgerConfigMock).when(ledgerMock).getConfig(); + + cursorMock = mock(ManagedCursorImpl.class); + doReturn("mockCursor").when(cursorMock).getName(); + } + + @AfterMethod(alwaysRun = true) + public void teardown() throws Exception { + if (pulsarTestContext != null) { + pulsarTestContext.close(); + pulsarTestContext = null; + } + } + + @DataProvider(name = "incompatibleKeySharedPolicies") + public Object[][] incompatibleKeySharedPolicies() { + KeySharedMeta ksmSticky = new KeySharedMeta().setKeySharedMode(KeySharedMode.STICKY); + ksmSticky.addHashRange().setStart(0).setEnd(2); + + KeySharedMeta ksmStickyAllowOutOfOrder = new KeySharedMeta().setKeySharedMode(KeySharedMode.STICKY) + .setAllowOutOfOrderDelivery(true); + ksmStickyAllowOutOfOrder.addHashRange().setStart(3).setEnd(5); + + KeySharedMeta ksmAutoSplit = new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT); + KeySharedMeta ksmAutoSplitAllowOutOfOrder = new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT) + .setAllowOutOfOrderDelivery(true); + + String errorMessagePrefix = "Subscription is of different type. "; + String errorMessageSubscriptionModeSticky = errorMessagePrefix + "Active subscription key_shared mode of " + + "'STICKY' is different than the connecting consumer's key_shared mode 'AUTO_SPLIT'."; + String errorMessageSubscriptionModeAutoSplit = errorMessagePrefix + "Active subscription key_shared mode of " + + "'AUTO_SPLIT' is different than the connecting consumer's key_shared mode 'STICKY'."; + String errorMessageOutOfOrderNotAllowed = errorMessagePrefix + "Active subscription does not allow out of " + + "order delivery while the connecting consumer allows it."; + String errorMessageOutOfOrderAllowed = errorMessagePrefix + "Active subscription allows out of order delivery " + + "while the connecting consumer does not allow it."; + + return new Object[][] { + { ksmAutoSplit, ksmSticky, errorMessageSubscriptionModeAutoSplit }, + { ksmAutoSplit, ksmStickyAllowOutOfOrder, errorMessageSubscriptionModeAutoSplit }, + { ksmAutoSplit, ksmAutoSplitAllowOutOfOrder, errorMessageOutOfOrderNotAllowed }, + + { ksmAutoSplitAllowOutOfOrder, ksmSticky, errorMessageSubscriptionModeAutoSplit }, + { ksmAutoSplitAllowOutOfOrder, ksmStickyAllowOutOfOrder, errorMessageSubscriptionModeAutoSplit }, + { ksmAutoSplitAllowOutOfOrder, ksmAutoSplit, errorMessageOutOfOrderAllowed }, + + { ksmSticky, ksmStickyAllowOutOfOrder, errorMessageOutOfOrderNotAllowed }, + { ksmSticky, ksmAutoSplit, errorMessageSubscriptionModeSticky }, + { ksmSticky, ksmAutoSplitAllowOutOfOrder, errorMessageSubscriptionModeSticky }, + + { ksmStickyAllowOutOfOrder, ksmSticky, errorMessageOutOfOrderAllowed }, + { ksmStickyAllowOutOfOrder, ksmAutoSplit, errorMessageSubscriptionModeSticky }, + { ksmStickyAllowOutOfOrder, ksmAutoSplitAllowOutOfOrder, errorMessageSubscriptionModeSticky } + }; + } + + @Test(dataProvider = "incompatibleKeySharedPolicies") + public void testIncompatibleKeySharedPoliciesNotAllowedInNonPersistentSub(KeySharedMeta consumer1Ksm, KeySharedMeta consumer2Ksm, + String expectedErrorMessage) throws Exception { + NonPersistentTopic topic = new NonPersistentTopic(successTopicName, pulsarTestContext.getBrokerService()); + NonPersistentSubscription sub = new NonPersistentSubscription(topic, subName, Map.of()); + + // two consumers with incompatible key_shared policies + Consumer keySharedConsumerMock1 = createKeySharedMockConsumer("consumer-1", consumer1Ksm); + Consumer keySharedConsumerMock2 = createKeySharedMockConsumer("consumer-2", consumer2Ksm); + + // first consumer defines key_shared mode of subscription and whether out of order delivery is allowed + sub.addConsumer(keySharedConsumerMock1).get(5, TimeUnit.SECONDS); + + try { + // add second consumer with incompatible key_shared policy + sub.addConsumer(keySharedConsumerMock2).get(5, TimeUnit.SECONDS); + fail(BrokerServiceException.SubscriptionBusyException.class.getSimpleName() + " not thrown"); + } catch (Exception e) { + // subscription throws exception when consumer with incompatible key_shared policy is added + Throwable cause = e.getCause(); + assertTrue(cause instanceof BrokerServiceException.SubscriptionBusyException); + assertEquals(cause.getMessage(), expectedErrorMessage); + } + } + + @Test(dataProvider = "incompatibleKeySharedPolicies") + public void testIncompatibleKeySharedPoliciesNotAllowedInPersistentSub(KeySharedMeta consumer1Ksm, KeySharedMeta consumer2Ksm, + String expectedErrorMessage) throws Exception { + PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, pulsarTestContext.getBrokerService()); + PersistentSubscription sub = new PersistentSubscription(topic, subName, cursorMock, false); + + // two consumers with incompatible key_shared policies + Consumer keySharedConsumerMock1 = createKeySharedMockConsumer("consumer-1", consumer1Ksm); + Consumer keySharedConsumerMock2 = createKeySharedMockConsumer("consumer-2", consumer2Ksm); + + // first consumer defines key_shared mode of subscription and whether out of order delivery is allowed + sub.addConsumer(keySharedConsumerMock1).get(5, TimeUnit.SECONDS); + + try { + // add second consumer with incompatible key_shared policy + sub.addConsumer(keySharedConsumerMock2).get(5, TimeUnit.SECONDS); + fail(BrokerServiceException.SubscriptionBusyException.class.getSimpleName() + " not thrown"); + } catch (Exception e) { + // subscription throws exception when consumer with incompatible key_shared policy is added + Throwable cause = e.getCause(); + assertTrue(cause instanceof BrokerServiceException.SubscriptionBusyException); + assertEquals(cause.getMessage(), expectedErrorMessage); + } + } + + protected Consumer createKeySharedMockConsumer(String name, KeySharedMeta ksm) { + Consumer consumer = BrokerTestUtil.createMockConsumer(name); + doReturn(CommandSubscribe.SubType.Key_Shared).when(consumer).subType(); + doReturn(ksm).when(consumer).getKeySharedMeta(); + doReturn(mock(PendingAcksMap.class)).when(consumer).getPendingAcks(); + return consumer; + } + +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionTestBase.java deleted file mode 100644 index 4bede327f1872..0000000000000 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionTestBase.java +++ /dev/null @@ -1,84 +0,0 @@ -/* - * 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.mockito.Mockito.doReturn; -import static org.mockito.Mockito.mock; -import org.apache.pulsar.broker.BrokerTestUtil; -import org.apache.pulsar.common.api.proto.CommandSubscribe; -import org.apache.pulsar.common.api.proto.KeySharedMeta; -import org.apache.pulsar.common.api.proto.KeySharedMode; -import org.testng.annotations.DataProvider; - -public abstract class SubscriptionTestBase { - - protected final String successTopicName = "persistent://prop/use/ns-abc/successTopic"; - protected final String subName = "subscriptionName"; - - @DataProvider(name = "incompatibleKeySharedPolicies") - public Object[][] incompatibleKeySharedPolicies() { - KeySharedMeta ksmSticky = new KeySharedMeta().setKeySharedMode(KeySharedMode.STICKY); - ksmSticky.addHashRange().setStart(0).setEnd(2); - - KeySharedMeta ksmStickyAllowOutOfOrder = new KeySharedMeta().setKeySharedMode(KeySharedMode.STICKY) - .setAllowOutOfOrderDelivery(true); - ksmStickyAllowOutOfOrder.addHashRange().setStart(3).setEnd(5); - - KeySharedMeta ksmAutoSplit = new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT); - KeySharedMeta ksmAutoSplitAllowOutOfOrder = new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT) - .setAllowOutOfOrderDelivery(true); - - String errorMessagePrefix = "Subscription is of different type. "; - String errorMessageSubscriptionModeSticky = errorMessagePrefix + "Active subscription key_shared mode of " - + "'STICKY' is different than the connecting consumer's key_shared mode 'AUTO_SPLIT'."; - String errorMessageSubscriptionModeAutoSplit = errorMessagePrefix + "Active subscription key_shared mode of " - + "'AUTO_SPLIT' is different than the connecting consumer's key_shared mode 'STICKY'."; - String errorMessageOutOfOrderNotAllowed = errorMessagePrefix + "Active subscription does not allow out of " - + "order delivery while the connecting consumer allows it."; - String errorMessageOutOfOrderAllowed = errorMessagePrefix + "Active subscription allows out of order delivery " - + "while the connecting consumer does not allow it."; - - return new Object[][] { - { ksmAutoSplit, ksmSticky, errorMessageSubscriptionModeAutoSplit }, - { ksmAutoSplit, ksmStickyAllowOutOfOrder, errorMessageSubscriptionModeAutoSplit }, - { ksmAutoSplit, ksmAutoSplitAllowOutOfOrder, errorMessageOutOfOrderNotAllowed }, - - { ksmAutoSplitAllowOutOfOrder, ksmSticky, errorMessageSubscriptionModeAutoSplit }, - { ksmAutoSplitAllowOutOfOrder, ksmStickyAllowOutOfOrder, errorMessageSubscriptionModeAutoSplit }, - { ksmAutoSplitAllowOutOfOrder, ksmAutoSplit, errorMessageOutOfOrderAllowed }, - - { ksmSticky, ksmStickyAllowOutOfOrder, errorMessageOutOfOrderNotAllowed }, - { ksmSticky, ksmAutoSplit, errorMessageSubscriptionModeSticky }, - { ksmSticky, ksmAutoSplitAllowOutOfOrder, errorMessageSubscriptionModeSticky }, - - { ksmStickyAllowOutOfOrder, ksmSticky, errorMessageOutOfOrderAllowed }, - { ksmStickyAllowOutOfOrder, ksmAutoSplit, errorMessageSubscriptionModeSticky }, - { ksmStickyAllowOutOfOrder, ksmAutoSplitAllowOutOfOrder, errorMessageSubscriptionModeSticky } - }; - } - - protected Consumer createKeySharedMockConsumer(String name, KeySharedMeta ksm) { - Consumer consumer = BrokerTestUtil.createMockConsumer(name); - doReturn(CommandSubscribe.SubType.Key_Shared).when(consumer).subType(); - doReturn(ksm).when(consumer).getKeySharedMeta(); - doReturn(mock(PendingAcksMap.class)).when(consumer).getPendingAcks(); - return consumer; - } - -} \ No newline at end of file diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscriptionTest.java deleted file mode 100644 index 6ccecef2d3223..0000000000000 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscriptionTest.java +++ /dev/null @@ -1,64 +0,0 @@ -/* - * 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.nonpersistent; - -import static org.testng.Assert.assertEquals; -import static org.testng.Assert.assertTrue; -import static org.testng.Assert.fail; -import java.util.Map; -import java.util.concurrent.TimeUnit; -import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionBusyException; -import org.apache.pulsar.broker.service.Consumer; -import org.apache.pulsar.broker.service.SubscriptionTestBase; -import org.apache.pulsar.broker.testcontext.PulsarTestContext; -import org.apache.pulsar.common.api.proto.KeySharedMeta; -import org.testng.annotations.Test; - -@Test(groups = "broker") -public class NonPersistentSubscriptionTest extends SubscriptionTestBase { - - @Test(dataProvider = "incompatibleKeySharedPolicies") - public void testIncompatibleKeySharedPoliciesNotAllowed(KeySharedMeta consumer1Ksm, KeySharedMeta consumer2Ksm, - String expectedErrorMessage) throws Exception { - PulsarTestContext context = PulsarTestContext.builderForNonStartableContext().build(); - NonPersistentTopic topic = new NonPersistentTopic(successTopicName, context.getBrokerService()); - NonPersistentSubscription sub = new NonPersistentSubscription(topic, subName, Map.of()); - - // two consumers with incompatible key_shared policies - Consumer keySharedConsumerMock1 = createKeySharedMockConsumer("consumer-1", consumer1Ksm); - Consumer keySharedConsumerMock2 = createKeySharedMockConsumer("consumer-2", consumer2Ksm); - - // first consumer defines key_shared mode of subscription and whether out of order delivery is allowed - sub.addConsumer(keySharedConsumerMock1).get(5, TimeUnit.SECONDS); - - try { - // add second consumer with incompatible key_shared policy - sub.addConsumer(keySharedConsumerMock2).get(5, TimeUnit.SECONDS); - fail(SubscriptionBusyException.class.getSimpleName() + " not thrown"); - } catch (Exception e) { - // subscription throws exception when consumer with incompatible key_shared policy is added - Throwable cause = e.getCause(); - assertTrue(cause instanceof SubscriptionBusyException); - assertEquals(cause.getMessage(), expectedErrorMessage); - } - - context.close(); - } - -} \ No newline at end of file diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentSubscriptionTest.java index 53e78c5ad173d..9a46a2919d12a 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentSubscriptionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentSubscriptionTest.java @@ -34,7 +34,6 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; -import java.util.concurrent.TimeUnit; import org.apache.bookkeeper.mledger.AsyncCallbacks; import org.apache.bookkeeper.mledger.ManagedLedger; import org.apache.bookkeeper.mledger.ManagedLedgerConfig; @@ -45,9 +44,7 @@ import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.commons.lang3.tuple.MutablePair; import org.apache.pulsar.broker.resources.NamespaceResources; -import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionBusyException; import org.apache.pulsar.broker.service.Consumer; -import org.apache.pulsar.broker.service.SubscriptionTestBase; import org.apache.pulsar.broker.testcontext.PulsarTestContext; import org.apache.pulsar.broker.transaction.buffer.impl.InMemTransactionBufferProvider; import org.apache.pulsar.broker.transaction.pendingack.PendingAckStore; @@ -57,7 +54,6 @@ import org.apache.pulsar.client.api.transaction.TxnID; import org.apache.pulsar.common.api.proto.CommandAck.AckType; import org.apache.pulsar.common.api.proto.CommandSubscribe; -import org.apache.pulsar.common.api.proto.KeySharedMeta; import org.apache.pulsar.common.api.proto.TxnAction; import org.apache.pulsar.common.policies.data.Policies; import org.apache.pulsar.transaction.common.exception.TransactionConflictException; @@ -67,7 +63,7 @@ import org.testng.annotations.Test; @Test(groups = "broker") -public class PersistentSubscriptionTest extends SubscriptionTestBase { +public class PersistentSubscriptionTest { private PulsarTestContext pulsarTestContext; private ManagedLedger ledgerMock; @@ -77,6 +73,9 @@ public class PersistentSubscriptionTest extends SubscriptionTestBase { private Consumer consumerMock; private ManagedLedgerConfig managedLedgerConfigMock; + final String successTopicName = "persistent://prop/use/ns-abc/successTopic"; + final String subName = "subscriptionName"; + final TxnID txnID1 = new TxnID(1,1); final TxnID txnID2 = new TxnID(1,2); @@ -227,34 +226,6 @@ public void testAcknowledgeUpdateCursorLastActive() throws Exception { assertTrue(persistentSubscription.cursor.getLastActive() > beforeAcknowledgeTimestamp); } - @Test(dataProvider = "incompatibleKeySharedPolicies") - public void testIncompatibleKeySharedPoliciesNotAllowed(KeySharedMeta consumer1Ksm, KeySharedMeta consumer2Ksm, - String expectedErrorMessage) throws Exception { - PulsarTestContext context = PulsarTestContext.builderForNonStartableContext().build(); - PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, context.getBrokerService()); - PersistentSubscription sub = new PersistentSubscription(topic, subName, cursorMock, false); - - // two consumers with incompatible key_shared policies - Consumer keySharedConsumerMock1 = createKeySharedMockConsumer("consumer-1", consumer1Ksm); - Consumer keySharedConsumerMock2 = createKeySharedMockConsumer("consumer-2", consumer2Ksm); - - // first consumer defines key_shared mode of subscription and whether out of order delivery is allowed - sub.addConsumer(keySharedConsumerMock1).get(5, TimeUnit.SECONDS); - - try { - // add second consumer with incompatible key_shared policy - sub.addConsumer(keySharedConsumerMock2).get(5, TimeUnit.SECONDS); - fail(SubscriptionBusyException.class.getSimpleName() + " not thrown"); - } catch (Exception e) { - // subscription throws exception when consumer with incompatible key_shared policy is added - Throwable cause = e.getCause(); - assertTrue(cause instanceof SubscriptionBusyException); - assertEquals(cause.getMessage(), expectedErrorMessage); - } - - context.close(); - } - public static class CustomTransactionPendingAckStoreProvider implements TransactionPendingAckStoreProvider { @Override public CompletableFuture newPendingAckStore(PersistentSubscription subscription) {