forked from apache/pulsar
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add messaging integration tests for different subscriptions. (apache#…
…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
- Loading branch information
1 parent
009bc29
commit b161328
Showing
8 changed files
with
831 additions
and
7 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
164 changes: 164 additions & 0 deletions
164
...ntegration/src/test/java/org/apache/pulsar/tests/integration/messaging/MessagingBase.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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(); | ||
| } | ||
| } | ||
| } |
66 changes: 66 additions & 0 deletions
66
...t/java/org/apache/pulsar/tests/integration/messaging/NonPersistentTopicMessagingTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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); | ||
| } | ||
| } |
Oops, something went wrong.