diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java index a8b2e0b24ebf5..c2d451deaaf13 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java @@ -196,6 +196,19 @@ public ChannelPromise sendMessages(final List entries, EntryBatchSizes ba return writePromise; } + // Note + // Must ensure that the message is written to the pendingAcks before sent is first , because this consumer + // is possible to disconnect at this time. + if (pendingAcks != null) { + for (int i = 0; i < entries.size(); i++) { + Entry entry = entries.get(i); + if (entry != null) { + int batchSize = batchSizes.getBatchSize(i); + pendingAcks.put(entry.getLedgerId(), entry.getEntryId(), batchSize, 0); + } + } + } + // reduce permit and increment unackedMsg count with total number of messages in batch-msgs MESSAGE_PERMITS_UPDATER.addAndGet(this, -totalMessages); incrementUnackedMessages(totalMessages); @@ -211,10 +224,6 @@ public ChannelPromise sendMessages(final List entries, EntryBatchSizes ba int batchSize = batchSizes.getBatchSize(i); - if (pendingAcks != null) { - pendingAcks.put(entry.getLedgerId(), entry.getEntryId(), batchSize, 0); - } - if (batchSize > 1 && !cnx.isBatchMessageCompatibleVersion()) { log.warn("[{}-{}] Consumer doesn't support batch messages - consumerId {}, msg id {}-{}", topicName, subscription, diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HashRangeStickyKeyConsumerSelector.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HashRangeStickyKeyConsumerSelector.java index 18b342799c116..940798fa47595 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HashRangeStickyKeyConsumerSelector.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HashRangeStickyKeyConsumerSelector.java @@ -21,6 +21,7 @@ import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerAssignException; import org.apache.pulsar.common.util.Murmur3_32Hash; +import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.Map.Entry; @@ -89,7 +90,7 @@ public synchronized void addConsumer(Consumer consumer) throws ConsumerAssignExc @Override public synchronized void removeConsumer(Consumer consumer) { - Integer removeRange = consumerRange.get(consumer); + Integer removeRange = consumerRange.remove(consumer); if (removeRange != null) { if (removeRange == rangeSize && rangeMap.size() > 1) { Map.Entry lowerEntry = rangeMap.lowerEntry(removeRange); @@ -98,7 +99,6 @@ public synchronized void removeConsumer(Consumer consumer) { consumerRange.put(lowerEntry.getValue(), removeRange); } else { rangeMap.remove(removeRange); - consumerRange.remove(consumer); } } } @@ -147,4 +147,12 @@ private boolean is2Power(int num) { if(num < 2) return false; return (num & num - 1) == 0; } + + Map getConsumerRange() { + return Collections.unmodifiableMap(consumerRange); + } + + Map getRangeConsumer() { + return Collections.unmodifiableMap(rangeMap); + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/HashRangeStickyKeyConsumerSelectorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/HashRangeStickyKeyConsumerSelectorTest.java index ff67c2036aeef..0bc5fb392df04 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/HashRangeStickyKeyConsumerSelectorTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/HashRangeStickyKeyConsumerSelectorTest.java @@ -33,7 +33,7 @@ public class HashRangeStickyKeyConsumerSelectorTest { @Test public void testConsumerSelect() throws ConsumerAssignException { - StickyKeyConsumerSelector selector = new HashRangeStickyKeyConsumerSelector(); + HashRangeStickyKeyConsumerSelector selector = new HashRangeStickyKeyConsumerSelector(); String key1 = "anyKey"; Assert.assertNull(selector.select(key1.getBytes())); @@ -41,9 +41,13 @@ public void testConsumerSelect() throws ConsumerAssignException { selector.addConsumer(consumer1); int consumer1Slot = DEFAULT_RANGE_SIZE; Assert.assertEquals(selector.select(key1.getBytes()), consumer1); + Assert.assertEquals(selector.getConsumerRange().size(), 1); + Assert.assertEquals(selector.getRangeConsumer().size(), 1); Consumer consumer2 = mock(Consumer.class); selector.addConsumer(consumer2); + Assert.assertEquals(selector.getConsumerRange().size(), 2); + Assert.assertEquals(selector.getRangeConsumer().size(), 2); int consumer2Slot = consumer1Slot >> 1; for (int i = 0; i < 100; i++) { @@ -58,6 +62,8 @@ public void testConsumerSelect() throws ConsumerAssignException { Consumer consumer3 = mock(Consumer.class); selector.addConsumer(consumer3); + Assert.assertEquals(selector.getConsumerRange().size(), 3); + Assert.assertEquals(selector.getRangeConsumer().size(), 3); int consumer3Slot = consumer2Slot >> 1; for (int i = 0; i < 100; i++) { @@ -74,6 +80,8 @@ public void testConsumerSelect() throws ConsumerAssignException { Consumer consumer4 = mock(Consumer.class); selector.addConsumer(consumer4); + Assert.assertEquals(selector.getConsumerRange().size(), 4); + Assert.assertEquals(selector.getRangeConsumer().size(), 4); int consumer4Slot = consumer1Slot - ((consumer1Slot - consumer2Slot) >> 1); for (int i = 0; i < 100; i++) { @@ -91,6 +99,8 @@ public void testConsumerSelect() throws ConsumerAssignException { } selector.removeConsumer(consumer1); + Assert.assertEquals(selector.getConsumerRange().size(), 3); + Assert.assertEquals(selector.getRangeConsumer().size(), 3); for (int i = 0; i < 100; i++) { String key = UUID.randomUUID().toString(); int slot = Murmur3_32Hash.getInstance().makeHash(key.getBytes()) % DEFAULT_RANGE_SIZE; @@ -104,6 +114,8 @@ public void testConsumerSelect() throws ConsumerAssignException { } selector.removeConsumer(consumer2); + Assert.assertEquals(selector.getConsumerRange().size(), 2); + Assert.assertEquals(selector.getRangeConsumer().size(), 2); for (int i = 0; i < 100; i++) { String key = UUID.randomUUID().toString(); int slot = Murmur3_32Hash.getInstance().makeHash(key.getBytes()) % DEFAULT_RANGE_SIZE; @@ -115,6 +127,8 @@ public void testConsumerSelect() throws ConsumerAssignException { } selector.removeConsumer(consumer3); + Assert.assertEquals(selector.getConsumerRange().size(), 1); + Assert.assertEquals(selector.getRangeConsumer().size(), 1); for (int i = 0; i < 100; i++) { String key = UUID.randomUUID().toString(); Assert.assertEquals(selector.select(key.getBytes()), consumer4); diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/messaging/MessagingBase.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/messaging/MessagingBase.java new file mode 100644 index 0000000000000..0043dc2f5178d --- /dev/null +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/messaging/MessagingBase.java @@ -0,0 +1,164 @@ +/** + * 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.tests.integration.messaging; + +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertTrue; +import static org.testng.AssertJUnit.assertEquals; + +import com.google.common.collect.Maps; +import com.google.common.collect.Sets; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.tests.integration.suites.PulsarTestSuite; +import org.testng.annotations.BeforeMethod; + +import java.lang.reflect.Method; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +@Slf4j +public abstract class MessagingBase extends PulsarTestSuite { + + protected String methodName; + + @BeforeMethod + public void beforeMethod(Method m) throws Exception { + methodName = m.getName(); + } + + protected String getNonPartitionedTopic(String topicPrefix, boolean isPersistent) throws Exception { + String nsName = generateNamespaceName(); + pulsarCluster.createNamespace(nsName); + + return generateTopicName(nsName, topicPrefix, true); + } + + protected String getPartitionedTopic(String topicPrefix, boolean isPersistent, int partitions) throws Exception { + assertTrue(partitions > 0, "partitions must greater than 1"); + String nsName = generateNamespaceName(); + pulsarCluster.createNamespace(nsName); + + String topicName = generateTopicName(nsName, topicPrefix, true); + pulsarCluster.createPartitionedTopic(topicName, partitions); + return topicName; + } + + protected > void receiveMessagesCheckOrderAndDuplicate + (List> consumerList, int messagesToReceive) throws PulsarClientException { + Set messagesReceived = Sets.newHashSet(); + for (Consumer consumer : consumerList) { + Message currentReceived = null; + Message lastReceived = null; + while (true) { + try { + currentReceived = consumer.receive(3, TimeUnit.SECONDS); + } catch (PulsarClientException e) { + log.info("no more messages to receive for consumer {}", consumer.getConsumerName()); + break; + } + // Make sure that messages are received in order + if (lastReceived == null) { + assertNotNull(currentReceived); + } else { + assertTrue(currentReceived != null + && (currentReceived.getValue().compareTo(lastReceived.getValue()) > 0), + "Received messages are not in order."); + } + lastReceived = currentReceived; + // Make sure that there are no duplicates + assertTrue(messagesReceived.add(currentReceived.getValue()), + "Received duplicate message " + currentReceived.getValue()); + } + if (currentReceived != null) { + consumer.acknowledgeCumulative(currentReceived); + } + } + assertEquals(messagesReceived.size(), messagesToReceive); + } + + protected void receiveMessagesCheckDuplicate + (List> consumerList, int messagesToReceive) throws PulsarClientException { + Set messagesReceived = Sets.newHashSet(); + for (Consumer consumer : consumerList) { + Message currentReceived = null; + while (true) { + try { + currentReceived = consumer.receive(3, TimeUnit.SECONDS); + } catch (PulsarClientException e) { + log.info("no more messages to receive for consumer {}", consumer.getConsumerName()); + break; + } + // Make sure that there are no duplicates + assertTrue(messagesReceived.add(currentReceived.getValue()), + "Received duplicate message " + currentReceived.getValue()); + } + if (currentReceived != null) { + consumer.acknowledgeCumulative(currentReceived); + } + } + assertEquals(messagesReceived.size(), messagesToReceive); + } + + protected void receiveMessagesCheckStickyKeyAndDuplicate + (List> consumerList, int messagesToReceive) throws PulsarClientException { + Map> consumerKeys = Maps.newHashMap(); + Set messagesReceived = Sets.newHashSet(); + for (Consumer consumer : consumerList) { + Message currentReceived = null; + while (true) { + try { + currentReceived = consumer.receive(3, TimeUnit.SECONDS); + } catch (PulsarClientException e) { + log.info("no more messages to receive for consumer {}", consumer.getConsumerName()); + break; + } + assertNotNull(currentReceived.getKey()); + consumerKeys.putIfAbsent(consumer.getConsumerName(), Sets.newHashSet()); + consumerKeys.get(consumer.getConsumerName()).add(currentReceived.getKey()); + // Make sure that there are no duplicates + assertTrue(messagesReceived.add(currentReceived.getValue()), + "Received duplicate message " + currentReceived.getValue()); + } + if (currentReceived != null) { + consumer.acknowledgeCumulative(currentReceived); + } + } + // Make sure key will not be distributed to multiple consumers + Set allKeys = Sets.newHashSet(); + consumerKeys.forEach((k, v) -> v.forEach(key -> { + assertTrue(allKeys.add(key), + "Key "+ key + "is distributed to multiple consumers" ); + })); + assertEquals(messagesReceived.size(), messagesToReceive); + } + + protected void closeConsumers(List> consumerList) throws PulsarClientException { + Iterator> iterator = consumerList.iterator(); + while (iterator.hasNext()) { + iterator.next().close(); + iterator.remove(); + } + } +} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/messaging/NonPersistentTopicMessagingTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/messaging/NonPersistentTopicMessagingTest.java new file mode 100644 index 0000000000000..7fb1fda6ce23e --- /dev/null +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/messaging/NonPersistentTopicMessagingTest.java @@ -0,0 +1,66 @@ +/** + * 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.tests.integration.messaging; + +import lombok.extern.slf4j.Slf4j; +import org.testng.annotations.Test; + +@Slf4j +public class NonPersistentTopicMessagingTest extends TopicMessagingBase { + + @Test(dataProvider = "ServiceUrls") + public void testNonPartitionedTopicMessagingWithExclusive(String serviceUrl) throws Exception { + nonPartitionedTopicSendAndReceiveWithExclusive(serviceUrl, false); + } + + @Test(dataProvider = "ServiceUrls") + public void testPartitionedTopicMessagingWithExclusive(String serviceUrl) throws Exception { + partitionedTopicSendAndReceiveWithExclusive(serviceUrl, false); + } + + @Test(dataProvider = "ServiceUrls") + public void testNonPartitionedTopicMessagingWithFailover(String serviceUrl) throws Exception { + nonPartitionedTopicSendAndReceiveWithFailover(serviceUrl, false); + } + + @Test(dataProvider = "ServiceUrls") + public void testPartitionedTopicMessagingWithFailover(String serviceUrl) throws Exception { + partitionedTopicSendAndReceiveWithFailover(serviceUrl, false); + } + + @Test(dataProvider = "ServiceUrls") + public void testNonPartitionedTopicMessagingWithShared(String serviceUrl) throws Exception { + nonPartitionedTopicSendAndReceiveWithShared(serviceUrl, false); + } + + @Test(dataProvider = "ServiceUrls") + public void testPartitionedTopicMessagingWithShared(String serviceUrl) throws Exception { + partitionedTopicSendAndReceiveWithShared(serviceUrl, false); + } + + @Test(dataProvider = "ServiceUrls") + public void testNonPartitionedTopicMessagingWithKeyShared(String serviceUrl) throws Exception { + nonPartitionedTopicSendAndReceiveWithKeyShared(serviceUrl, true); + } + + @Test(dataProvider = "ServiceUrls") + public void testPartitionedTopicMessagingWithKeyShared(String serviceUrl) throws Exception { + partitionedTopicSendAndReceiveWithKeyShared(serviceUrl, true); + } +} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/messaging/PersistentTopicMessagingTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/messaging/PersistentTopicMessagingTest.java new file mode 100644 index 0000000000000..39b37ac61e8f7 --- /dev/null +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/messaging/PersistentTopicMessagingTest.java @@ -0,0 +1,66 @@ +/** + * 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.tests.integration.messaging; + +import lombok.extern.slf4j.Slf4j; +import org.testng.annotations.Test; + +@Slf4j +public class PersistentTopicMessagingTest extends TopicMessagingBase { + + @Test(dataProvider = "ServiceUrls") + public void testNonPartitionedTopicMessagingWithExclusive(String serviceUrl) throws Exception { + nonPartitionedTopicSendAndReceiveWithExclusive(serviceUrl, true); + } + + @Test(dataProvider = "ServiceUrls") + public void testPartitionedTopicMessagingWithExclusive(String serviceUrl) throws Exception { + partitionedTopicSendAndReceiveWithExclusive(serviceUrl, true); + } + + @Test(dataProvider = "ServiceUrls") + public void testNonPartitionedTopicMessagingWithFailover(String serviceUrl) throws Exception { + nonPartitionedTopicSendAndReceiveWithFailover(serviceUrl, true); + } + + @Test(dataProvider = "ServiceUrls") + public void testPartitionedTopicMessagingWithFailover(String serviceUrl) throws Exception { + partitionedTopicSendAndReceiveWithFailover(serviceUrl, true); + } + + @Test(dataProvider = "ServiceUrls") + public void testNonPartitionedTopicMessagingWithShared(String serviceUrl) throws Exception { + nonPartitionedTopicSendAndReceiveWithShared(serviceUrl, true); + } + + @Test(dataProvider = "ServiceUrls") + public void testPartitionedTopicMessagingWithShared(String serviceUrl) throws Exception { + partitionedTopicSendAndReceiveWithShared(serviceUrl, true); + } + + @Test(dataProvider = "ServiceUrls") + public void testNonPartitionedTopicMessagingWithKeyShared(String serviceUrl) throws Exception { + nonPartitionedTopicSendAndReceiveWithKeyShared(serviceUrl, true); + } + + @Test(dataProvider = "ServiceUrls") + public void testPartitionedTopicMessagingWithKeyShared(String serviceUrl) throws Exception { + partitionedTopicSendAndReceiveWithKeyShared(serviceUrl, true); + } +} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/messaging/TopicMessagingBase.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/messaging/TopicMessagingBase.java new file mode 100644 index 0000000000000..26097cf8a04da --- /dev/null +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/messaging/TopicMessagingBase.java @@ -0,0 +1,491 @@ +/** + * 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.tests.integration.messaging; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertNull; +import static org.testng.Assert.assertTrue; + +import lombok.Cleanup; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.SubscriptionType; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.UUID; + +@Slf4j +public class TopicMessagingBase extends MessagingBase { + + protected void nonPartitionedTopicSendAndReceiveWithExclusive(String serviceUrl, boolean isPersistent) throws Exception { + log.info("-- Starting {} test --", methodName); + final String topicName = getNonPartitionedTopic("test-non-partitioned-consume-exclusive", isPersistent); + @Cleanup + final PulsarClient client = PulsarClient.builder() + .serviceUrl(serviceUrl) + .build(); + @Cleanup + final Consumer consumer = client.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName("test-sub") + .subscriptionType(SubscriptionType.Exclusive) + .subscribe(); + Exception exception = null; + try { + client.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName("test-sub") + .subscriptionType(SubscriptionType.Exclusive) + .subscribe(); + } catch (PulsarClientException e) { + exception = e; + } + assertNotNull(exception); + final int messagesToSend = 10; + final String producerName = "producerForExclusive"; + @Cleanup + final Producer producer = client.newProducer(Schema.STRING) + .topic(topicName) + .enableBatching(false) + .producerName(producerName) + .create(); + for (int i = 0; i < messagesToSend; i++) { + MessageId messageId = producer.newMessage().value(producer.getProducerName() + "-" + i).send(); + assertNotNull(messageId); + } + log.info("public messages complete."); + receiveMessagesCheckOrderAndDuplicate(Collections.singletonList(consumer), messagesToSend); + log.info("-- Exiting {} test --", methodName); + } + + protected void partitionedTopicSendAndReceiveWithExclusive(String serviceUrl, boolean isPersistent) throws Exception { + log.info("-- Starting {} test --", methodName); + final int partitions = 3; + String topicName = getPartitionedTopic("test-partitioned-consume-exclusive", isPersistent, partitions); + @Cleanup + final PulsarClient client = PulsarClient.builder() + .serviceUrl(serviceUrl) + .build(); + List> consumerList = new ArrayList<>(3); + for (int i = 0; i < partitions; i++) { + Consumer consumer = client.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName("test-sub") + .subscriptionType(SubscriptionType.Exclusive) + .subscribe(); + consumerList.add(consumer); + } + assertEquals(partitions, consumerList.size()); + Exception exception = null; + try { + client.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName("test-sub") + .subscriptionType(SubscriptionType.Exclusive) + .subscribe(); + } catch (PulsarClientException e) { + exception = e; + } + assertNotNull(exception); + final int messagesToSend = 10; + final String producerName = "producerForExclusive"; + @Cleanup + final Producer producer = client.newProducer(Schema.STRING) + .topic(topicName) + .enableBatching(false) + .producerName(producerName) + .create(); + for (int i = 0; i < messagesToSend; i++) { + MessageId messageId = producer.newMessage().value(producer.getProducerName() + "-" + i).send(); + assertNotNull(messageId); + } + log.info("public messages complete."); + receiveMessagesCheckOrderAndDuplicate(consumerList, messagesToSend); + // To simulate a consumer crashed + Consumer crashedConsumer = consumerList.remove(0); + crashedConsumer.close(); + for (int i = 0; i < messagesToSend; i++) { + MessageId messageId = producer.newMessage().value(producer.getProducerName() + "-" + i).send(); + assertNotNull(messageId); + } + receiveMessagesCheckOrderAndDuplicate(consumerList, messagesToSend); + closeConsumers(consumerList); + log.info("-- Exiting {} test --", methodName); + } + + protected void nonPartitionedTopicSendAndReceiveWithFailover(String serviceUrl, boolean isPersistent) throws Exception { + log.info("-- Starting {} test --", methodName); + final String topicName = getNonPartitionedTopic("test-non-partitioned-consume-failover", isPersistent); + @Cleanup + final PulsarClient client = PulsarClient.builder() + .serviceUrl(serviceUrl) + .build(); + List> consumerList = new ArrayList<>(3); + final Consumer consumer = client.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName("test-sub") + .subscriptionType(SubscriptionType.Failover) + .subscribe(); + consumerList.add(consumer); + Exception exception = null; + Consumer standbyConsumer = null; + try { + standbyConsumer = client.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName("test-sub") + .subscriptionType(SubscriptionType.Failover) + .subscribe(); + } catch (PulsarClientException e) { + exception = e; + } + assertNull(exception); + assertNotNull(standbyConsumer); + assertTrue(standbyConsumer.isConnected()); + final int messagesToSend = 10; + final String producerName = "producerForFailover"; + @Cleanup + final Producer producer = client.newProducer(Schema.STRING) + .topic(topicName) + .enableBatching(false) + .producerName(producerName) + .create(); + for (int i = 0; i < messagesToSend; i++) { + MessageId messageId = producer.newMessage().value(producer.getProducerName() + "-" + i).send(); + assertNotNull(messageId); + } + log.info("public messages complete."); + receiveMessagesCheckOrderAndDuplicate(consumerList, messagesToSend); + // To simulate a consumer crashed + Consumer crashedConsumer = consumerList.remove(0); + crashedConsumer.close(); + consumerList.add(standbyConsumer); + for (int i = 0; i < messagesToSend; i++) { + MessageId messageId = producer.newMessage().value(producer.getProducerName() + "-" + i).send(); + assertNotNull(messageId); + } + receiveMessagesCheckOrderAndDuplicate(consumerList, messagesToSend); + closeConsumers(consumerList); + log.info("-- Exiting {} test --", methodName); + } + + protected void partitionedTopicSendAndReceiveWithFailover(String serviceUrl, boolean isPersistent) throws Exception { + log.info("-- Starting {} test --", methodName); + final int partitions = 3; + String topicName = getPartitionedTopic("test-partitioned-consume-failover", isPersistent, partitions); + @Cleanup + final PulsarClient client = PulsarClient.builder() + .serviceUrl(serviceUrl) + .build(); + List> consumerList = new ArrayList<>(3); + for (int i = 0; i < partitions; i++) { + Consumer consumer = client.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName("test-sub") + .subscriptionType(SubscriptionType.Failover) + .subscribe(); + consumerList.add(consumer); + } + assertEquals(partitions, consumerList.size()); + Exception exception = null; + Consumer standbyConsumer = null; + try { + standbyConsumer = client.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName("test-sub") + .subscriptionType(SubscriptionType.Failover) + .subscribe(); + } catch (PulsarClientException e) { + exception = e; + } + assertNull(exception); + assertNotNull(standbyConsumer); + assertTrue(standbyConsumer.isConnected()); + final int messagesToSend = 10; + final String producerName = "producerForFailover"; + @Cleanup + final Producer producer = client.newProducer(Schema.STRING) + .topic(topicName) + .enableBatching(false) + .producerName(producerName) + .create(); + for (int i = 0; i < messagesToSend; i++) { + MessageId messageId = producer.newMessage().value(producer.getProducerName() + "-" + i).send(); + assertNotNull(messageId); + } + log.info("public messages complete."); + receiveMessagesCheckOrderAndDuplicate(consumerList, messagesToSend); + // To simulate a consumer crashed + Consumer crashedConsumer = consumerList.remove(0); + crashedConsumer.close(); + consumerList.add(standbyConsumer); + for (int i = 0; i < messagesToSend; i++) { + MessageId messageId = producer.newMessage().value(producer.getProducerName() + "-" + i).send(); + assertNotNull(messageId); + } + receiveMessagesCheckOrderAndDuplicate(consumerList, messagesToSend); + closeConsumers(consumerList); + log.info("-- Exiting {} test --", methodName); + } + + protected void nonPartitionedTopicSendAndReceiveWithShared(String serviceUrl, boolean isPersistent) throws Exception { + log.info("-- Starting {} test --", methodName); + final String topicName = getNonPartitionedTopic("test-non-partitioned-consume-shared", isPersistent); + @Cleanup + final PulsarClient client = PulsarClient.builder() + .serviceUrl(serviceUrl) + .build(); + List> consumerList = new ArrayList<>(2); + final Consumer consumer = client.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName("test-sub") + .subscriptionType(SubscriptionType.Shared) + .subscribe(); + consumerList.add(consumer); + Exception exception = null; + Consumer moreConsumer = null; + try { + moreConsumer = client.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName("test-sub") + .subscriptionType(SubscriptionType.Shared) + .subscribe(); + } catch (PulsarClientException e) { + exception = e; + } + assertNull(exception); + assertNotNull(moreConsumer); + assertTrue(moreConsumer.isConnected()); + consumerList.add(moreConsumer); + final int messagesToSend = 10; + final String producerName = "producerForShared"; + @Cleanup + final Producer producer = client.newProducer(Schema.STRING) + .topic(topicName) + .enableBatching(false) + .producerName(producerName) + .create(); + for (int i = 0; i < messagesToSend; i++) { + MessageId messageId = producer.newMessage().value(producer.getProducerName() + "-" + i).send(); + assertNotNull(messageId); + } + log.info("public messages complete."); + receiveMessagesCheckDuplicate(consumerList, messagesToSend); + // To simulate a consumer crashed + Consumer crashedConsumer = consumerList.remove(0); + crashedConsumer.close(); + for (int i = 0; i < messagesToSend; i++) { + MessageId messageId = producer.newMessage().value(producer.getProducerName() + "-" + i).send(); + assertNotNull(messageId); + } + receiveMessagesCheckDuplicate(consumerList, messagesToSend); + closeConsumers(consumerList); + log.info("-- Exiting {} test --", methodName); + } + + protected void partitionedTopicSendAndReceiveWithShared(String serviceUrl, boolean isPersistent) throws Exception { + log.info("-- Starting {} test --", methodName); + final int partitions = 3; + String topicName = getPartitionedTopic("test-partitioned-consume-shared", isPersistent, partitions); + @Cleanup + final PulsarClient client = PulsarClient.builder() + .serviceUrl(serviceUrl) + .build(); + List> consumerList = new ArrayList<>(3); + for (int i = 0; i < partitions; i++) { + Consumer consumer = client.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName("test-sub") + .subscriptionType(SubscriptionType.Shared) + .subscribe(); + consumerList.add(consumer); + } + assertEquals(partitions, consumerList.size()); + Exception exception = null; + Consumer moreConsumer = null; + try { + moreConsumer = client.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName("test-sub") + .subscriptionType(SubscriptionType.Shared) + .subscribe(); + } catch (PulsarClientException e) { + exception = e; + } + assertNull(exception); + assertNotNull(moreConsumer); + assertTrue(moreConsumer.isConnected()); + final int messagesToSend = 10; + final String producerName = "producerForFailover"; + @Cleanup + final Producer producer = client.newProducer(Schema.STRING) + .topic(topicName) + .enableBatching(false) + .producerName(producerName) + .create(); + for (int i = 0; i < messagesToSend; i++) { + MessageId messageId = producer.newMessage().value(producer.getProducerName() + "-" + i).send(); + assertNotNull(messageId); + } + log.info("public messages complete."); + receiveMessagesCheckDuplicate(consumerList, messagesToSend); + // To simulate a consumer crashed + Consumer crashedConsumer = consumerList.remove(0); + crashedConsumer.close(); + for (int i = 0; i < messagesToSend; i++) { + MessageId messageId = producer.newMessage().value(producer.getProducerName() + "-" + i).send(); + assertNotNull(messageId); + } + receiveMessagesCheckDuplicate(consumerList, messagesToSend); + closeConsumers(consumerList); + log.info("-- Exiting {} test --", methodName); + } + + protected void nonPartitionedTopicSendAndReceiveWithKeyShared(String serviceUrl, boolean isPersistent) throws Exception { + log.info("-- Starting {} test --", methodName); + final String topicName = getNonPartitionedTopic("test-non-partitioned-consume-key-shared", isPersistent); + @Cleanup + final PulsarClient client = PulsarClient.builder() + .serviceUrl(serviceUrl) + .build(); + List> consumerList = new ArrayList<>(4); + final Consumer consumer = client.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName("test-sub") + .subscriptionType(SubscriptionType.Key_Shared) + .subscribe(); + consumerList.add(consumer); + Exception exception = null; + Consumer moreConsumer = null; + try { + moreConsumer = client.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName("test-sub") + .subscriptionType(SubscriptionType.Key_Shared) + .subscribe(); + } catch (PulsarClientException e) { + exception = e; + } + assertNull(exception); + assertNotNull(moreConsumer); + assertTrue(moreConsumer.isConnected()); + consumerList.add(moreConsumer); + final int messagesToSend = 10; + final String producerName = "producerForKeyShared"; + @Cleanup + final Producer producer = client.newProducer(Schema.STRING) + .topic(topicName) + .enableBatching(false) + .producerName(producerName) + .create(); + for (int i = 0; i < messagesToSend; i++) { + MessageId messageId = producer.newMessage() + .key(UUID.randomUUID().toString()) + .value(producer.getProducerName() + "-" + i) + .send(); + assertNotNull(messageId); + } + log.info("public messages complete."); + receiveMessagesCheckStickyKeyAndDuplicate(consumerList, messagesToSend); + // To simulate a consumer crashed + Consumer crashedConsumer = consumerList.remove(0); + crashedConsumer.close(); + for (int i = 0; i < messagesToSend; i++) { + MessageId messageId = producer.newMessage() + .key(UUID.randomUUID().toString()) + .value(producer.getProducerName() + "-" + i) + .send(); + assertNotNull(messageId); + } + receiveMessagesCheckStickyKeyAndDuplicate(consumerList, messagesToSend); + closeConsumers(consumerList); + log.info("-- Exiting {} test --", methodName); + } + + protected void partitionedTopicSendAndReceiveWithKeyShared(String serviceUrl, boolean isPersistent) throws Exception { + log.info("-- Starting {} test --", methodName); + final int partitions = 3; + String topicName = getPartitionedTopic("test-partitioned-consume-key-shared", isPersistent, partitions); + @Cleanup + final PulsarClient client = PulsarClient.builder() + .serviceUrl(serviceUrl) + .build(); + List> consumerList = new ArrayList<>(3); + for (int i = 0; i < partitions; i++) { + Consumer consumer = client.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName("test-sub") + .subscriptionType(SubscriptionType.Key_Shared) + .subscribe(); + consumerList.add(consumer); + } + assertEquals(partitions, consumerList.size()); + Exception exception = null; + Consumer moreConsumer = null; + try { + moreConsumer = client.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName("test-sub") + .subscriptionType(SubscriptionType.Key_Shared) + .subscribe(); + } catch (PulsarClientException e) { + exception = e; + } + assertNull(exception); + assertNotNull(moreConsumer); + assertTrue(moreConsumer.isConnected()); + final int messagesToSend = 10; + final String producerName = "producerForKeyShared"; + @Cleanup + final Producer producer = client.newProducer(Schema.STRING) + .topic(topicName) + .enableBatching(false) + .producerName(producerName) + .create(); + for (int i = 0; i < messagesToSend; i++) { + MessageId messageId = producer.newMessage() + .key(UUID.randomUUID().toString()) + .value(producer.getProducerName() + "-" + i) + .send(); + assertNotNull(messageId); + } + log.info("publish messages complete."); + receiveMessagesCheckStickyKeyAndDuplicate(consumerList, messagesToSend); + // To simulate a consumer crashed + Consumer crashedConsumer = consumerList.remove(0); + crashedConsumer.close(); + for (int i = 0; i < messagesToSend; i++) { + MessageId messageId = producer.newMessage() + .key(UUID.randomUUID().toString()) + .value(producer.getProducerName() + "-" + i) + .send(); + assertNotNull(messageId); + } + receiveMessagesCheckStickyKeyAndDuplicate(consumerList, messagesToSend); + closeConsumers(consumerList); + log.info("-- Exiting {} test --", methodName); + } +} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java index f15b280041186..4a12346c767ea 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java @@ -527,6 +527,12 @@ public ContainerExecResult createNamespace(String nsName) throws Exception { "--clusters", clusterName); } + public ContainerExecResult createPartitionedTopic(String topicName, int partitions) throws Exception { + return runAdminCommandOnAnyBroker( + "topics", "create-partitioned-topic", topicName, + "-p", String.valueOf(partitions)); + } + public ContainerExecResult enableDeduplication(String nsName, boolean enabled) throws Exception { return runAdminCommandOnAnyBroker( "namespaces", "set-deduplication", "public/" + nsName,