You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2019/05/24 13:58:11 UTC
[pulsar] branch master updated: Add messaging integration tests for
different subscriptions. (#4352)
This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new b161328 Add messaging integration tests for different subscriptions. (#4352)
b161328 is described below
commit b16132870d6695a2ef07427003679a2431b704af
Author: lipenghui <pe...@apache.org>
AuthorDate: Fri May 24 21:58:04 2019 +0800
Add messaging integration tests for different subscriptions. (#4352)
* Add messaging integration tests.
* Add messaging integration tests.
* remove unused consumer from consumer range map.
* put all messages to pendingAcks before write messages to ctx.
* check entry != null
---
.../org/apache/pulsar/broker/service/Consumer.java | 17 +-
.../HashRangeStickyKeyConsumerSelector.java | 12 +-
.../HashRangeStickyKeyConsumerSelectorTest.java | 16 +-
.../tests/integration/messaging/MessagingBase.java | 164 +++++++
.../messaging/NonPersistentTopicMessagingTest.java | 66 +++
.../messaging/PersistentTopicMessagingTest.java | 66 +++
.../integration/messaging/TopicMessagingBase.java | 491 +++++++++++++++++++++
.../integration/topologies/PulsarCluster.java | 6 +
8 files changed, 831 insertions(+), 7 deletions(-)
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 9dccf2d..e2c7fe1 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
@@ -198,6 +198,19 @@ public class Consumer {
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);
@@ -213,10 +226,6 @@ public class Consumer {
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 18b3427..940798f 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 @@ package org.apache.pulsar.broker.service;
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 class HashRangeStickyKeyConsumerSelector implements StickyKeyConsumerSele
@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<Integer, Consumer> lowerEntry = rangeMap.lowerEntry(removeRange);
@@ -98,7 +99,6 @@ public class HashRangeStickyKeyConsumerSelector implements StickyKeyConsumerSele
consumerRange.put(lowerEntry.getValue(), removeRange);
} else {
rangeMap.remove(removeRange);
- consumerRange.remove(consumer);
}
}
}
@@ -147,4 +147,12 @@ public class HashRangeStickyKeyConsumerSelector implements StickyKeyConsumerSele
if(num < 2) return false;
return (num & num - 1) == 0;
}
+
+ Map<Consumer, Integer> getConsumerRange() {
+ return Collections.unmodifiableMap(consumerRange);
+ }
+
+ Map<Integer, Consumer> 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 ff67c20..0bc5fb3 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 class HashRangeStickyKeyConsumerSelectorTest {
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 class HashRangeStickyKeyConsumerSelectorTest {
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 class HashRangeStickyKeyConsumerSelectorTest {
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 class HashRangeStickyKeyConsumerSelectorTest {
}
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 class HashRangeStickyKeyConsumerSelectorTest {
}
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 class HashRangeStickyKeyConsumerSelectorTest {
}
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 0000000..0043dc2
--- /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 <T extends Comparable<T>> void receiveMessagesCheckOrderAndDuplicate
+ (List<Consumer<T>> consumerList, int messagesToReceive) throws PulsarClientException {
+ Set<T> messagesReceived = Sets.newHashSet();
+ for (Consumer<T> consumer : consumerList) {
+ Message<T> currentReceived = null;
+ Message<T> 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 <T> void receiveMessagesCheckDuplicate
+ (List<Consumer<T>> consumerList, int messagesToReceive) throws PulsarClientException {
+ Set<T> messagesReceived = Sets.newHashSet();
+ for (Consumer<T> consumer : consumerList) {
+ Message<T> 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 <T> void receiveMessagesCheckStickyKeyAndDuplicate
+ (List<Consumer<T>> consumerList, int messagesToReceive) throws PulsarClientException {
+ Map<String, Set<String>> consumerKeys = Maps.newHashMap();
+ Set<T> messagesReceived = Sets.newHashSet();
+ for (Consumer<T> consumer : consumerList) {
+ Message<T> 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<String> 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 <T> void closeConsumers(List<Consumer<T>> consumerList) throws PulsarClientException {
+ Iterator<Consumer<T>> 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 0000000..7fb1fda
--- /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 0000000..39b37ac
--- /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 0000000..26097cf
--- /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<String> 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<String> 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<Consumer<String>> consumerList = new ArrayList<>(3);
+ for (int i = 0; i < partitions; i++) {
+ Consumer<String> 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<String> 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<String> 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<Consumer<String>> consumerList = new ArrayList<>(3);
+ final Consumer<String> consumer = client.newConsumer(Schema.STRING)
+ .topic(topicName)
+ .subscriptionName("test-sub")
+ .subscriptionType(SubscriptionType.Failover)
+ .subscribe();
+ consumerList.add(consumer);
+ Exception exception = null;
+ Consumer<String> 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<String> 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<String> 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<Consumer<String>> consumerList = new ArrayList<>(3);
+ for (int i = 0; i < partitions; i++) {
+ Consumer<String> 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<String> 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<String> 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<String> 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<Consumer<String>> consumerList = new ArrayList<>(2);
+ final Consumer<String> consumer = client.newConsumer(Schema.STRING)
+ .topic(topicName)
+ .subscriptionName("test-sub")
+ .subscriptionType(SubscriptionType.Shared)
+ .subscribe();
+ consumerList.add(consumer);
+ Exception exception = null;
+ Consumer<String> 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<String> 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<String> 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<Consumer<String>> consumerList = new ArrayList<>(3);
+ for (int i = 0; i < partitions; i++) {
+ Consumer<String> 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<String> 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<String> 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<String> 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<Consumer<String>> consumerList = new ArrayList<>(4);
+ final Consumer<String> consumer = client.newConsumer(Schema.STRING)
+ .topic(topicName)
+ .subscriptionName("test-sub")
+ .subscriptionType(SubscriptionType.Key_Shared)
+ .subscribe();
+ consumerList.add(consumer);
+ Exception exception = null;
+ Consumer<String> 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<String> 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<String> 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<Consumer<String>> consumerList = new ArrayList<>(3);
+ for (int i = 0; i < partitions; i++) {
+ Consumer<String> 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<String> 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<String> 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<String> 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 f15b280..4a12346 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 class PulsarCluster {
"--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,