You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2021/03/14 23:45:13 UTC
[pulsar] branch master updated: Add testng group annotations to
pulsar broker tests (#9902)
This is an automated email from the ASF dual-hosted git repository.
penghui 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 5190af3 Add testng group annotations to pulsar broker tests (#9902)
5190af3 is described below
commit 5190af3ec8a202701620ba9e4a8c9d4ed7558118
Author: Ali Ahmed <al...@gmail.com>
AuthorDate: Sun Mar 14 16:44:43 2021 -0700
Add testng group annotations to pulsar broker tests (#9902)
---
.../api/AuthenticatedProducerConsumerTest.java | 4 +-
.../AuthenticationTlsHostnameVerificationTest.java | 1 +
.../api/AuthorizationProducerConsumerTest.java | 1 +
.../pulsar/client/api/BrokerServiceLookupTest.java | 4 +-
.../org/apache/pulsar/client/api/BytesKeyTest.java | 1 +
.../client/api/ClientDeduplicationFailureTest.java | 22 +-
.../pulsar/client/api/ClientDeduplicationTest.java | 14 +-
.../apache/pulsar/client/api/ClientErrorsTest.java | 11 +-
.../pulsar/client/api/ConsumerAckListTest.java | 6 +-
.../client/api/ConsumerBatchReceiveTest.java | 47 +++-
.../pulsar/client/api/ConsumerCleanupTest.java | 1 +
.../pulsar/client/api/ConsumerRedeliveryTest.java | 3 +-
.../pulsar/client/api/DeadLetterTopicTest.java | 1 +
.../client/api/DispatcherBlockConsumerTest.java | 21 +-
.../api/ExposeMessageRedeliveryCountTest.java | 3 +-
.../apache/pulsar/client/api/InterceptorsTest.java | 1 +
.../client/api/KeySharedSubscriptionTest.java | 22 +-
.../apache/pulsar/client/api/MemoryLimitTest.java | 9 +-
.../client/api/MessageDispatchThrottlingTest.java | 5 +-
.../pulsar/client/api/MockBrokerServiceHooks.java | 46 ++--
.../client/api/MutualAuthenticationTest.java | 9 +-
.../client/api/NonDurableSubscriptionTest.java | 1 +
.../pulsar/client/api/NonPersistentTopicTest.java | 3 +-
.../pulsar/client/api/PartitionCreationTest.java | 10 +-
.../api/PartitionedProducerConsumerTest.java | 3 +-
.../pulsar/client/api/ProducerCleanupTest.java | 1 +
.../pulsar/client/api/ProducerConsumerBase.java | 4 +-
.../pulsar/client/api/ProducerCreationTest.java | 1 +
.../pulsar/client/api/ProxyProtocolTest.java | 1 +
...MultiListenersWithInternalListenerNameTest.java | 1 +
...tiListenersWithoutInternalListenerNameTest.java | 1 +
.../apache/pulsar/client/api/RetryTopicTest.java | 1 +
.../pulsar/client/api/ServiceUrlProviderTest.java | 7 +-
.../client/api/SimpleProducerConsumerStatTest.java | 5 +-
.../client/api/SimpleProducerConsumerTest.java | 100 ++++----
.../apache/pulsar/client/api/SimpleSchemaTest.java | 3 +-
.../api/SimpleTypedProducerConsumerTest.java | 3 +-
.../SubscriptionMessageDispatchThrottlingTest.java | 33 ++-
.../org/apache/pulsar/client/api/TenantTest.java | 1 +
.../pulsar/client/api/TlsHostVerification.java | 2 +
.../pulsar/client/api/TlsProducerConsumerBase.java | 6 +-
.../pulsar/client/api/TlsProducerConsumerTest.java | 9 +-
.../org/apache/pulsar/client/api/TlsSniTest.java | 1 +
.../TokenAuthenticatedProducerConsumerTest.java | 3 +-
...kenOauth2AuthenticatedProducerConsumerTest.java | 6 +-
.../apache/pulsar/client/api/TopicReaderTest.java | 11 +-
.../client/api/v1/V1_ProducerConsumerBase.java | 2 +-
.../client/api/v1/V1_ProducerConsumerTest.java | 13 +-
.../client/impl/AdminApiKeyStoreTlsAuthTest.java | 6 +-
.../impl/BatchMessageIdImplSerializationTest.java | 6 +-
.../impl/BatchMessageIndexAckDisableTest.java | 7 +-
.../client/impl/BatchMessageIndexAckTest.java | 4 +-
.../client/impl/BrokerClientIntegrationTest.java | 31 +--
.../client/impl/CompactedOutBatchMessageTest.java | 2 +
.../pulsar/client/impl/ConnectionPoolTest.java | 1 +
.../client/impl/ConsumeBaseExceptionTest.java | 1 +
.../client/impl/ConsumerAckResponseTest.java | 4 +-
.../client/impl/ConsumerConfigurationTest.java | 1 +
.../impl/ConsumerDedupPermitsUpdateTest.java | 126 ++++++++++
.../client/impl/ConsumerUnsubscribeTest.java | 5 +-
...eyStoreTlsProducerConsumerTestWithAuthTest.java | 270 +++++++++++++++++++++
...toreTlsProducerConsumerTestWithoutAuthTest.java | 255 +++++++++++++++++++
.../apache/pulsar/client/impl/KeyStoreTlsTest.java | 1 +
.../pulsar/client/impl/MessageChecksumTest.java | 1 +
.../pulsar/client/impl/MessageChunkingTest.java | 3 +-
.../apache/pulsar/client/impl/MessageIdTest.java | 1 +
.../pulsar/client/impl/MessageParserTest.java | 1 +
.../client/impl/MessagePublishThrottlingTest.java | 1 +
.../pulsar/client/impl/MessageRedeliveryTest.java | 1 +
.../pulsar/client/impl/MultiTopicsReaderTest.java | 8 +-
.../pulsar/client/impl/NegativeAcksTest.java | 1 +
.../client/impl/PatternTopicsConsumerImplTest.java | 10 +-
.../PerMessageUnAcknowledgedRedeliveryTest.java | 1 +
.../pulsar/client/impl/ProducerSemaphoreTest.java | 2 +-
.../client/impl/PulsarMultiHostClientTest.java | 5 +-
.../pulsar/client/impl/RawMessageSerDeserTest.java | 4 +-
.../apache/pulsar/client/impl/RawReaderTest.java | 7 +-
.../org/apache/pulsar/client/impl/ReaderTest.java | 7 +-
.../pulsar/client/impl/SchemaDeleteTest.java | 3 +-
.../client/impl/SequenceIdWithErrorTest.java | 5 +-
.../pulsar/client/impl/TopicDoesNotExistsTest.java | 2 +-
.../pulsar/client/impl/TopicFromMessageTest.java | 1 +
.../impl/TopicPublishThrottlingInitTest.java | 1 +
.../pulsar/client/impl/TopicsConsumerImplTest.java | 6 +-
.../client/impl/TransactionEndToEndTest.java | 6 +-
.../impl/UnAcknowledgedMessagesTimeoutTest.java | 3 +-
.../pulsar/client/impl/ZeroQueueSizeTest.java | 15 +-
.../pulsar/common/naming/NamespaceBundleTest.java | 9 +-
.../pulsar/common/naming/NamespaceBundlesTest.java | 68 +++---
.../common/naming/ServiceConfigurationTest.java | 7 +-
.../pulsar/compaction/CompactedTopicTest.java | 8 +-
.../apache/pulsar/compaction/CompactionTest.java | 1 +
.../apache/pulsar/compaction/CompactorTest.java | 2 +-
.../service/web/DiscoveryServiceWebTest.java | 12 +-
.../worker/PulsarFunctionE2ESecurityTest.java | 8 +-
.../worker/PulsarFunctionLocalRunTest.java | 27 ++-
.../worker/PulsarFunctionPublishTest.java | 2 +-
.../functions/worker/PulsarFunctionTlsTest.java | 4 +-
.../worker/PulsarWorkerAssignmentTest.java | 13 +-
.../stats/client/PulsarBrokerStatsClientTest.java | 3 +-
.../websocket/proxy/ProxyAuthenticationTest.java | 18 +-
.../websocket/proxy/ProxyAuthorizationTest.java | 1 +
.../websocket/proxy/ProxyConfigurationTest.java | 1 +
.../websocket/proxy/ProxyPublishConsumeTest.java | 11 +-
.../proxy/ProxyPublishConsumeTlsTest.java | 3 +-
.../proxy/ProxyPublishConsumeWithoutZKTest.java | 3 +-
.../websocket/proxy/SimpleConsumerSocket.java | 3 +-
.../websocket/proxy/SimpleProducerSocket.java | 2 +-
.../proxy/v1/V1_ProxyAuthenticationTest.java | 17 +-
109 files changed, 1130 insertions(+), 372 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java
index 1e5507b..5647e00 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java
@@ -18,7 +18,6 @@
*/
package org.apache.pulsar.client.api;
-import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.spy;
import java.util.EnumSet;
@@ -51,6 +50,7 @@ import org.testng.annotations.Test;
import com.google.common.collect.Sets;
+@Test(groups = "broker-api")
public class AuthenticatedProducerConsumerTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(AuthenticatedProducerConsumerTest.class);
@@ -128,7 +128,7 @@ public class AuthenticatedProducerConsumerTest extends ProducerConsumerBase {
return new Object[][] { { 0 }, { 1000 } };
}
- public void testSyncProducerAndConsumer(int batchMessageDelayMs) throws Exception {
+ private void testSyncProducerAndConsumer(int batchMessageDelayMs) throws Exception {
Consumer<byte[]> consumer = pulsarClient.newConsumer().topic("persistent://my-property/my-ns/my-topic")
.subscriptionName("my-subscriber-name").subscribe();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticationTlsHostnameVerificationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticationTlsHostnameVerificationTest.java
index a95e755..4bc9617 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticationTlsHostnameVerificationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticationTlsHostnameVerificationTest.java
@@ -45,6 +45,7 @@ import org.testng.annotations.Test;
import com.google.common.collect.Sets;
+@Test(groups = "broker-api")
public class AuthenticationTlsHostnameVerificationTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(AuthenticationTlsHostnameVerificationTest.class);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java
index eec6f1e..895143e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java
@@ -58,6 +58,7 @@ import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.Test;
+@Test(groups = "broker-api")
public class AuthorizationProducerConsumerTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(AuthorizationProducerConsumerTest.class);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java
index 3d9ff9a..d612fec 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java
@@ -99,6 +99,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "broker-api")
public class BrokerServiceLookupTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(BrokerServiceLookupTest.class);
@@ -192,9 +193,6 @@ public class BrokerServiceLookupTest extends ProducerConsumerBase {
consumer.close();
producer.close();
- loadManager1 = null;
- loadManager2 = null;
-
}
/**
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BytesKeyTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BytesKeyTest.java
index 691d329..f0e9f4a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BytesKeyTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/BytesKeyTest.java
@@ -26,6 +26,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "broker-api")
public class BytesKeyTest extends ProducerConsumerBase {
@BeforeMethod
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientDeduplicationFailureTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientDeduplicationFailureTest.java
index 16e602a..99d7b1d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientDeduplicationFailureTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientDeduplicationFailureTest.java
@@ -20,9 +20,8 @@ package org.apache.pulsar.client.api;
import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.retryStrategically;
import static org.mockito.Mockito.spy;
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertTrue;
-import static org.testng.Assert.fail;
+import static org.testng.Assert.*;
+
import com.google.common.base.Function;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
@@ -57,6 +56,7 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@Slf4j
+@Test(groups = "broker-api")
public class ClientDeduplicationFailureTest {
LocalBookkeeperEnsemble bkEnsemble;
@@ -131,7 +131,7 @@ public class ClientDeduplicationFailureTest {
private Thread thread;
private Producer<String> producer;
private long i = 1;
- private AtomicLong atomicLong = new AtomicLong(0);
+ private final AtomicLong atomicLong = new AtomicLong(0);
private CompletableFuture<MessageId> lastMessageFuture;
public ProducerThread(Producer<String> producer) {
@@ -256,7 +256,7 @@ public class ClientDeduplicationFailureTest {
log.info("# of messages read: {}", count);
- assertTrue(prevMessage != null);
+ assertNotNull(prevMessage);
assertEquals(prevMessage.getSequenceId(), producerThread.getLastSeqId());
}
@@ -314,13 +314,13 @@ public class ClientDeduplicationFailureTest {
}, 5, 200);
TopicStats topicStats1 = admin.topics().getStats(sourceTopic);
- assertTrue(topicStats1!= null);
- assertTrue(topicStats1.subscriptions.get(subscriptionName1) != null);
+ assertNotNull(topicStats1);
+ assertNotNull(topicStats1.subscriptions.get(subscriptionName1));
assertEquals(topicStats1.subscriptions.get(subscriptionName1).consumers.size(), 1);
assertEquals(topicStats1.subscriptions.get(subscriptionName1).consumers.get(0).consumerName, consumerName1);
TopicStats topicStats2 = admin.topics().getStats(sourceTopic);
- assertTrue(topicStats2!= null);
- assertTrue(topicStats2.subscriptions.get(subscriptionName2) != null);
+ assertNotNull(topicStats2);
+ assertNotNull(topicStats2.subscriptions.get(subscriptionName2));
assertEquals(topicStats2.subscriptions.get(subscriptionName2).consumers.size(), 1);
assertEquals(topicStats2.subscriptions.get(subscriptionName2).consumers.get(0).consumerName, consumerName2);
@@ -395,11 +395,11 @@ public class ClientDeduplicationFailureTest {
retryStrategically((test) -> msgRecvd.size() >= 20, 5, 200);
assertEquals(msgRecvd.size(), 20);
- for (int i=0; i<10; i++) {
+ for (int i = 0; i < 10; i++) {
assertEquals(msgRecvd.get(i).getValue(), "foo-" + i);
assertEquals(msgRecvd.get(i).getSequenceId(), i);
}
- for (int i=10; i<20; i++) {
+ for (int i = 10; i <20; i++) {
assertEquals(msgRecvd.get(i).getValue(), "foo-" + (i + 10));
assertEquals(msgRecvd.get(i).getSequenceId(), i + 10);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientDeduplicationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientDeduplicationTest.java
index bdb0cc4..304bb6e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientDeduplicationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientDeduplicationTest.java
@@ -30,7 +30,9 @@ import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
+@Test(groups = "flaky")
public class ClientDeduplicationTest extends ProducerConsumerBase {
+
@BeforeClass
@Override
protected void setup() throws Exception {
@@ -58,7 +60,7 @@ public class ClientDeduplicationTest extends ProducerConsumerBase {
@Test
public void testProducerSequenceAfterReconnect() throws Exception {
- String topic = "persistent://my-property/my-ns/testProducerSequenceAfterReconnect";
+ final String topic = "persistent://my-property/my-ns/testProducerSequenceAfterReconnect";
admin.namespaces().setDeduplicationStatus("my-property/my-ns", true);
ProducerBuilder<byte[]> producerBuilder = pulsarClient.newProducer().topic(topic)
@@ -178,8 +180,14 @@ public class ClientDeduplicationTest extends ProducerConsumerBase {
admin.namespaces().setDeduplicationStatus("my-property/my-ns", true);
// Set infinite timeout
- ProducerBuilder<byte[]> producerBuilder = pulsarClient.newProducer().topic(topic)
- .producerName("my-producer-name").enableBatching(true).batchingMaxMessages(10).sendTimeout(0, TimeUnit.SECONDS);
+ ProducerBuilder<byte[]> producerBuilder =
+ pulsarClient.newProducer()
+ .topic(topic)
+ .producerName("my-producer-name")
+ .enableBatching(true)
+ .batchingMaxMessages(10)
+ .sendTimeout(0, TimeUnit.SECONDS);
+
Producer<byte[]> producer = producerBuilder.create();
assertEquals(producer.getLastSequenceId(), -1L);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java
index 2ce30a3..5936632 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java
@@ -41,12 +41,11 @@ import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
-/**
- */
+@Test(groups = "broker-api")
public class ClientErrorsTest {
MockBrokerService mockBrokerService;
- private static int ASYNC_EVENT_COMPLETION_WAIT = 100;
+ private static final int ASYNC_EVENT_COMPLETION_WAIT = 100;
private final String ASSERTION_ERROR = "AssertionError";
@@ -56,13 +55,13 @@ public class ClientErrorsTest {
mockBrokerService.start();
}
- @AfterClass(alwaysRun = true)
+ @AfterClass(alwaysRun = true, groups = "broker-api")
public void teardown() {
mockBrokerService.stop();
}
@Test
- public void testMockBrokerService() throws Exception {
+ public void testMockBrokerService() {
// test default actions of mock broker service
try {
PulsarClient client = PulsarClient.builder().serviceUrl(mockBrokerService.getBrokerAddress()).build();
@@ -563,7 +562,7 @@ public class ClientErrorsTest {
}
// Run this test multiple times to reproduce race conditions on reconnection logic
- @Test(invocationCount = 10)
+ @Test(invocationCount = 10, groups = "broker-api")
public void testProducerReconnect() throws Exception {
AtomicInteger numOfConnections = new AtomicInteger();
AtomicReference<ChannelHandlerContext> channelCtx = new AtomicReference<>();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerAckListTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerAckListTest.java
index 5238796..e101a51 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerAckListTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerAckListTest.java
@@ -32,6 +32,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
+@Test(groups = "broker-api")
public class ConsumerAckListTest extends ProducerConsumerBase {
@BeforeClass
@@ -60,18 +61,20 @@ public class ConsumerAckListTest extends ProducerConsumerBase {
ackListMessage(false,true, ackReceiptEnabled);
}
- public void ackListMessage(boolean isBatch, boolean isPartitioned, boolean ackReceiptEnabled) throws Exception {
+ private void ackListMessage(boolean isBatch, boolean isPartitioned, boolean ackReceiptEnabled) throws Exception {
final String topic = "persistent://my-property/my-ns/batch-ack-" + UUID.randomUUID();
final String subName = "testBatchAck-sub" + UUID.randomUUID();
final int messageNum = ThreadLocalRandom.current().nextInt(50, 100);
if (isPartitioned) {
admin.topics().createPartitionedTopic(topic, 3);
}
+
@Cleanup
Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
.enableBatching(isBatch)
.batchingMaxPublishDelay(50, TimeUnit.MILLISECONDS)
.topic(topic).create();
+
@Cleanup
Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
.subscriptionType(SubscriptionType.Shared)
@@ -81,6 +84,7 @@ public class ConsumerAckListTest extends ProducerConsumerBase {
.enableBatchIndexAcknowledgment(ackReceiptEnabled)
.isAckReceiptEnabled(ackReceiptEnabled)
.subscribe();
+
sendMessagesAsyncAndWait(producer, messageNum);
List<MessageId> messages = new ArrayList<>();
for (int i = 0; i < messageNum; i++) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerBatchReceiveTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerBatchReceiveTest.java
index af1bc98..b1025e7 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerBatchReceiveTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerBatchReceiveTest.java
@@ -32,6 +32,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
+@Test(groups = "flaky")
public class ConsumerBatchReceiveTest extends ProducerConsumerBase {
@BeforeClass
@@ -325,39 +326,57 @@ public class ConsumerBatchReceiveTest extends ProducerConsumerBase {
}
@Test(dataProvider = "batchReceivePolicy")
- public void testBatchReceiveNonPartitionedTopic(BatchReceivePolicy batchReceivePolicy, boolean batchProduce, int receiverQueueSize, boolean isEnableAckReceipt) throws Exception {
+ public void testBatchReceiveNonPartitionedTopic(BatchReceivePolicy batchReceivePolicy,
+ boolean batchProduce,
+ int receiverQueueSize,
+ boolean isEnableAckReceipt) throws Exception {
final String topic = "persistent://my-property/my-ns/batch-receive-non-partition-" + UUID.randomUUID();
testBatchReceive(topic, batchReceivePolicy, batchProduce, receiverQueueSize, isEnableAckReceipt);
}
@Test(dataProvider = "batchReceivePolicy")
- public void testBatchReceivePartitionedTopic(BatchReceivePolicy batchReceivePolicy, boolean batchProduce, int receiverQueueSize, boolean isEnableAckReceipt) throws Exception {
+ public void testBatchReceivePartitionedTopic(BatchReceivePolicy batchReceivePolicy,
+ boolean batchProduce,
+ int receiverQueueSize,
+ boolean isEnableAckReceipt) throws Exception {
final String topic = "persistent://my-property/my-ns/batch-receive-" + UUID.randomUUID();
admin.topics().createPartitionedTopic(topic, 3);
testBatchReceive(topic, batchReceivePolicy, batchProduce, receiverQueueSize, isEnableAckReceipt);
}
@Test(dataProvider = "batchReceivePolicy")
- public void testAsyncBatchReceiveNonPartitionedTopic(BatchReceivePolicy batchReceivePolicy, boolean batchProduce, int receiverQueueSize, boolean isEnableAckReceipt) throws Exception {
+ public void testAsyncBatchReceiveNonPartitionedTopic(BatchReceivePolicy batchReceivePolicy,
+ boolean batchProduce,
+ int receiverQueueSize,
+ boolean isEnableAckReceipt) throws Exception {
final String topic = "persistent://my-property/my-ns/batch-receive-non-partition-async-" + UUID.randomUUID();
testBatchReceiveAsync(topic, batchReceivePolicy, batchProduce, receiverQueueSize, isEnableAckReceipt);
}
@Test(dataProvider = "batchReceivePolicy")
- public void testAsyncBatchReceivePartitionedTopic(BatchReceivePolicy batchReceivePolicy, boolean batchProduce, int receiverQueueSize, boolean isEnableAckReceipt) throws Exception {
+ public void testAsyncBatchReceivePartitionedTopic(BatchReceivePolicy batchReceivePolicy,
+ boolean batchProduce,
+ int receiverQueueSize,
+ boolean isEnableAckReceipt) throws Exception {
final String topic = "persistent://my-property/my-ns/batch-receive-async-" + UUID.randomUUID();
admin.topics().createPartitionedTopic(topic, 3);
testBatchReceiveAsync(topic, batchReceivePolicy, batchProduce, receiverQueueSize, isEnableAckReceipt);
}
@Test(dataProvider = "batchReceivePolicy")
- public void testBatchReceiveAndRedeliveryNonPartitionedTopic(BatchReceivePolicy batchReceivePolicy, boolean batchProduce, int receiverQueueSize, boolean isEnableAckReceipt) throws Exception {
+ public void testBatchReceiveAndRedeliveryNonPartitionedTopic(BatchReceivePolicy batchReceivePolicy,
+ boolean batchProduce,
+ int receiverQueueSize,
+ boolean isEnableAckReceipt) throws Exception {
final String topic = "persistent://my-property/my-ns/batch-receive-and-redelivery-non-partition-" + UUID.randomUUID();
testBatchReceiveAndRedelivery(topic, batchReceivePolicy, batchProduce, receiverQueueSize, isEnableAckReceipt);
}
@Test(dataProvider = "batchReceivePolicy")
- public void testBatchReceiveAndRedeliveryPartitionedTopic(BatchReceivePolicy batchReceivePolicy, boolean batchProduce, int receiverQueueSize, boolean isEnableAckReceipt) throws Exception {
+ public void testBatchReceiveAndRedeliveryPartitionedTopic(BatchReceivePolicy batchReceivePolicy,
+ boolean batchProduce,
+ int receiverQueueSize,
+ boolean isEnableAckReceipt) throws Exception {
final String topic = "persistent://my-property/my-ns/batch-receive-and-redelivery-" + UUID.randomUUID();
admin.topics().createPartitionedTopic(topic, 3);
testBatchReceiveAndRedelivery(topic, batchReceivePolicy, batchProduce, receiverQueueSize, isEnableAckReceipt);
@@ -385,7 +404,9 @@ public class ConsumerBatchReceiveTest extends ProducerConsumerBase {
}
- private void receiveAllBatchesAndVerifyBatchSizeIsEqualToMaxNumMessages(Consumer<String> consumer, BatchReceivePolicy batchReceivePolicy, int numOfExpectedBatches) throws PulsarClientException {
+ private void receiveAllBatchesAndVerifyBatchSizeIsEqualToMaxNumMessages(Consumer<String> consumer,
+ BatchReceivePolicy batchReceivePolicy,
+ int numOfExpectedBatches) throws PulsarClientException {
Messages<String> messages;
for (int i = 0; i < numOfExpectedBatches; i++) {
messages = consumer.batchReceive();
@@ -415,7 +436,11 @@ public class ConsumerBatchReceiveTest extends ProducerConsumerBase {
batchReceiveAndCheck(consumer, 100);
}
- private void testBatchReceiveAsync(String topic, BatchReceivePolicy batchReceivePolicy, boolean batchProduce, int receiverQueueSize, boolean isEnableAckReceipt) throws Exception {
+ private void testBatchReceiveAsync(String topic,
+ BatchReceivePolicy batchReceivePolicy,
+ boolean batchProduce,
+ int receiverQueueSize,
+ boolean isEnableAckReceipt) throws Exception {
if (batchReceivePolicy.getTimeoutMs() <= 0) {
return;
}
@@ -444,7 +469,11 @@ public class ConsumerBatchReceiveTest extends ProducerConsumerBase {
latch.await();
}
- private void testBatchReceiveAndRedelivery(String topic, BatchReceivePolicy batchReceivePolicy, boolean batchProduce, int receiverQueueSize, boolean isEnableAckReceipt) throws Exception {
+ private void testBatchReceiveAndRedelivery(String topic,
+ BatchReceivePolicy batchReceivePolicy,
+ boolean batchProduce,
+ int receiverQueueSize,
+ boolean isEnableAckReceipt) throws Exception {
ProducerBuilder<String> producerBuilder = pulsarClient.newProducer(Schema.STRING).topic(topic);
if (!batchProduce) {
producerBuilder.enableBatching(false);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerCleanupTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerCleanupTest.java
index abe93547..2c4c07b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerCleanupTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerCleanupTest.java
@@ -29,6 +29,7 @@ import org.testng.annotations.Test;
import java.util.UUID;
+@Test(groups = "broker-api")
public class ConsumerCleanupTest extends ProducerConsumerBase {
@BeforeClass
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerRedeliveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerRedeliveryTest.java
index 37337a8..768eeed 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerRedeliveryTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerRedeliveryTest.java
@@ -29,7 +29,6 @@ import lombok.Cleanup;
import org.apache.pulsar.client.impl.ConsumerImpl;
import org.apache.pulsar.client.impl.MessageIdImpl;
-import org.apache.pulsar.common.util.FutureUtil;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
@@ -41,7 +40,9 @@ import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.assertEquals;
+@Test(groups = "broker-api")
public class ConsumerRedeliveryTest extends ProducerConsumerBase {
+
@BeforeClass
@Override
protected void setup() throws Exception {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicTest.java
index 80e3c0d..06f6faf 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicTest.java
@@ -40,6 +40,7 @@ import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
+@Test(groups = "flaky")
public class DeadLetterTopicTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(DeadLetterTopicTest.class);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DispatcherBlockConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DispatcherBlockConsumerTest.java
index 604edc3..d24991a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DispatcherBlockConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DispatcherBlockConsumerTest.java
@@ -66,17 +66,18 @@ import com.google.common.collect.Multimap;
import com.google.common.collect.Queues;
import com.google.common.collect.Sets;
+@Test(groups = "flaky")
public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(DispatcherBlockConsumerTest.class);
- @BeforeMethod
+ @BeforeMethod(groups = { "broker" })
@Override
protected void setup() throws Exception {
super.internalSetup();
super.producerBaseSetup();
}
- @AfterMethod(alwaysRun = true)
+ @AfterMethod(alwaysRun = true, groups = { "broker" })
@Override
protected void cleanup() throws Exception {
super.internalCleanup();
@@ -391,7 +392,7 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
* @throws Exception
*/
@Test(enabled = false) // See https://github.com/apache/pulsar/issues/5438
- public void testRedeliveryOnBlockedDistpatcher() throws Exception {
+ public void testRedeliveryOnBlockedDispatcher() throws Exception {
log.info("-- Starting {} test --", methodName);
int unAckedMessages = pulsar.getConfiguration().getMaxUnackedMessagesPerSubscription();
@@ -620,7 +621,7 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
int receivedMsgCount = 0;
for (int i = 0; i < totalProducedMsgs; i++) {
Message<?> msg = consumer.receive(500, TimeUnit.MILLISECONDS);
- assertTrue(msg != null);
+ assertNotNull(msg);
if (!unackMessages.contains(i)) {
consumer.acknowledge(msg);
}
@@ -673,10 +674,9 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
* produced messages
* </pre>
*
- * @throws Exception
*/
@Test(timeOut = 10000)
- public void testBlockBrokerDispatching() throws Exception {
+ public void testBlockBrokerDispatching() {
log.info("-- Starting {} test --", methodName);
int unAckedMessages = pulsar.getConfiguration().getMaxUnackedMessagesPerBroker();
@@ -701,8 +701,8 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
Field field = BrokerService.class.getDeclaredField("blockedDispatchers");
field.setAccessible(true);
@SuppressWarnings("unchecked")
- ConcurrentOpenHashSet<PersistentDispatcherMultipleConsumers> blockedDispatchers = (ConcurrentOpenHashSet<PersistentDispatcherMultipleConsumers>) field
- .get(pulsar.getBrokerService());
+ ConcurrentOpenHashSet<PersistentDispatcherMultipleConsumers> blockedDispatchers =
+ (ConcurrentOpenHashSet<PersistentDispatcherMultipleConsumers>) field.get(pulsar.getBrokerService());
final int receiverQueueSize = 10;
final int totalProducedMsgs = maxUnAckPerBroker * 3;
@@ -863,7 +863,6 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
} finally {
pulsar.getConfiguration().setMaxUnackedMessagesPerBroker(unAckedMessages);
pulsar.getConfiguration().setMaxUnackedMessagesPerSubscriptionOnBrokerBlocked(unAckedMessagePercentage);
-
executor.shutdownNow();
}
}
@@ -906,8 +905,8 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
Field field = BrokerService.class.getDeclaredField("blockedDispatchers");
field.setAccessible(true);
- ConcurrentOpenHashSet<PersistentDispatcherMultipleConsumers> blockedDispatchers = (ConcurrentOpenHashSet<PersistentDispatcherMultipleConsumers>) field
- .get(pulsar.getBrokerService());
+ ConcurrentOpenHashSet<PersistentDispatcherMultipleConsumers> blockedDispatchers =
+ (ConcurrentOpenHashSet<PersistentDispatcherMultipleConsumers>) field.get(pulsar.getBrokerService());
final int receiverQueueSize = 10;
final int totalProducedMsgs = maxUnAckPerBroker * 3;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ExposeMessageRedeliveryCountTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ExposeMessageRedeliveryCountTest.java
index fecbc7e..ce3bbdc 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ExposeMessageRedeliveryCountTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ExposeMessageRedeliveryCountTest.java
@@ -28,6 +28,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
+@Test(groups = "broker-api")
public class ExposeMessageRedeliveryCountTest extends ProducerConsumerBase {
@BeforeMethod
@@ -118,7 +119,7 @@ public class ExposeMessageRedeliveryCountTest extends ProducerConsumerBase {
}
@Test(timeOut = 30000)
- public void testRedeliveryCountWhenConsumerDisconnected() throws PulsarClientException, InterruptedException {
+ public void testRedeliveryCountWhenConsumerDisconnected() throws PulsarClientException {
String topic = "persistent://my-property/my-ns/testRedeliveryCountWhenConsumerDisconnected";
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/InterceptorsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/InterceptorsTest.java
index 4d2f60b..a12cf51 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/InterceptorsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/InterceptorsTest.java
@@ -40,6 +40,7 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+@Test(groups = "broker-api")
public class InterceptorsTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(InterceptorsTest.class);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java
index 7084bb7..bbec3cd 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java
@@ -48,7 +48,6 @@ import org.apache.curator.shaded.com.google.common.collect.Lists;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.persistent.PersistentStickyKeyDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
-import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.schema.KeyValue;
import org.apache.pulsar.common.util.Murmur3_32Hash;
import org.awaitility.Awaitility;
@@ -60,6 +59,7 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+@Test(groups = "flaky")
public class KeySharedSubscriptionTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(KeySharedSubscriptionTest.class);
@@ -270,7 +270,9 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
}
@Test(dataProvider = "data")
- public void testNonKeySendAndReceiveWithHashRangeAutoSplitStickyKeyConsumerSelector(String topicType, boolean enableBatch) throws PulsarClientException {
+ public void testNonKeySendAndReceiveWithHashRangeAutoSplitStickyKeyConsumerSelector(String topicType,
+ boolean enableBatch)
+ throws PulsarClientException {
this.conf.setSubscriptionKeySharedEnable(true);
String topic = topicType + "://public/default/key_shared_none_key-" + UUID.randomUUID();
@@ -296,7 +298,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
}
@Test(dataProvider = "batch")
- public void testNonKeySendAndReceiveWithHashRangeExclusiveStickyKeyConsumerSelector(boolean enableBatch) throws PulsarClientException {
+ public void testNonKeySendAndReceiveWithHashRangeExclusiveStickyKeyConsumerSelector(boolean enableBatch)
+ throws PulsarClientException {
this.conf.setSubscriptionKeySharedEnable(true);
String topic = "persistent://public/default/key_shared_none_key_exclusive-" + UUID.randomUUID();
@@ -334,7 +337,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
}
@Test(dataProvider = "batch")
- public void testOrderingKeyWithHashRangeAutoSplitStickyKeyConsumerSelector(boolean enableBatch) throws PulsarClientException {
+ public void testOrderingKeyWithHashRangeAutoSplitStickyKeyConsumerSelector(boolean enableBatch)
+ throws PulsarClientException {
this.conf.setSubscriptionKeySharedEnable(true);
String topic = "persistent://public/default/key_shared_ordering_key-" + UUID.randomUUID();
@@ -362,7 +366,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
}
@Test(dataProvider = "batch")
- public void testOrderingKeyWithHashRangeExclusiveStickyKeyConsumerSelector(boolean enableBatch) throws PulsarClientException {
+ public void testOrderingKeyWithHashRangeExclusiveStickyKeyConsumerSelector(boolean enableBatch)
+ throws PulsarClientException {
this.conf.setSubscriptionKeySharedEnable(true);
String topic = "persistent://public/default/key_shared_exclusive_ordering_key-" + UUID.randomUUID();
@@ -424,7 +429,7 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
.subscribe();
}
- @Test()
+ @Test
public void testCannotUseAcknowledgeCumulative() throws PulsarClientException {
this.conf.setSubscriptionKeySharedEnable(true);
String topic = "persistent://public/default/key_shared_ack_cumulative-" + UUID.randomUUID();
@@ -1025,7 +1030,8 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
return createConsumer(topic, null);
}
- private Consumer<Integer> createConsumer(String topic, KeySharedPolicy keySharedPolicy) throws PulsarClientException {
+ private Consumer<Integer> createConsumer(String topic, KeySharedPolicy keySharedPolicy)
+ throws PulsarClientException {
ConsumerBuilder<Integer> builder = pulsarClient.newConsumer(Schema.INT32);
builder.topic(topic)
.subscriptionName("key_shared")
@@ -1143,7 +1149,7 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
}
Assert.assertEquals(check.getValue().intValue(), received);
int redeliveryCount = check.getValue() / 2;
- log.info("[{}] Consumer wait for {} messages redelivery ...", redeliveryCount);
+ log.info("[{}] Consumer wait for {} messages redelivery ...", check, redeliveryCount);
// messages not acked, test redelivery
lastMessageForKey = new HashMap<>();
for (int i = 0; i < redeliveryCount; i++) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MemoryLimitTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MemoryLimitTest.java
index 6b108dc..ec98e7d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MemoryLimitTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MemoryLimitTest.java
@@ -32,6 +32,7 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+@Test(groups = "broker-api")
public class MemoryLimitTest extends ProducerConsumerBase {
@DataProvider(name = "batching")
@@ -56,8 +57,8 @@ public class MemoryLimitTest extends ProducerConsumerBase {
super.internalCleanup();
}
- @Test(dataProvider = "batching")
- public void testRejectMessages(boolean enableBatch)
+ @Test
+ public void testRejectMessages()
throws Exception {
String topic = newTopicName();
@@ -99,8 +100,8 @@ public class MemoryLimitTest extends ProducerConsumerBase {
producer.send(new byte[1024]);
}
- @Test(dataProvider = "batching")
- public void testRejectMessagesOnMultipleTopics(boolean enableBatch) throws Exception {
+ @Test
+ public void testRejectMessagesOnMultipleTopics() throws Exception {
String t1 = newTopicName();
String t2 = newTopicName();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MessageDispatchThrottlingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MessageDispatchThrottlingTest.java
index b5603a8..7bb2699 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MessageDispatchThrottlingTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MessageDispatchThrottlingTest.java
@@ -49,6 +49,7 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+@Test(groups = "flaky")
public class MessageDispatchThrottlingTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(MessageDispatchThrottlingTest.class);
@@ -79,7 +80,7 @@ public class MessageDispatchThrottlingTest extends ProducerConsumerBase {
@DataProvider(name = "subscriptionAndDispatchRateType")
public Object[][] subDisTypeProvider() {
- List<Object[]> mergeList = new LinkedList<Object[]>();
+ List<Object[]> mergeList = new LinkedList<>();
for (Object[] sub : subscriptionsProvider()) {
for (Object[] dispatch : dispatchRateProvider()) {
mergeList.add(merge(sub, dispatch));
@@ -245,7 +246,7 @@ public class MessageDispatchThrottlingTest extends ProducerConsumerBase {
*
* @throws Exception
*/
- @Test()
+ @Test
public void testClusterMsgByteRateLimitingClusterConfig() throws Exception {
log.info("-- Starting {} test --", methodName);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerServiceHooks.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerServiceHooks.java
index e20f092..a68e2d5 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerServiceHooks.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MockBrokerServiceHooks.java
@@ -33,50 +33,48 @@ import org.apache.pulsar.common.api.proto.CommandSend;
import org.apache.pulsar.common.api.proto.CommandSubscribe;
import org.apache.pulsar.common.api.proto.CommandUnsubscribe;
-/**
- */
public interface MockBrokerServiceHooks {
- public interface CommandConnectHook {
- public void apply(ChannelHandlerContext ctx, CommandConnect connect);
+ interface CommandConnectHook {
+ void apply(ChannelHandlerContext ctx, CommandConnect connect);
}
- public interface CommandPartitionLookupHook {
- public void apply(ChannelHandlerContext ctx, CommandPartitionedTopicMetadata connect);
+ interface CommandPartitionLookupHook {
+ void apply(ChannelHandlerContext ctx, CommandPartitionedTopicMetadata connect);
}
- public interface CommandTopicLookupHook {
- public void apply(ChannelHandlerContext ctx, CommandLookupTopic connect);
+ interface CommandTopicLookupHook {
+ void apply(ChannelHandlerContext ctx, CommandLookupTopic connect);
}
- public interface CommandSubscribeHook {
- public void apply(ChannelHandlerContext ctx, CommandSubscribe subscribe);
+ interface CommandSubscribeHook {
+ void apply(ChannelHandlerContext ctx, CommandSubscribe subscribe);
}
- public interface CommandProducerHook {
- public void apply(ChannelHandlerContext ctx, CommandProducer producer);
+ interface CommandProducerHook {
+ void apply(ChannelHandlerContext ctx, CommandProducer producer);
}
- public interface CommandSendHook {
- public void apply(ChannelHandlerContext ctx, CommandSend send, ByteBuf headersAndPayload);
+ interface CommandSendHook {
+ void apply(ChannelHandlerContext ctx, CommandSend send, ByteBuf headersAndPayload);
}
- public interface CommandAckHook {
- public void apply(ChannelHandlerContext ctx, CommandAck ack);
+ interface CommandAckHook {
+ void apply(ChannelHandlerContext ctx, CommandAck ack);
}
- public interface CommandFlowHook {
- public void apply(ChannelHandlerContext ctx, CommandFlow flow);
+ interface CommandFlowHook {
+ void apply(ChannelHandlerContext ctx, CommandFlow flow);
}
- public interface CommandUnsubscribeHook {
- public void apply(ChannelHandlerContext ctx, CommandUnsubscribe unsubscribe);
+ interface CommandUnsubscribeHook {
+ void apply(ChannelHandlerContext ctx, CommandUnsubscribe unsubscribe);
}
- public interface CommandCloseProducerHook {
- public void apply(ChannelHandlerContext ctx, CommandCloseProducer closeProducer);
+ interface CommandCloseProducerHook {
+ void apply(ChannelHandlerContext ctx, CommandCloseProducer closeProducer);
}
- public interface CommandCloseConsumerHook {
- public void apply(ChannelHandlerContext ctx, CommandCloseConsumer closeConsumer);
+ interface CommandCloseConsumerHook {
+ void apply(ChannelHandlerContext ctx, CommandCloseConsumer closeConsumer);
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MutualAuthenticationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MutualAuthenticationTest.java
index 2c01d29..008df50 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MutualAuthenticationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MutualAuthenticationTest.java
@@ -45,17 +45,18 @@ import static java.nio.charset.StandardCharsets.UTF_8;
* Test Mutual Authentication.
* Test connect set success, and producer consumer works well.
*/
+@Test(groups = "broker-api")
public class MutualAuthenticationTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(MutualAuthenticationTest.class);
private MutualAuthentication mutualAuth;
- private static String[] clientAuthStrings = {
+ private static final String[] clientAuthStrings = {
"MutualClientAuthInit", // step 0
"MutualClientStep1" // step 1
};
- private static String[] serverAuthStrings = {
+ private static final String[] serverAuthStrings = {
"ResponseMutualClientAuthInit", // step 0
};
@@ -181,11 +182,11 @@ public class MutualAuthenticationTest extends ProducerConsumerBase {
}
}
- @BeforeMethod
+ @BeforeMethod(groups = "broker-api")
@Override
protected void setup() throws Exception {
mutualAuth = new MutualAuthentication();
- Set<String> superUserRoles = new HashSet<String>();
+ Set<String> superUserRoles = new HashSet<>();
superUserRoles.add("admin");
conf.setSuperUserRoles(superUserRoles);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonDurableSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonDurableSubscriptionTest.java
index 5aadbce..35f1519 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonDurableSubscriptionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonDurableSubscriptionTest.java
@@ -36,6 +36,7 @@ import static org.testng.AssertJUnit.assertFalse;
import static org.testng.AssertJUnit.assertNull;
import static org.testng.AssertJUnit.assertTrue;
+@Test(groups = "broker-api")
public class NonDurableSubscriptionTest extends ProducerConsumerBase {
@BeforeMethod
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentTopicTest.java
index ffdaa28..e416a3e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentTopicTest.java
@@ -73,6 +73,7 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+@Test(groups = "broker-api")
public class NonPersistentTopicTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(NonPersistentTopicTest.class);
private final String configClusterName = "r1";
@@ -879,7 +880,6 @@ public class NonPersistentTopicTest extends ProducerConsumerBase {
} finally {
conf.setMaxConcurrentNonPersistentMessagePerConnection(defaultNonPersistentMessageRate);
}
-
}
class ReplicationClusterManager {
@@ -1052,7 +1052,6 @@ public class NonPersistentTopicTest extends ProducerConsumerBase {
bkEnsemble3.stop();
globalZkS.stop();
}
-
}
private void rolloverPerIntervalStats(PulsarService pulsar) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionCreationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionCreationTest.java
index faad5b5..782c9fc 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionCreationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionCreationTest.java
@@ -18,8 +18,6 @@
*/
package org.apache.pulsar.client.api;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import org.apache.bookkeeper.util.ZkUtils;
import org.apache.pulsar.broker.admin.ZkAdminPaths;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.impl.MultiTopicsConsumerImpl;
@@ -27,9 +25,6 @@ import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.metadata.api.MetadataStoreException;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.ZooDefs;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
@@ -38,8 +33,7 @@ import org.testng.annotations.Test;
import java.util.concurrent.TimeUnit;
-import static org.apache.pulsar.broker.admin.AdminResource.jsonMapper;
-
+@Test(groups = "broker-api")
public class PartitionCreationTest extends ProducerConsumerBase {
@DataProvider(name = "topicDomainProvider")
@@ -125,7 +119,7 @@ public class PartitionCreationTest extends ProducerConsumerBase {
}
@Test(timeOut = 60000, dataProvider = "restCreateMissedPartitions")
- public void testCreateMissedPartitions(boolean useRestApi) throws JsonProcessingException, KeeperException, InterruptedException, PulsarAdminException, PulsarClientException, MetadataStoreException {
+ public void testCreateMissedPartitions(boolean useRestApi) throws PulsarAdminException, PulsarClientException, MetadataStoreException {
conf.setAllowAutoTopicCreation(false);
final String topic = "testCreateMissedPartitions-useRestApi-" + useRestApi;
String path = ZkAdminPaths.partitionedTopicPath(TopicName.get(topic));
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionedProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionedProducerConsumerTest.java
index 3539abf..cb717f5 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionedProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionedProducerConsumerTest.java
@@ -49,6 +49,7 @@ import com.google.common.collect.Sets;
import io.netty.util.concurrent.DefaultThreadFactory;
+@Test(groups = "broker-api")
public class PartitionedProducerConsumerTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(PartitionedProducerConsumerTest.class);
@@ -899,7 +900,7 @@ public class PartitionedProducerConsumerTest extends ProducerConsumerBase {
}
- private class AlwaysTwoMessageRouter implements MessageRouter {
+ private static class AlwaysTwoMessageRouter implements MessageRouter {
@Override
public int choosePartition(Message<?> msg, TopicMetadata metadata) {
return 2;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCleanupTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCleanupTest.java
index c3d99a2..251dd8a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCleanupTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCleanupTest.java
@@ -28,6 +28,7 @@ import org.testng.annotations.Test;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
+@Test(groups = "broker-api")
public class ProducerCleanupTest extends ProducerConsumerBase {
@BeforeMethod
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerConsumerBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerConsumerBase.java
index 8c819e9..1e1ea7b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerConsumerBase.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerConsumerBase.java
@@ -33,12 +33,12 @@ import org.testng.annotations.BeforeMethod;
public abstract class ProducerConsumerBase extends MockedPulsarServiceBaseTest {
protected String methodName;
- @BeforeMethod
+ @BeforeMethod(groups = { "broker", "websocket", "broker-api", "broker-discovery", "broker-impl", "extra", "flaky" })
public void beforeMethod(Method m) throws Exception {
methodName = m.getName();
}
- public void producerBaseSetup() throws Exception {
+ protected void producerBaseSetup() throws Exception {
admin.clusters().createCluster("test", new ClusterData(pulsar.getWebServiceAddress()));
admin.tenants().createTenant("my-property",
new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCreationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCreationTest.java
index 3ed6ef7..0d24a4b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCreationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCreationTest.java
@@ -27,6 +27,7 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+@Test(groups = "broker-api")
public class ProducerCreationTest extends ProducerConsumerBase {
@BeforeMethod
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProxyProtocolTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProxyProtocolTest.java
index d264b83..e5b5324 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProxyProtocolTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProxyProtocolTest.java
@@ -32,6 +32,7 @@ import org.testng.annotations.Test;
import lombok.Cleanup;
+@Test(groups = "broker-api")
public class ProxyProtocolTest extends TlsProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(ProxyProtocolTest.class);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PulsarMultiListenersWithInternalListenerNameTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PulsarMultiListenersWithInternalListenerNameTest.java
index dccd67e..1d5a23b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PulsarMultiListenersWithInternalListenerNameTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PulsarMultiListenersWithInternalListenerNameTest.java
@@ -39,6 +39,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
+@Test(groups = "broker-api")
public class PulsarMultiListenersWithInternalListenerNameTest extends MockedPulsarServiceBaseTest {
private ExecutorService executorService;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PulsarMultiListenersWithoutInternalListenerNameTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PulsarMultiListenersWithoutInternalListenerNameTest.java
index baaa4b7..c81f180 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PulsarMultiListenersWithoutInternalListenerNameTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PulsarMultiListenersWithoutInternalListenerNameTest.java
@@ -39,6 +39,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
+@Test(groups = "broker-api")
public class PulsarMultiListenersWithoutInternalListenerNameTest extends MockedPulsarServiceBaseTest {
private ExecutorService executorService;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/RetryTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/RetryTopicTest.java
index de00379..e9ba509 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/RetryTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/RetryTopicTest.java
@@ -28,6 +28,7 @@ import java.util.concurrent.TimeUnit;
import static org.testng.Assert.assertNull;
+@Test(groups = "broker-api")
public class RetryTopicTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(RetryTopicTest.class);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ServiceUrlProviderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ServiceUrlProviderTest.java
index 4e58fc0..76b38b5 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ServiceUrlProviderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ServiceUrlProviderTest.java
@@ -33,6 +33,7 @@ import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@Slf4j
+@Test(groups = "broker-api")
public class ServiceUrlProviderTest extends ProducerConsumerBase {
@BeforeClass
@@ -133,11 +134,11 @@ public class ServiceUrlProviderTest extends ProducerConsumerBase {
client.close();
}
- class TestServiceUrlProvider implements ServiceUrlProvider {
+ static class TestServiceUrlProvider implements ServiceUrlProvider {
private PulsarClient pulsarClient;
- private String serviceUrl;
+ private final String serviceUrl;
public TestServiceUrlProvider(String serviceUrl) {
this.serviceUrl = serviceUrl;
@@ -158,7 +159,7 @@ public class ServiceUrlProviderTest extends ProducerConsumerBase {
}
}
- class AutoChangedServiceUrlProvider extends TestServiceUrlProvider {
+ static class AutoChangedServiceUrlProvider extends TestServiceUrlProvider {
public AutoChangedServiceUrlProvider(String serviceUrl) {
super(serviceUrl);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerStatTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerStatTest.java
index 41fc357..17d4b05 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerStatTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerStatTest.java
@@ -48,6 +48,7 @@ import com.google.common.util.concurrent.RateLimiter;
import com.google.gson.JsonArray;
import com.google.gson.JsonObject;
+@Test(groups = "broker-api")
public class SimpleProducerConsumerStatTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(SimpleProducerConsumerStatTest.class);
@@ -296,7 +297,7 @@ public class SimpleProducerConsumerStatTest extends ProducerConsumerBase {
ProducerBuilder<byte[]> producerBuilder = pulsarClient.newProducer()
.topic("persistent://my-property/tp1/my-ns/my-topic5").sendTimeout(1, TimeUnit.SECONDS);
if (batchMessageDelayMs != 0) {
- producerBuilder.enableBatching(true).batchingMaxPublishDelay(2 * batchMessageDelayMs, TimeUnit.MILLISECONDS)
+ producerBuilder.enableBatching(true).batchingMaxPublishDelay(2L * batchMessageDelayMs, TimeUnit.MILLISECONDS)
.batchingMaxMessages(5);
}
@@ -361,7 +362,7 @@ public class SimpleProducerConsumerStatTest extends ProducerConsumerBase {
log.info("-- Exiting {} test --", methodName);
}
- public void validatingLogInfo(Consumer<?> consumer, Producer<?> producer, boolean verifyAckCount)
+ private void validatingLogInfo(Consumer<?> consumer, Producer<?> producer, boolean verifyAckCount)
throws InterruptedException {
// Waiting for recording last stat info
Thread.sleep(1000);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
index 5d44a05..9f8edb0 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
@@ -109,10 +109,11 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+@Test(groups = "flaky")
public class SimpleProducerConsumerTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(SimpleProducerConsumerTest.class);
- @BeforeMethod
+ @BeforeMethod(groups = { "broker", "flaky" })
@Override
protected void setup() throws Exception {
super.internalSetup();
@@ -140,7 +141,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
return new Object[][] { { true }, { false } };
}
- @AfterMethod(alwaysRun = true)
+ @AfterMethod(alwaysRun = true, groups = { "broker" })
@Override
protected void cleanup() throws Exception {
super.internalCleanup();
@@ -660,7 +661,6 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
} catch (PulsarClientException.AlreadyClosedException e) {
// ok
}
-
}
@Test
@@ -972,11 +972,9 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
final String topic = "persistent://my-property/my-ns/" + topicName;
ProducerBuilder<byte[]> producerBuilder = pulsarClient.newProducer().topic(topic);
- if (batchMessageDelayMs != 0) {
- producerBuilder.batchingMaxPublishDelay(batchMessageDelayMs, TimeUnit.MILLISECONDS);
- producerBuilder.batchingMaxMessages(5);
- producerBuilder.enableBatching(true);
- }
+ producerBuilder.batchingMaxPublishDelay(batchMessageDelayMs, TimeUnit.MILLISECONDS);
+ producerBuilder.batchingMaxMessages(5);
+ producerBuilder.enableBatching(true);
Producer<byte[]> producer = producerBuilder.create();
PersistentTopic topicRef = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topic).get();
@@ -1074,10 +1072,8 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
.subscriptionType(SubscriptionType.Shared).receiverQueueSize(receiverSize).subscribe();
ProducerBuilder<byte[]> producerBuilder = pulsarClient.newProducer().topic(topic);
- if (batchMessageDelayMs != 0) {
- producerBuilder.enableBatching(true).batchingMaxPublishDelay(batchMessageDelayMs, TimeUnit.MILLISECONDS)
- .batchingMaxMessages(5);
- }
+ producerBuilder.enableBatching(true).batchingMaxPublishDelay(batchMessageDelayMs, TimeUnit.MILLISECONDS)
+ .batchingMaxMessages(5);
Producer<byte[]> producer = producerBuilder.create();
PersistentTopic topicRef = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topic).get();
@@ -1177,7 +1173,6 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
final Set<String> consumeMsgs = Sets.newHashSet();
Consumer<byte[]> consumer = pulsarClient.newConsumer().topic("persistent://my-property/my-ns/my-topic1")
.subscriptionName("my-subscriber-name").subscribe();
- ;
// produce message
Producer<byte[]> producer = pulsarClient.newProducer().topic("persistent://my-property/my-ns/my-topic1")
@@ -1209,7 +1204,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
}
@Test
- public void testtSendCallBackReturnSequenceId() throws Exception {
+ public void testSendCallBackReturnSequenceId() throws Exception {
log.info("-- Starting {} test --", methodName);
ProducerBuilder<byte[]> producerBuilder = pulsarClient.newProducer()
@@ -1308,14 +1303,14 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
consumerMsgSet2.add(msg);
}
- consumerMsgSet1.stream().forEach(m -> {
+ consumerMsgSet1.forEach(m -> {
try {
consumer2.acknowledge(m);
} catch (PulsarClientException e) {
fail();
}
});
- consumerMsgSet2.stream().forEach(m -> {
+ consumerMsgSet2.forEach(m -> {
try {
consumer1.acknowledge(m);
} catch (PulsarClientException e) {
@@ -1374,10 +1369,9 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
* messages This test case enables checksum sending while producing message and broker verifies the checksum for the
* message.
*
- * @throws Exception
*/
@Test(dataProvider = "ackReceiptEnabled")
- public void testConsumerBlockingWithUnAckedMessages(boolean ackReceiptEnabled) throws Exception {
+ public void testConsumerBlockingWithUnAckedMessages(boolean ackReceiptEnabled) {
log.info("-- Starting {} test --", methodName);
int unAckedMessages = pulsar.getConfiguration().getMaxUnackedMessagesPerConsumer();
@@ -1523,10 +1517,9 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
* Verify: Consumer1 which doesn't send ack will not impact Consumer2 which sends ack for consumed message.
*
*
- * @throws Exception
*/
@Test(dataProvider = "ackReceiptEnabled")
- public void testMutlipleSharedConsumerBlockingWithUnAckedMessages(boolean ackReceiptEnabled) throws Exception {
+ public void testMultipleSharedConsumerBlockingWithUnActedMessages(boolean ackReceiptEnabled) {
log.info("-- Starting {} test --", methodName);
int unAckedMessages = pulsar.getConfiguration().getMaxUnackedMessagesPerConsumer();
@@ -1625,7 +1618,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
}
@Test
- public void testShouldNotBlockConsumerIfRedeliverBeforeReceive() throws Exception {
+ public void testShouldNotBlockConsumerIfRedeliverBeforeReceive() {
log.info("-- Starting {} test --", methodName);
int unAckedMessages = pulsar.getConfiguration().getMaxUnackedMessagesPerConsumer();
@@ -1683,7 +1676,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
}
@Test(dataProvider = "ackReceiptEnabled")
- public void testUnackBlockRedeliverMessages(boolean ackReceiptEnabled) throws Exception {
+ public void testUnackBlockRedeliverMessages(boolean ackReceiptEnabled) {
log.info("-- Starting {} test --", methodName);
int unAckedMessages = pulsar.getConfiguration().getMaxUnackedMessagesPerConsumer();
@@ -1751,7 +1744,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
}
@Test(dataProvider = "batchAndAckReceipt")
- public void testUnackedBlockAtBatch(int batchMessageDelayMs, boolean ackReceiptEnabled) throws Exception {
+ public void testUnackedBlockAtBatch(int batchMessageDelayMs, boolean ackReceiptEnabled) {
log.info("-- Starting {} test --", methodName);
int unAckedMessages = pulsar.getConfiguration().getMaxUnackedMessagesPerConsumer();
@@ -1791,7 +1784,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
// (2) Consumer1: consume without ack:
// try to consume messages: but will be able to consume number of messages = maxUnackedMessages
- Message<byte[]> msg = null;
+ Message<byte[]> msg;
List<Message<byte[]>> messages = Lists.newArrayList();
for (int i = 0; i < totalProducedMsgs; i++) {
msg = consumer1.receive(1, TimeUnit.SECONDS);
@@ -1843,10 +1836,9 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
* Verify: Consumer2 sends ack of Consumer1 and consumer1 should be unblock if it is blocked due to unack-messages
*
*
- * @throws Exception
*/
@Test
- public void testBlockUnackConsumerAckByDifferentConsumer() throws Exception {
+ public void testBlockUnackConsumerAckByDifferentConsumer() {
log.info("-- Starting {} test --", methodName);
int unAckedMessages = pulsar.getConfiguration().getMaxUnackedMessagesPerConsumer();
@@ -1944,10 +1936,8 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
ProducerBuilder<byte[]> producerBuilder = pulsarClient.newProducer()
.topic("persistent://my-property/my-ns/my-topic1");
final int batchMessageDelayMs = 300;
- if (batchMessageDelayMs != 0) {
- producerBuilder.enableBatching(true).batchingMaxPublishDelay(batchMessageDelayMs, TimeUnit.MILLISECONDS)
- .batchingMaxMessages(5);
- }
+ producerBuilder.enableBatching(true).batchingMaxPublishDelay(batchMessageDelayMs, TimeUnit.MILLISECONDS)
+ .batchingMaxMessages(5);
Producer<byte[]> producer = producerBuilder.create();
for (int i = 0; i < totalMsg; i++) {
@@ -1976,10 +1966,9 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
*
* Usecase: produce message with 10ms interval: so, consumer can consume only 10 messages without acking
*
- * @throws Exception
*/
@Test
- public void testBlockUnackedConsumerRedeliverySpecificMessagesProduceWithPause() throws Exception {
+ public void testBlockUnackedConsumerRedeliverySpecificMessagesProduceWithPause() {
log.info("-- Starting {} test --", methodName);
int unAckedMessages = pulsar.getConfiguration().getMaxUnackedMessagesPerConsumer();
@@ -2004,7 +1993,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
}
// (2) try to consume messages: but will be able to consume number of messages = unAckedMessagesBufferSize
- Message<byte[]> msg = null;
+ Message<byte[]> msg;
List<Message<byte[]>> messages1 = Lists.newArrayList();
for (int i = 0; i < totalProducedMsgs; i++) {
msg = consumer.receive(1, TimeUnit.SECONDS);
@@ -2058,10 +2047,9 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
* Usecase: Consumer starts consuming only after all messages have been produced. So, consumer consumes total
* receiver-queue-size number messages => ask for redelivery and receives all messages again.
*
- * @throws Exception
*/
@Test
- public void testBlockUnackedConsumerRedeliverySpecificMessagesCloseConsumerWhileProduce() throws Exception {
+ public void testBlockUnackedConsumerRedeliverySpecificMessagesCloseConsumerWhileProduce() {
log.info("-- Starting {} test --", methodName);
int unAckedMessages = pulsar.getConfiguration().getMaxUnackedMessagesPerConsumer();
@@ -2259,7 +2247,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
List<Message<byte[]>> messages = Lists.newArrayList();
- // let consumer1 and consumer2 cosume messages up to the queue will be full
+ // let consumer1 and consumer2 consume messages up to the queue will be full
for (int i = 0; i < totalPublishMessages; i++) {
Message<byte[]> msg = c1.receive(500, TimeUnit.MILLISECONDS);
if (msg != null) {
@@ -2463,14 +2451,14 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
log.info("-- Exiting {} test --", methodName);
}
- @Test(groups = "encryption")
+ @Test
public void testECDSAEncryption() throws Exception {
log.info("-- Starting {} test --", methodName);
String topicName = "persistent://my-property/my-ns/myecdsa-topic1-" + System.currentTimeMillis();
class EncKeyReader implements CryptoKeyReader {
- EncryptionKeyInfo keyInfo = new EncryptionKeyInfo();
+ final EncryptionKeyInfo keyInfo = new EncryptionKeyInfo();
@Override
public EncryptionKeyInfo getPublicKey(String keyName, Map<String, String> keyMeta) {
@@ -2545,14 +2533,14 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
log.info("-- Exiting {} test --", methodName);
}
- @Test(groups = "encryption")
+ @Test
public void testRSAEncryption() throws Exception {
log.info("-- Starting {} test --", methodName);
String topicName = "persistent://my-property/my-ns/myrsa-topic1-"+ System.currentTimeMillis();
class EncKeyReader implements CryptoKeyReader {
- EncryptionKeyInfo keyInfo = new EncryptionKeyInfo();
+ final EncryptionKeyInfo keyInfo = new EncryptionKeyInfo();
@Override
public EncryptionKeyInfo getPublicKey(String keyName, Map<String, String> keyMeta) {
@@ -2632,7 +2620,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
log.info("-- Exiting {} test --", methodName);
}
- @Test(groups = "encryption")
+ @Test
public void testDefaultCryptoKeyReader() throws Exception {
final String topic = "persistent://my-property/my-ns/default-crypto-key-reader" + System.currentTimeMillis();
final String ecdsaPublicKeyFile = "file:./src/test/resources/certificate/public-key.client-ecdsa.pem";
@@ -2676,7 +2664,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
producer1.close();
producer2.close();
- for (Consumer<byte[]> consumer : (List<Consumer<byte[]>>) Lists.newArrayList(consumer1, consumer2)) {
+ for (Consumer<byte[]> consumer : Lists.newArrayList(consumer1, consumer2)) {
MessageImpl<byte[]> msg = null;
for (int i = 0; i < numMsg * 2; i++) {
@@ -2709,7 +2697,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
producer3.close();
producer4.close();
- for (Consumer<byte[]> consumer : (List<Consumer<byte[]>>) Lists.newArrayList(consumer3, consumer4)) {
+ for (Consumer<byte[]> consumer : Lists.newArrayList(consumer3, consumer4)) {
MessageImpl<byte[]> msg = null;
for (int i = 0; i < numMsg * 4; i++) {
@@ -2728,7 +2716,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
consumer4.unsubscribe();
}
- @Test(groups = "encryption")
+ @Test
public void testRedeliveryOfFailedMessages() throws Exception {
log.info("-- Starting {} test --", methodName);
@@ -2737,7 +2725,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
Map<String, String> metadata = Maps.newHashMap();
metadata.put("version", encryptionKeyVersion);
class EncKeyReader implements CryptoKeyReader {
- EncryptionKeyInfo keyInfo = new EncryptionKeyInfo();
+ final EncryptionKeyInfo keyInfo = new EncryptionKeyInfo();
@Override
public EncryptionKeyInfo getPublicKey(String keyName, Map<String, String> keyMeta) {
@@ -2775,7 +2763,6 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
}
class InvalidKeyReader implements CryptoKeyReader {
- EncryptionKeyInfo keyInfo = new EncryptionKeyInfo();
@Override
public EncryptionKeyInfo getPublicKey(String keyName, Map<String, String> keyMeta) {
@@ -2825,7 +2812,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
int numberOfMessages = 100;
String message = "my-message";
- Set<String> messages = new HashSet(); // Since messages are in random order
+ Set<String> messages = new HashSet<>(); // Since messages are in random order
for (int i = 0; i<numberOfMessages; i++) {
producer.send((message + i).getBytes());
}
@@ -2865,13 +2852,13 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
log.info("-- Exiting {} test --", methodName);
}
- @Test(groups = "encryption")
+ @Test
public void testEncryptionFailure() throws Exception {
log.info("-- Starting {} test --", methodName);
class EncKeyReader implements CryptoKeyReader {
- EncryptionKeyInfo keyInfo = new EncryptionKeyInfo();
+ final EncryptionKeyInfo keyInfo = new EncryptionKeyInfo();
@Override
public EncryptionKeyInfo getPublicKey(String keyName, Map<String, String> keyMeta) {
@@ -2992,7 +2979,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
log.info("-- Exiting {} test --", methodName);
}
- @Test(groups = "encryption")
+ @Test
public void testEncryptionConsumerWithoutCryptoReader() throws Exception {
log.info("-- Starting {} test --", methodName);
@@ -3001,7 +2988,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
Map<String, String> metadata = Maps.newHashMap();
metadata.put("version", encryptionKeyVersion);
class EncKeyReader implements CryptoKeyReader {
- EncryptionKeyInfo keyInfo = new EncryptionKeyInfo();
+ final EncryptionKeyInfo keyInfo = new EncryptionKeyInfo();
@Override
public EncryptionKeyInfo getPublicKey(String keyName, Map<String, String> keyMeta) {
@@ -3058,8 +3045,9 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
log.info("-- Exiting {} test --", methodName);
}
- private String decryptMessage(TopicMessageImpl<byte[]> msg, String encryptionKeyName, CryptoKeyReader reader)
- throws Exception {
+ private String decryptMessage(TopicMessageImpl<byte[]> msg,
+ String encryptionKeyName,
+ CryptoKeyReader reader) throws Exception {
Optional<EncryptionContext> ctx = msg.getEncryptionCtx();
Assert.assertTrue(ctx.isPresent());
EncryptionContext encryptionCtx = ctx
@@ -3399,9 +3387,9 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
.getValue().activeConsumerName;
subsCount.compute(activeConsumerName, (k, v) -> v != null ? v + 1 : 1);
});
- return subsCount.size() == 3 && subsCount.get("bbb1") == evenDistributionCount
- && subsCount.get("bbb2") == evenDistributionCount
- && subsCount.get("bbb3") == evenDistributionCount;
+ return subsCount.size() == 3 && subsCount.get("bbb1").equals(evenDistributionCount)
+ && subsCount.get("bbb2").equals(evenDistributionCount)
+ && subsCount.get("bbb3").equals(evenDistributionCount);
} catch (PulsarAdminException e) {
// Ok
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java
index f670ef1..a0de21d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java
@@ -61,6 +61,7 @@ import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+@Test(groups = "broker-api")
public class SimpleSchemaTest extends ProducerConsumerBase {
@DataProvider(name = "batchingModes")
@@ -431,7 +432,7 @@ public class SimpleSchemaTest extends ProducerConsumerBase {
}
@Test
- public void newConsumerWithSchemaOnExistingTopicWithoutSchema() throws Exception {
+ public void newConsumerWithSchemaOnExistingTopicWithoutSchema() {
String topic = "my-property/my-ns/schema-test";
try (Producer<byte[]> p = pulsarClient.newProducer().topic(topic).create();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java
index c3a7324..dc60411 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java
@@ -50,6 +50,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "broker-api")
public class SimpleTypedProducerConsumerTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(SimpleTypedProducerConsumerTest.class);
@@ -309,7 +310,7 @@ public class SimpleTypedProducerConsumerTest extends ProducerConsumerBase {
}
@Test(expectedExceptions = {PulsarClientException.class})
- public void testAvroConsumerWithWrongPrestoredSchema() throws Exception {
+ public void testAvroConsumerWithWrongRestoredSchema() throws Exception {
log.info("-- Starting {} test --", methodName);
byte[] randomSchemaBytes = ("{\n" +
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
index 1bec4d2..bceb7c0 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
@@ -39,6 +39,7 @@ import org.testng.annotations.Test;
import static org.awaitility.Awaitility.await;
+@Test(groups = "flaky")
public class SubscriptionMessageDispatchThrottlingTest extends MessageDispatchThrottlingTest {
private static final Logger log = LoggerFactory.getLogger(SubscriptionMessageDispatchThrottlingTest.class);
@@ -87,9 +88,9 @@ public class SubscriptionMessageDispatchThrottlingTest extends MessageDispatchTh
DispatchRateLimiter subRateLimiter = null;
Dispatcher subDispatcher = topic.getSubscription(subName).getDispatcher();
if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
- subRateLimiter = ((PersistentDispatcherMultipleConsumers) subDispatcher).getRateLimiter().get();
+ subRateLimiter = subDispatcher.getRateLimiter().get();
} else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
- subRateLimiter = ((PersistentDispatcherSingleActiveConsumer) subDispatcher).getRateLimiter().get();
+ subRateLimiter = subDispatcher.getRateLimiter().get();
} else {
Assert.fail("Should only have PersistentDispatcher in this test");
}
@@ -170,9 +171,9 @@ public class SubscriptionMessageDispatchThrottlingTest extends MessageDispatchTh
DispatchRateLimiter subRateLimiter = null;
Dispatcher subDispatcher = topic.getSubscription(subName).getDispatcher();
if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
- subRateLimiter = ((PersistentDispatcherMultipleConsumers) subDispatcher).getRateLimiter().get();
+ subRateLimiter = subDispatcher.getRateLimiter().get();
} else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
- subRateLimiter = ((PersistentDispatcherSingleActiveConsumer) subDispatcher).getRateLimiter().get();
+ subRateLimiter = subDispatcher.getRateLimiter().get();
} else {
Assert.fail("Should only have PersistentDispatcher in this test");
}
@@ -252,9 +253,9 @@ public class SubscriptionMessageDispatchThrottlingTest extends MessageDispatchTh
DispatchRateLimiter subRateLimiter = null;
Dispatcher subDispatcher = topic.getSubscription(subName).getDispatcher();
if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
- subRateLimiter = ((PersistentDispatcherMultipleConsumers) subDispatcher).getRateLimiter().get();
+ subRateLimiter = subDispatcher.getRateLimiter().get();
} else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
- subRateLimiter = ((PersistentDispatcherSingleActiveConsumer) subDispatcher).getRateLimiter().get();
+ subRateLimiter = subDispatcher.getRateLimiter().get();
} else {
Assert.fail("Should only have PersistentDispatcher in this test");
}
@@ -335,9 +336,9 @@ public class SubscriptionMessageDispatchThrottlingTest extends MessageDispatchTh
DispatchRateLimiter subRateLimiter = null;
Dispatcher subDispatcher = topic.getSubscription(subName).getDispatcher();
if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
- subRateLimiter = ((PersistentDispatcherMultipleConsumers) subDispatcher).getRateLimiter().get();
+ subRateLimiter = subDispatcher.getRateLimiter().get();
} else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
- subRateLimiter = ((PersistentDispatcherSingleActiveConsumer) subDispatcher).getRateLimiter().get();
+ subRateLimiter = subDispatcher.getRateLimiter().get();
} else {
Assert.fail("Should only have PersistentDispatcher in this test");
}
@@ -489,9 +490,9 @@ public class SubscriptionMessageDispatchThrottlingTest extends MessageDispatchTh
DispatchRateLimiter subRateLimiter = null;
Dispatcher subDispatcher = topic.getSubscription(subName1).getDispatcher();
if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
- subRateLimiter = ((PersistentDispatcherMultipleConsumers) subDispatcher).getRateLimiter().get();
+ subRateLimiter = subDispatcher.getRateLimiter().get();
} else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
- subRateLimiter = ((PersistentDispatcherSingleActiveConsumer) subDispatcher).getRateLimiter().get();
+ subRateLimiter = subDispatcher.getRateLimiter().get();
} else {
Assert.fail("Should only have PersistentDispatcher in this test");
}
@@ -504,13 +505,7 @@ public class SubscriptionMessageDispatchThrottlingTest extends MessageDispatchTh
DispatchRate dispatchRate = new DispatchRate(nsMessageRate, 0, 1);
admin.namespaces().setSubscriptionDispatchRate(namespace, dispatchRate);
- if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
- subRateLimiter = ((PersistentDispatcherMultipleConsumers) subDispatcher).getRateLimiter().get();
- } else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
- subRateLimiter = ((PersistentDispatcherSingleActiveConsumer) subDispatcher).getRateLimiter().get();
- } else {
- Assert.fail("Should only have PersistentDispatcher in this test");
- }
+ subRateLimiter = subDispatcher.getRateLimiter().get();
for (int i = 0; i < 5; i++) {
if (subRateLimiter.getDispatchRateOnMsg() != nsMessageRate) {
@@ -537,9 +532,9 @@ public class SubscriptionMessageDispatchThrottlingTest extends MessageDispatchTh
subDispatcher = topic2.getSubscription(subName2).getDispatcher();
if (subDispatcher instanceof PersistentDispatcherMultipleConsumers) {
- subRateLimiter = ((PersistentDispatcherMultipleConsumers) subDispatcher).getRateLimiter().get();
+ subRateLimiter = subDispatcher.getRateLimiter().get();
} else if (subDispatcher instanceof PersistentDispatcherSingleActiveConsumer) {
- subRateLimiter = ((PersistentDispatcherSingleActiveConsumer) subDispatcher).getRateLimiter().get();
+ subRateLimiter = subDispatcher.getRateLimiter().get();
} else {
Assert.fail("Should only have PersistentDispatcher in this test");
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TenantTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TenantTest.java
index 07b17ab..11e4daa 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TenantTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TenantTest.java
@@ -28,6 +28,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "broker-api")
public class TenantTest extends MockedPulsarServiceBaseTest {
@BeforeMethod
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsHostVerification.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsHostVerification.java
index 3aed06a..01b1211 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsHostVerification.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsHostVerification.java
@@ -27,7 +27,9 @@ import org.apache.pulsar.client.impl.auth.AuthenticationTls;
import org.testng.Assert;
import org.testng.annotations.Test;
+@Test(groups = "broker-api")
public class TlsHostVerification extends TlsProducerConsumerBase {
+
@Test
public void testTlsHostVerificationAdminClient() throws Exception {
Map<String, String> authParams = new HashMap<>();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerBase.java
index 9ec369a..e81ca0f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerBase.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerBase.java
@@ -30,11 +30,11 @@ import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.impl.auth.AuthenticationTls;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.TenantInfo;
-import org.testng.annotations.AfterMethod;
-import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.*;
import com.google.common.collect.Sets;
+@Test(groups = "broker-api")
public class TlsProducerConsumerBase extends ProducerConsumerBase {
protected final String TLS_TRUST_CERT_FILE_PATH = "./src/test/resources/authentication/tls/cacert.pem";
protected final String TLS_CLIENT_CERT_FILE_PATH = "./src/test/resources/authentication/tls/client-cert.pem";
@@ -59,7 +59,7 @@ public class TlsProducerConsumerBase extends ProducerConsumerBase {
super.internalCleanup();
}
- protected void internalSetUpForBroker() throws Exception {
+ protected void internalSetUpForBroker() {
conf.setBrokerServicePortTls(Optional.of(0));
conf.setWebServicePortTls(Optional.of(0));
conf.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerTest.java
index e48dcca..78aafbf 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerTest.java
@@ -38,6 +38,7 @@ import org.testng.annotations.Test;
import lombok.Cleanup;
+@Test(groups = "broker-api")
public class TlsProducerConsumerTest extends TlsProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(TlsProducerConsumerTest.class);
@@ -52,7 +53,7 @@ public class TlsProducerConsumerTest extends TlsProducerConsumerBase {
log.info("-- Starting {} test --", methodName);
final int MESSAGE_SIZE = 16 * 1024 + 1;
- log.info("-- message size --", MESSAGE_SIZE);
+ log.info("-- message size -- {}", MESSAGE_SIZE);
internalSetUpForClient(true, pulsar.getBrokerServiceUrlTls());
internalSetUpForNamespace();
@@ -86,7 +87,7 @@ public class TlsProducerConsumerTest extends TlsProducerConsumerBase {
log.info("-- Starting {} test --", methodName);
final int MESSAGE_SIZE = 16 * 1024 + 1;
- log.info("-- message size --", MESSAGE_SIZE);
+ log.info("-- message size -- {}", MESSAGE_SIZE);
internalSetUpForNamespace();
// Test 1 - Using TLS on binary protocol without sending certs - expect failure
@@ -114,7 +115,7 @@ public class TlsProducerConsumerTest extends TlsProducerConsumerBase {
log.info("-- Starting {} test --", methodName);
final int MESSAGE_SIZE = 16 * 1024 + 1;
- log.info("-- message size --", MESSAGE_SIZE);
+ log.info("-- message size -- {}", MESSAGE_SIZE);
internalSetUpForNamespace();
// Test 1 - Using TLS on https without sending certs - expect failure
@@ -232,7 +233,7 @@ public class TlsProducerConsumerTest extends TlsProducerConsumerBase {
} catch (PulsarClientException e) {
// Ok..
}
-
+
trustStoreIndex.set(0);
consumer = pulsarClient.newConsumer().topic("persistent://my-property/use/my-ns/my-topic1")
.subscriptionName("my-subscriber-name").subscribe();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsSniTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsSniTest.java
index fc8c242..c3f7a46 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsSniTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsSniTest.java
@@ -29,6 +29,7 @@ import org.testng.annotations.Test;
import lombok.Cleanup;
+@Test(groups = "broker-api")
public class TlsSniTest extends TlsProducerConsumerBase {
/**
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java
index c94b3e4..51d755a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java
@@ -42,6 +42,7 @@ import org.testng.annotations.Test;
* client: org.apache.pulsar.client.impl.auth.AuthenticationToken
* broker: org.apache.pulsar.broker.authentication.AuthenticationProviderToken
*/
+@Test(groups = "broker-api")
public class TokenAuthenticatedProducerConsumerTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(TokenAuthenticatedProducerConsumerTest.class);
@@ -96,7 +97,7 @@ public class TokenAuthenticatedProducerConsumerTest extends ProducerConsumerBase
return new Object[][] { { 0 }, { 1000 } };
}
- public void testSyncProducerAndConsumer() throws Exception {
+ private void testSyncProducerAndConsumer() throws Exception {
Consumer<byte[]> consumer = pulsarClient.newConsumer().topic("persistent://my-property/my-ns/my-topic")
.subscriptionName("my-subscriber-name").subscribe();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenOauth2AuthenticatedProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenOauth2AuthenticatedProducerConsumerTest.java
index 742e3ab..03bb770 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenOauth2AuthenticatedProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenOauth2AuthenticatedProducerConsumerTest.java
@@ -40,6 +40,7 @@ import org.apache.pulsar.common.policies.data.TenantInfo;
import org.awaitility.Awaitility;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
@@ -50,6 +51,7 @@ import org.testng.annotations.Test;
* client: org.apache.pulsar.client.impl.auth.oauth2.AuthenticationOAuth2
* broker: org.apache.pulsar.broker.authentication.AuthenticationProviderToken
*/
+@Test(groups = "broker-api")
public class TokenOauth2AuthenticatedProducerConsumerTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(TokenOauth2AuthenticatedProducerConsumerTest.class);
@@ -107,7 +109,7 @@ public class TokenOauth2AuthenticatedProducerConsumerTest extends ProducerConsum
.authentication(authentication));
}
- @AfterMethod(alwaysRun = true)
+ @AfterMethod(alwaysRun = true, groups = "broker-api")
@Override
protected void cleanup() throws Exception {
super.internalCleanup();
@@ -118,7 +120,7 @@ public class TokenOauth2AuthenticatedProducerConsumerTest extends ProducerConsum
return new Object[][] { { 0 }, { 1000 } };
}
- public void testSyncProducerAndConsumer() throws Exception {
+ private void testSyncProducerAndConsumer() throws Exception {
Consumer<byte[]> consumer = pulsarClient.newConsumer().topic("persistent://my-property/my-ns/my-topic")
.subscriptionName("my-subscriber-name").subscribe();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicReaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicReaderTest.java
index 0c3b9d3..723249e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicReaderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicReaderTest.java
@@ -63,6 +63,7 @@ import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+@Test(groups = "flaky")
public class TopicReaderTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(TopicReaderTest.class);
@@ -513,13 +514,13 @@ public class TopicReaderTest extends ProducerConsumerBase {
producer.close();
}
- @Test(groups = "encryption")
+ @Test
public void testECDSAEncryption() throws Exception {
log.info("-- Starting {} test --", methodName);
class EncKeyReader implements CryptoKeyReader {
- EncryptionKeyInfo keyInfo = new EncryptionKeyInfo();
+ final EncryptionKeyInfo keyInfo = new EncryptionKeyInfo();
@Override
public EncryptionKeyInfo getPublicKey(String keyName, Map<String, String> keyMeta) {
@@ -583,13 +584,13 @@ public class TopicReaderTest extends ProducerConsumerBase {
log.info("-- Exiting {} test --", methodName);
}
- @Test(groups = "encryption")
+ @Test
public void testMultiReaderECDSAEncryption() throws Exception {
log.info("-- Starting {} test --", methodName);
class EncKeyReader implements CryptoKeyReader {
- EncryptionKeyInfo keyInfo = new EncryptionKeyInfo();
+ final EncryptionKeyInfo keyInfo = new EncryptionKeyInfo();
@Override
public EncryptionKeyInfo getPublicKey(String keyName, Map<String, String> keyMeta) {
@@ -652,7 +653,7 @@ public class TopicReaderTest extends ProducerConsumerBase {
reader.close();
}
- @Test(groups = "encryption")
+ @Test
public void testDefaultCryptoKeyReader() throws Exception {
final String topic = "persistent://my-property/my-ns/test-reader-default-crypto-key-reader"
+ System.currentTimeMillis();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v1/V1_ProducerConsumerBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v1/V1_ProducerConsumerBase.java
index ee8ea84..7a60721 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v1/V1_ProducerConsumerBase.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v1/V1_ProducerConsumerBase.java
@@ -32,7 +32,7 @@ import org.testng.annotations.BeforeMethod;
public abstract class V1_ProducerConsumerBase extends MockedPulsarServiceBaseTest {
protected String methodName;
- @BeforeMethod
+ @BeforeMethod(groups = {"broker-api, websocket"})
public void beforeMethod(Method m) throws Exception {
methodName = m.getName();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v1/V1_ProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v1/V1_ProducerConsumerTest.java
index 6f00f0e..48beb46 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v1/V1_ProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v1/V1_ProducerConsumerTest.java
@@ -86,6 +86,7 @@ import org.testng.annotations.Test;
/**
* Basic tests using the deprecated client APIs from Pulsar-1.x
*/
+@Test(groups = "flaky")
public class V1_ProducerConsumerTest extends V1_ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(V1_ProducerConsumerTest.class);
@@ -360,9 +361,7 @@ public class V1_ProducerConsumerTest extends V1_ProducerConsumerBase {
} catch (PulsarClientException e) {
Assert.assertTrue(e instanceof PulsarClientException.AlreadyClosedException);
}
-
-
-
+
Consumer<byte[]> consumer = pulsarClient.newConsumer()
.topic("persistent://my-property/use/my-ns/my-topic6")
.subscriptionName("my-subscriber-name")
@@ -2077,7 +2076,7 @@ public class V1_ProducerConsumerTest extends V1_ProducerConsumerBase {
log.info("-- Exiting {} test --", methodName);
}
- @Test(groups = "encryption")
+ @Test
public void testECDSAEncryption() throws Exception {
log.info("-- Starting {} test --", methodName);
@@ -2153,7 +2152,7 @@ public class V1_ProducerConsumerTest extends V1_ProducerConsumerBase {
log.info("-- Exiting {} test --", methodName);
}
- @Test(groups = "encryption")
+ @Test
public void testRSAEncryption() throws Exception {
log.info("-- Starting {} test --", methodName);
@@ -2238,7 +2237,7 @@ public class V1_ProducerConsumerTest extends V1_ProducerConsumerBase {
log.info("-- Exiting {} test --", methodName);
}
- @Test(groups = "encryption")
+ @Test
public void testEncryptionFailure() throws Exception {
log.info("-- Starting {} test --", methodName);
@@ -2369,7 +2368,7 @@ public class V1_ProducerConsumerTest extends V1_ProducerConsumerBase {
// Receive should proceed and discard encrypted messages
msg = consumer.receive(5, TimeUnit.SECONDS);
- Assert.assertNull(msg, "Message received even aftet ConsumerCryptoFailureAction.DISCARD is set.");
+ Assert.assertNull(msg, "Message received even after ConsumerCryptoFailureAction.DISCARD is set.");
log.info("-- Exiting {} test --", methodName);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AdminApiKeyStoreTlsAuthTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AdminApiKeyStoreTlsAuthTest.java
index 7752f81..a1c5487 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AdminApiKeyStoreTlsAuthTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AdminApiKeyStoreTlsAuthTest.java
@@ -19,7 +19,7 @@
package org.apache.pulsar.client.impl;
import static org.apache.pulsar.client.impl.auth.AuthenticationKeyStoreTls.mapToString;
-import static org.testng.Assert.fail;
+import static org.testng.AssertJUnit.fail;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Sets;
@@ -60,7 +60,9 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@Slf4j
+@Test(groups = "broker-impl")
public class AdminApiKeyStoreTlsAuthTest extends ProducerConsumerBase {
+
protected final String BROKER_KEYSTORE_FILE_PATH =
"./src/test/resources/authentication/keystoretls/broker.keystore.jks";
protected final String BROKER_TRUSTSTORE_FILE_PATH =
@@ -190,7 +192,7 @@ public class AdminApiKeyStoreTlsAuthTest extends ProducerConsumerBase {
new TenantInfo(ImmutableSet.of("proxy"),
ImmutableSet.of("test")));
admin.namespaces().createNamespace("tenant1/ns1");
- admin.namespaces().getNamespaces("tenant1").contains("tenant1/ns1");
+ Assert.assertTrue(admin.namespaces().getNamespaces("tenant1").contains("tenant1/ns1"));
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BatchMessageIdImplSerializationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BatchMessageIdImplSerializationTest.java
index 2f2d280..56d32e1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BatchMessageIdImplSerializationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BatchMessageIdImplSerializationTest.java
@@ -22,13 +22,11 @@ import static org.testng.Assert.assertEquals;
import java.io.IOException;
-import org.apache.pulsar.client.impl.BatchMessageIdImpl;
import org.testng.annotations.Test;
-/**
- */
-
+@Test(groups = "broker-impl")
public class BatchMessageIdImplSerializationTest {
+
@Test
public void testSerialization1() throws Exception {
BatchMessageIdImpl id = new BatchMessageIdImpl(1, 2, 3, 4);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BatchMessageIndexAckDisableTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BatchMessageIndexAckDisableTest.java
index 46f11fb..da94689 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BatchMessageIndexAckDisableTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BatchMessageIndexAckDisableTest.java
@@ -42,6 +42,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
@Slf4j
+@Test(groups = "broker-impl")
public class BatchMessageIndexAckDisableTest extends ProducerConsumerBase {
@BeforeMethod
@@ -64,7 +65,8 @@ public class BatchMessageIndexAckDisableTest extends ProducerConsumerBase {
}
@Test(dataProvider = "ackReceiptEnabled")
- public void testBatchMessageIndexAckForSharedSubscription(boolean ackReceiptEnabled) throws PulsarClientException, ExecutionException, InterruptedException {
+ public void testBatchMessageIndexAckForSharedSubscription(boolean ackReceiptEnabled) throws
+ PulsarClientException, ExecutionException, InterruptedException {
final String topic = "testBatchMessageIndexAckForSharedSubscription";
@Cleanup
@@ -105,7 +107,8 @@ public class BatchMessageIndexAckDisableTest extends ProducerConsumerBase {
}
@Test(dataProvider = "ackReceiptEnabled")
- public void testBatchMessageIndexAckForExclusiveSubscription(boolean ackReceiptEnabled) throws PulsarClientException, ExecutionException, InterruptedException {
+ public void testBatchMessageIndexAckForExclusiveSubscription(boolean ackReceiptEnabled) throws
+ PulsarClientException, ExecutionException, InterruptedException {
final String topic = "testBatchMessageIndexAckForExclusiveSubscription";
@Cleanup
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BatchMessageIndexAckTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BatchMessageIndexAckTest.java
index dee8c4c..c9bdb75 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BatchMessageIndexAckTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BatchMessageIndexAckTest.java
@@ -51,6 +51,7 @@ import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.doReturn;
@Slf4j
+@Test(groups = "broker-impl")
public class BatchMessageIndexAckTest extends ProducerConsumerBase {
@BeforeMethod
@@ -245,7 +246,8 @@ public class BatchMessageIndexAckTest extends ProducerConsumerBase {
}
@Test(dataProvider = "ackReceiptEnabled")
- public void testBatchMessageIndexAckForExclusiveSubscription(boolean ackReceiptEnabled) throws PulsarClientException, ExecutionException, InterruptedException {
+ public void testBatchMessageIndexAckForExclusiveSubscription(boolean ackReceiptEnabled) throws
+ PulsarClientException, ExecutionException, InterruptedException {
final String topic = "testBatchMessageIndexAckForExclusiveSubscription";
@Cleanup
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
index 5fddeba..14fce51 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
@@ -110,6 +110,7 @@ import org.testng.annotations.Test;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
+@Test(groups = "broker-impl")
public class BrokerClientIntegrationTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(BrokerClientIntegrationTest.class);
@@ -257,7 +258,6 @@ public class BrokerClientIntegrationTest extends ProducerConsumerBase {
prod1.close();
prod2.close();
cons1.close();
-
}
/**
@@ -782,13 +782,12 @@ public class BrokerClientIntegrationTest extends ProducerConsumerBase {
.getTopics();
// non-complete topic future so, create topic should timeout
topics.put(topicName, new CompletableFuture<>());
- PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(lookupUrl.toString())
- .operationTimeout(2, TimeUnit.SECONDS).statsInterval(0, TimeUnit.SECONDS).build();
- try {
+ try (PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(lookupUrl.toString())
+ .operationTimeout(2, TimeUnit.SECONDS).statsInterval(0, TimeUnit.SECONDS).build()) {
+
Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
} finally {
topics.clear();
- pulsarClient.close();
}
}
@@ -869,13 +868,13 @@ public class BrokerClientIntegrationTest extends ProducerConsumerBase {
.withSchemaWriter(writer)
.build();
Schema<TestMessageObject> schema = Schema.AVRO(schemaDefinition);
- PulsarClient client = PulsarClient.builder()
+
+ try (PulsarClient client = PulsarClient.builder()
.serviceUrl(lookupUrl.toString())
- .build();
+ .build(); Producer<TestMessageObject> producer = client.newProducer(schema).topic(topicName).create();
+ Consumer<TestMessageObject> consumer =
+ client.newConsumer(schema).topic(topicName).subscriptionName("my-subscriber-name").subscribe()) {
- try(Producer<TestMessageObject> producer = client.newProducer(schema).topic(topicName).create();
- Consumer<TestMessageObject> consumer =
- client.newConsumer(schema).topic(topicName).subscriptionName("my-subscriber-name").subscribe()) {
assertNotNull(producer);
assertNotNull(consumer);
producer.newMessage().value(object).send();
@@ -883,8 +882,6 @@ public class BrokerClientIntegrationTest extends ProducerConsumerBase {
Assert.assertEquals(object.getValue(), testObject.getValue());
Mockito.verify(writer, Mockito.times(1)).write(Mockito.any());
Mockito.verify(reader, Mockito.times(1)).read(Mockito.any(byte[].class), Mockito.any(byte[].class));
- } finally {
- client.close();
}
}
@@ -901,12 +898,12 @@ public class BrokerClientIntegrationTest extends ProducerConsumerBase {
.withSchemaWriter(writer)
.build();
Schema<TestMessageObject> schema = Schema.JSON(schemaDefinition);
- PulsarClient client = PulsarClient.builder()
+
+ try (PulsarClient client = PulsarClient.builder()
.serviceUrl(lookupUrl.toString())
- .build();
+ .build(); Producer<TestMessageObject> producer = client.newProducer(schema).topic(topicName).create();
+ Consumer<TestMessageObject> consumer = client.newConsumer(schema).topic(topicName).subscriptionName("my-subscriber-name").subscribe()) {
- try(Producer<TestMessageObject> producer = client.newProducer(schema).topic(topicName).create();
- Consumer<TestMessageObject> consumer = client.newConsumer(schema).topic(topicName).subscriptionName("my-subscriber-name").subscribe()) {
assertNotNull(producer);
assertNotNull(consumer);
@@ -919,8 +916,6 @@ public class BrokerClientIntegrationTest extends ProducerConsumerBase {
Mockito.verify(writer, Mockito.times(1)).write(Mockito.any());
Mockito.verify(reader, Mockito.times(1)).read(Mockito.any(byte[].class));
- } finally {
- client.close();
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/CompactedOutBatchMessageTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/CompactedOutBatchMessageTest.java
index 9de071f..f074847 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/CompactedOutBatchMessageTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/CompactedOutBatchMessageTest.java
@@ -33,7 +33,9 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "broker-impl")
public class CompactedOutBatchMessageTest extends ProducerConsumerBase {
+
@BeforeMethod
@Override
protected void setup() throws Exception {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConnectionPoolTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConnectionPoolTest.java
index a375eb8..2be860c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConnectionPoolTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConnectionPoolTest.java
@@ -36,6 +36,7 @@ import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.stream.IntStream;
+@Test(groups = "broker-impl")
public class ConnectionPoolTest extends MockedPulsarServiceBaseTest {
String serviceUrl;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumeBaseExceptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumeBaseExceptionTest.java
index 7bcaf4c..76a7f7e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumeBaseExceptionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumeBaseExceptionTest.java
@@ -26,6 +26,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "broker-impl")
public class ConsumeBaseExceptionTest extends ProducerConsumerBase {
@BeforeMethod
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerAckResponseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerAckResponseTest.java
index f21abaa..203b3a1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerAckResponseTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerAckResponseTest.java
@@ -39,7 +39,7 @@ import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
-
+@Test(groups = "broker-impl")
public class ConsumerAckResponseTest extends ProducerConsumerBase {
private final static TransactionImpl transaction = mock(TransactionImpl.class);
@@ -63,7 +63,7 @@ public class ConsumerAckResponseTest extends ProducerConsumerBase {
}
@Test
- public void testAckResponse() throws PulsarClientException, InterruptedException, ExecutionException {
+ public void testAckResponse() throws PulsarClientException, InterruptedException {
String topic = "testAckResponse";
@Cleanup
Producer<Integer> producer = pulsarClient.newProducer(Schema.INT32)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerConfigurationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerConfigurationTest.java
index fe2d4d1..5feef1b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerConfigurationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerConfigurationTest.java
@@ -29,6 +29,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "broker-impl")
public class ConsumerConfigurationTest extends MockedPulsarServiceBaseTest {
private static String persistentTopic = "persistent://my-property/use/my-ns/persist";
private static String nonPersistentTopic = "non-persistent://my-property/use/my-ns/nopersist";
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerDedupPermitsUpdateTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerDedupPermitsUpdateTest.java
new file mode 100644
index 0000000..4c9922a
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerDedupPermitsUpdateTest.java
@@ -0,0 +1,126 @@
+/**
+ * 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.client.impl;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
+
+import java.util.concurrent.TimeUnit;
+
+import lombok.Cleanup;
+
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.Schema;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-impl")
+public class ConsumerDedupPermitsUpdateTest extends ProducerConsumerBase {
+
+ @BeforeClass
+ @Override
+ protected void setup() throws Exception {
+ super.internalSetup();
+ producerBaseSetup();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @DataProvider(name = "combinations")
+ public Object[][] combinations() {
+ return new Object[][] {
+ // batching-enabled - queue-size
+ { false, 0 },
+ { false, 1 },
+ { false, 10 },
+ { false, 100 },
+ { true, 1 },
+ { true, 10 },
+ { true, 100 },
+ };
+ }
+
+ @Test(timeOut = 30000, dataProvider = "combinations")
+ public void testConsumerDedup(boolean batchingEnabled, int receiverQueueSize) throws Exception {
+ String topic = BrokerTestUtil.newUniqueName("persistent://my-property/my-ns/my-topic");
+
+ @Cleanup
+ Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic(topic)
+ .subscriptionName("test")
+ // Use high ack delay to simulate a message being tracked as dup
+ .acknowledgmentGroupTime(1, TimeUnit.HOURS)
+ .receiverQueueSize(receiverQueueSize)
+ .subscribe();
+
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topic)
+ .enableBatching(batchingEnabled)
+ .batchingMaxMessages(10)
+ .batchingMaxPublishDelay(1, TimeUnit.HOURS)
+ .create();
+
+ for (int i = 0; i < 30; i++) {
+ producer.sendAsync("hello-" + i);
+ }
+ producer.flush();
+
+ // Consumer receives and acks all the messages, though the acks
+ // are still cached in client lib
+ for (int i = 0; i < 30; i++) {
+ Message<String> msg = consumer.receive();
+ assertEquals(msg.getValue(), "hello-" + i);
+ consumer.acknowledge(msg);
+ }
+
+ // Trigger redelivery by unloading the topic.
+ admin.topics().unload(topic);
+
+ // Consumer dedup logic will detect the dups and not bubble them up to the application
+ // (With zero-queue we cannot use receive with timeout)
+ if (receiverQueueSize > 0) {
+ Message<String> msg = consumer.receive(100, TimeUnit.MILLISECONDS);
+ assertNull(msg);
+ }
+
+ // The flow permits in consumer shouldn't have been messed up by the deduping
+ // and we should be able to get new messages through
+ for (int i = 0; i < 30; i++) {
+ producer.sendAsync("new-message-" + i);
+ }
+ producer.flush();
+
+ for (int i = 0; i < 30; i++) {
+ Message<String> msg = consumer.receive();
+ assertEquals(msg.getValue(), "new-message-" + i);
+ consumer.acknowledge(msg);
+ }
+ }
+
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerUnsubscribeTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerUnsubscribeTest.java
index 0d8fcc9..70ed1ec 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerUnsubscribeTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerUnsubscribeTest.java
@@ -27,8 +27,7 @@ import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
-/**
- */
+@Test(groups = "broker-impl")
public class ConsumerUnsubscribeTest {
MockBrokerService mockBrokerService;
@@ -39,7 +38,7 @@ public class ConsumerUnsubscribeTest {
mockBrokerService.start();
}
- @AfterClass(alwaysRun = true)
+ @AfterClass
public void teardown() {
mockBrokerService.stop();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeyStoreTlsProducerConsumerTestWithAuthTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeyStoreTlsProducerConsumerTestWithAuthTest.java
new file mode 100644
index 0000000..e0f7c9a
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeyStoreTlsProducerConsumerTestWithAuthTest.java
@@ -0,0 +1,270 @@
+/**
+ * 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.client.impl;
+
+import static org.mockito.Mockito.spy;
+
+import com.google.common.collect.Sets;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.authentication.AuthenticationProviderTls;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.ClientBuilder;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.auth.AuthenticationKeyStoreTls;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.TenantInfo;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+// TLS authentication and authorization based on KeyStore type config.
+@Slf4j
+@Test(groups = "broker-impl")
+public class KeyStoreTlsProducerConsumerTestWithAuthTest extends ProducerConsumerBase {
+
+ protected final String BROKER_KEYSTORE_FILE_PATH =
+ "./src/test/resources/authentication/keystoretls/broker.keystore.jks";
+ protected final String BROKER_TRUSTSTORE_FILE_PATH =
+ "./src/test/resources/authentication/keystoretls/broker.truststore.jks";
+ protected final String BROKER_KEYSTORE_PW = "111111";
+ protected final String BROKER_TRUSTSTORE_PW = "111111";
+
+ protected final String CLIENT_KEYSTORE_FILE_PATH =
+ "./src/test/resources/authentication/keystoretls/client.keystore.jks";
+ protected final String CLIENT_TRUSTSTORE_FILE_PATH =
+ "./src/test/resources/authentication/keystoretls/client.truststore.jks";
+ protected final String CLIENT_KEYSTORE_PW = "111111";
+ protected final String CLIENT_TRUSTSTORE_PW = "111111";
+
+ protected final String CLIENT_KEYSTORE_CN = "clientuser";
+ protected final String KEYSTORE_TYPE = "JKS";
+
+ private final String clusterName = "use";
+
+ @BeforeMethod
+ @Override
+ protected void setup() throws Exception {
+ // TLS configuration for Broker
+ internalSetUpForBroker();
+
+ // Start Broker
+
+ super.init();
+ }
+
+ @AfterMethod(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ protected void internalSetUpForBroker() {
+ conf.setBrokerServicePortTls(Optional.of(0));
+ conf.setWebServicePortTls(Optional.of(0));
+ conf.setTlsEnabledWithKeyStore(true);
+
+ conf.setTlsKeyStoreType(KEYSTORE_TYPE);
+ conf.setTlsKeyStore(BROKER_KEYSTORE_FILE_PATH);
+ conf.setTlsKeyStorePassword(BROKER_KEYSTORE_PW);
+
+ conf.setTlsTrustStoreType(KEYSTORE_TYPE);
+ conf.setTlsTrustStore(CLIENT_TRUSTSTORE_FILE_PATH);
+ conf.setTlsTrustStorePassword(CLIENT_TRUSTSTORE_PW);
+
+ conf.setClusterName(clusterName);
+ conf.setTlsRequireTrustedClientCertOnConnect(true);
+
+ // config for authentication and authorization.
+ conf.setSuperUserRoles(Sets.newHashSet(CLIENT_KEYSTORE_CN));
+ conf.setAuthenticationEnabled(true);
+ conf.setAuthorizationEnabled(true);
+ Set<String> providers = new HashSet<>();
+ providers.add(AuthenticationProviderTls.class.getName());
+ conf.setAuthenticationProviders(providers);
+ conf.setNumExecutorThreadPoolSize(5);
+ }
+
+ protected void internalSetUpForClient(boolean addCertificates, String lookupUrl) throws Exception {
+ if (pulsarClient != null) {
+ pulsarClient.close();
+ }
+
+ Set<String> tlsProtocols = Sets.newConcurrentHashSet();
+ tlsProtocols.add("TLSv1.2");
+
+ ClientBuilder clientBuilder = PulsarClient.builder().serviceUrl(lookupUrl)
+ .enableTls(true)
+ .useKeyStoreTls(true)
+ .tlsTrustStorePath(BROKER_TRUSTSTORE_FILE_PATH)
+ .tlsTrustStorePassword(BROKER_TRUSTSTORE_PW)
+ .allowTlsInsecureConnection(false)
+ .tlsProtocols(tlsProtocols)
+ .operationTimeout(1000, TimeUnit.MILLISECONDS);
+ if (addCertificates) {
+ Map<String, String> authParams = new HashMap<>();
+ authParams.put(AuthenticationKeyStoreTls.KEYSTORE_TYPE, KEYSTORE_TYPE);
+ authParams.put(AuthenticationKeyStoreTls.KEYSTORE_PATH, CLIENT_KEYSTORE_FILE_PATH);
+ authParams.put(AuthenticationKeyStoreTls.KEYSTORE_PW, CLIENT_KEYSTORE_PW);
+ clientBuilder.authentication(AuthenticationKeyStoreTls.class.getName(), authParams);
+ }
+ replacePulsarClient(clientBuilder);
+ }
+
+ protected void internalSetUpForNamespace() throws Exception {
+ Map<String, String> authParams = new HashMap<>();
+ authParams.put(AuthenticationKeyStoreTls.KEYSTORE_PATH, CLIENT_KEYSTORE_FILE_PATH);
+ authParams.put(AuthenticationKeyStoreTls.KEYSTORE_PW, CLIENT_KEYSTORE_PW);
+
+ if (admin != null) {
+ admin.close();
+ }
+
+ admin = spy(PulsarAdmin.builder().serviceHttpUrl(brokerUrlTls.toString())
+ .useKeyStoreTls(true)
+ .tlsTrustStorePath(BROKER_TRUSTSTORE_FILE_PATH)
+ .tlsTrustStorePassword(BROKER_TRUSTSTORE_PW)
+ .allowTlsInsecureConnection(false)
+ .authentication(AuthenticationKeyStoreTls.class.getName(), authParams).build());
+ admin.clusters().createCluster(clusterName, new ClusterData(brokerUrl.toString(), brokerUrlTls.toString(),
+ pulsar.getBrokerServiceUrl(), pulsar.getBrokerServiceUrlTls()));
+ admin.tenants().createTenant("my-property",
+ new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("use")));
+ admin.namespaces().createNamespace("my-property/my-ns");
+ }
+
+ /**
+ * verifies that messages whose size is larger than 2^14 bytes (max size of single TLS chunk) can be
+ * produced/consumed
+ *
+ * @throws Exception
+ */
+ @Test(timeOut = 30000)
+ public void testTlsLargeSizeMessage() throws Exception {
+ log.info("-- Starting {} test --", methodName);
+
+ final int MESSAGE_SIZE = 16 * 1024 + 1;
+ log.info("-- message size -- {}", MESSAGE_SIZE);
+ String topicName = "persistent://my-property/use/my-ns/testTlsLargeSizeMessage"
+ + System.currentTimeMillis();
+
+ internalSetUpForClient(true, pulsar.getBrokerServiceUrlTls());
+ internalSetUpForNamespace();
+
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName)
+ .subscriptionName("my-subscriber-name").subscribe();
+
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName)
+ .create();
+ for (int i = 0; i < 10; i++) {
+ byte[] message = new byte[MESSAGE_SIZE];
+ Arrays.fill(message, (byte) i);
+ producer.send(message);
+ }
+
+ Message<byte[]> msg = null;
+ for (int i = 0; i < 10; i++) {
+ msg = consumer.receive(5, TimeUnit.SECONDS);
+ byte[] expected = new byte[MESSAGE_SIZE];
+ Arrays.fill(expected, (byte) i);
+ Assert.assertEquals(expected, msg.getData());
+ }
+ // Acknowledge the consumption of all messages at once
+ consumer.acknowledgeCumulative(msg);
+ consumer.close();
+ log.info("-- Exiting {} test --", methodName);
+ }
+
+ @Test
+ public void testTlsClientAuthOverBinaryProtocol() throws Exception {
+ log.info("-- Starting {} test --", methodName);
+
+ final int MESSAGE_SIZE = 16 * 1024 + 1;
+ log.info("-- message size -- {}", MESSAGE_SIZE);
+ String topicName = "persistent://my-property/use/my-ns/testTlsClientAuthOverBinaryProtocol"
+ + System.currentTimeMillis();
+
+ internalSetUpForNamespace();
+
+ // Test 1 - Using TLS on binary protocol without sending certs - expect failure
+ internalSetUpForClient(false, pulsar.getBrokerServiceUrlTls());
+
+ try {
+ pulsarClient.newConsumer().topic(topicName)
+ .subscriptionName("my-subscriber-name").subscriptionType(SubscriptionType.Exclusive).subscribe();
+ Assert.fail("Server should have failed the TLS handshake since client didn't .");
+ } catch (Exception ex) {
+ // OK
+ }
+
+ // Using TLS on binary protocol - sending certs
+ internalSetUpForClient(true, pulsar.getBrokerServiceUrlTls());
+
+ // Should not fail since certs are sent
+ pulsarClient.newConsumer()
+ .topic(topicName)
+ .subscriptionName("my-subscriber-name")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribe();
+ }
+
+ @Test
+ public void testTlsClientAuthOverHTTPProtocol() throws Exception {
+ log.info("-- Starting {} test --", methodName);
+
+ final int MESSAGE_SIZE = 16 * 1024 + 1;
+ log.info("-- message size -- {}", MESSAGE_SIZE);
+ String topicName = "persistent://my-property/use/my-ns/testTlsClientAuthOverHTTPProtocol"
+ + System.currentTimeMillis();
+
+ internalSetUpForNamespace();
+
+ // Test 1 - Using TLS on https without sending certs - expect failure
+ internalSetUpForClient(false, pulsar.getWebServiceAddressTls());
+ try {
+ pulsarClient.newConsumer().topic(topicName)
+ .subscriptionName("my-subscriber-name").subscriptionType(SubscriptionType.Exclusive).subscribe();
+ Assert.fail("Server should have failed the TLS handshake since client didn't .");
+ } catch (Exception ex) {
+ // OK
+ }
+
+ // Test 2 - Using TLS on https - sending certs
+ internalSetUpForClient(true, pulsar.getWebServiceAddressTls());
+ // Should not fail since certs are sent
+ pulsarClient.newConsumer()
+ .topic(topicName)
+ .subscriptionName("my-subscriber-name")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribe();
+ }
+
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeyStoreTlsProducerConsumerTestWithoutAuthTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeyStoreTlsProducerConsumerTestWithoutAuthTest.java
new file mode 100644
index 0000000..38a9aa2
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeyStoreTlsProducerConsumerTestWithoutAuthTest.java
@@ -0,0 +1,255 @@
+/**
+ * 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.client.impl;
+
+import static org.mockito.Mockito.spy;
+
+import com.google.common.collect.Sets;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.ClientBuilder;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.auth.AuthenticationKeyStoreTls;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.TenantInfo;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+// TLS test without authentication and authorization based on KeyStore type config.
+@Slf4j
+@Test(groups = "broker-impl")
+public class KeyStoreTlsProducerConsumerTestWithoutAuthTest extends ProducerConsumerBase {
+ protected final String BROKER_KEYSTORE_FILE_PATH =
+ "./src/test/resources/authentication/keystoretls/broker.keystore.jks";
+ protected final String BROKER_TRUSTSTORE_FILE_PATH =
+ "./src/test/resources/authentication/keystoretls/broker.truststore.jks";
+ protected final String BROKER_KEYSTORE_PW = "111111";
+ protected final String BROKER_TRUSTSTORE_PW = "111111";
+
+ protected final String CLIENT_KEYSTORE_FILE_PATH =
+ "./src/test/resources/authentication/keystoretls/client.keystore.jks";
+ protected final String CLIENT_TRUSTSTORE_FILE_PATH =
+ "./src/test/resources/authentication/keystoretls/client.truststore.jks";
+ protected final String CLIENT_KEYSTORE_PW = "111111";
+ protected final String CLIENT_TRUSTSTORE_PW = "111111";
+
+ protected final String KEYSTORE_TYPE = "JKS";
+
+ private final String clusterName = "use";
+ Set<String> tlsProtocols = Sets.newConcurrentHashSet();
+
+ @BeforeMethod
+ @Override
+ protected void setup() throws Exception {
+ // TLS configuration for Broker
+ internalSetUpForBroker();
+
+ // Start Broker
+ super.init();
+ }
+
+ @AfterMethod(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ protected void internalSetUpForBroker() {
+ conf.setBrokerServicePortTls(Optional.of(0));
+ conf.setWebServicePortTls(Optional.of(0));
+ conf.setTlsEnabledWithKeyStore(true);
+
+ conf.setTlsKeyStoreType(KEYSTORE_TYPE);
+ conf.setTlsKeyStore(BROKER_KEYSTORE_FILE_PATH);
+ conf.setTlsKeyStorePassword(BROKER_KEYSTORE_PW);
+
+ conf.setTlsTrustStoreType(KEYSTORE_TYPE);
+ conf.setTlsTrustStore(CLIENT_TRUSTSTORE_FILE_PATH);
+ conf.setTlsTrustStorePassword(CLIENT_TRUSTSTORE_PW);
+
+ conf.setClusterName(clusterName);
+ conf.setTlsRequireTrustedClientCertOnConnect(true);
+ tlsProtocols.add("TLSv1.2");
+ conf.setTlsProtocols(tlsProtocols);
+ conf.setNumExecutorThreadPoolSize(5);
+ }
+
+ protected void internalSetUpForClient(boolean addCertificates, String lookupUrl) throws Exception {
+ if (pulsarClient != null) {
+ pulsarClient.close();
+ }
+
+ ClientBuilder clientBuilder = PulsarClient.builder().serviceUrl(lookupUrl)
+ .enableTls(true)
+ .useKeyStoreTls(true)
+ .tlsTrustStorePath(BROKER_TRUSTSTORE_FILE_PATH)
+ .tlsTrustStorePassword(BROKER_TRUSTSTORE_PW)
+ .allowTlsInsecureConnection(false)
+ .operationTimeout(1000, TimeUnit.MILLISECONDS);
+ if (addCertificates) {
+ Map<String, String> authParams = new HashMap<>();
+ authParams.put(AuthenticationKeyStoreTls.KEYSTORE_TYPE, KEYSTORE_TYPE);
+ authParams.put(AuthenticationKeyStoreTls.KEYSTORE_PATH, CLIENT_KEYSTORE_FILE_PATH);
+ authParams.put(AuthenticationKeyStoreTls.KEYSTORE_PW, CLIENT_KEYSTORE_PW);
+ clientBuilder.authentication(AuthenticationKeyStoreTls.class.getName(), authParams);
+ }
+ replacePulsarClient(clientBuilder);
+ }
+
+ protected void internalSetUpForNamespace() throws Exception {
+ Map<String, String> authParams = new HashMap<>();
+ authParams.put(AuthenticationKeyStoreTls.KEYSTORE_PATH, CLIENT_KEYSTORE_FILE_PATH);
+ authParams.put(AuthenticationKeyStoreTls.KEYSTORE_PW, CLIENT_KEYSTORE_PW);
+
+ if (admin != null) {
+ admin.close();
+ }
+
+ admin = spy(PulsarAdmin.builder().serviceHttpUrl(brokerUrlTls.toString())
+ .useKeyStoreTls(true)
+ .tlsTrustStorePath(BROKER_TRUSTSTORE_FILE_PATH)
+ .tlsTrustStorePassword(BROKER_TRUSTSTORE_PW)
+ .allowTlsInsecureConnection(true)
+ .authentication(AuthenticationKeyStoreTls.class.getName(), authParams).build());
+ admin.clusters().createCluster(clusterName, new ClusterData(brokerUrl.toString(), brokerUrlTls.toString(),
+ pulsar.getBrokerServiceUrl(), pulsar.getBrokerServiceUrlTls()));
+ admin.tenants().createTenant("my-property",
+ new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("use")));
+ admin.namespaces().createNamespace("my-property/my-ns");
+ }
+
+ /**
+ * verifies that messages whose size is larger than 2^14 bytes (max size of single TLS chunk) can be
+ * produced/consumed
+ *
+ * @throws Exception
+ */
+ @Test(timeOut = 30000)
+ public void testTlsLargeSizeMessage() throws Exception {
+ log.info("-- Starting {} test --", methodName);
+
+ final int MESSAGE_SIZE = 16 * 1024 + 1;
+ log.info("-- message size -- {}", MESSAGE_SIZE);
+ String topicName = "persistent://my-property/use/my-ns/testTlsLargeSizeMessage"
+ + System.currentTimeMillis();
+
+ internalSetUpForClient(true, pulsar.getBrokerServiceUrlTls());
+ internalSetUpForNamespace();
+
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName)
+ .subscriptionName("my-subscriber-name").subscribe();
+
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName)
+ .create();
+ for (int i = 0; i < 10; i++) {
+ byte[] message = new byte[MESSAGE_SIZE];
+ Arrays.fill(message, (byte) i);
+ producer.send(message);
+ }
+
+ Message<byte[]> msg = null;
+ for (int i = 0; i < 10; i++) {
+ msg = consumer.receive(5, TimeUnit.SECONDS);
+ byte[] expected = new byte[MESSAGE_SIZE];
+ Arrays.fill(expected, (byte) i);
+ Assert.assertEquals(expected, msg.getData());
+ }
+ // Acknowledge the consumption of all messages at once
+ consumer.acknowledgeCumulative(msg);
+ consumer.close();
+ log.info("-- Exiting {} test --", methodName);
+ }
+
+ @Test(timeOut = 300000)
+ public void testTlsClientAuthOverBinaryProtocol() throws Exception {
+ log.info("-- Starting {} test --", methodName);
+
+ final int MESSAGE_SIZE = 16 * 1024 + 1;
+ log.info("-- message size -- {}", MESSAGE_SIZE);
+ String topicName = "persistent://my-property/use/my-ns/testTlsClientAuthOverBinaryProtocol"
+ + System.currentTimeMillis();
+
+ internalSetUpForNamespace();
+
+ // Test 1 - Using TLS on binary protocol without sending certs - expect failure
+ internalSetUpForClient(false, pulsar.getBrokerServiceUrlTls());
+
+ try {
+ pulsarClient.newConsumer().topic(topicName)
+ .subscriptionName("my-subscriber-name").subscriptionType(SubscriptionType.Exclusive).subscribe();
+ Assert.fail("Server should have failed the TLS handshake since client didn't .");
+ } catch (Exception ex) {
+ // OK
+ }
+
+ // Test 2 - Using TLS on binary protocol - sending certs
+ internalSetUpForClient(true, pulsar.getBrokerServiceUrlTls());
+ try {
+ pulsarClient.newConsumer().topic(topicName)
+ .subscriptionName("my-subscriber-name").subscriptionType(SubscriptionType.Exclusive).subscribe();
+ } catch (Exception ex) {
+ Assert.fail("Should not fail since certs are sent.");
+ }
+ }
+
+ @Test(timeOut = 30000)
+ public void testTlsClientAuthOverHTTPProtocol() throws Exception {
+ log.info("-- Starting {} test --", methodName);
+
+ final int MESSAGE_SIZE = 16 * 1024 + 1;
+ log.info("-- message size -- {}", MESSAGE_SIZE);
+ String topicName = "persistent://my-property/use/my-ns/testTlsClientAuthOverHTTPProtocol"
+ + System.currentTimeMillis();
+
+ internalSetUpForNamespace();
+
+ // Test 1 - Using TLS on https without sending certs - expect failure
+ internalSetUpForClient(false, pulsar.getWebServiceAddressTls());
+ try {
+ pulsarClient.newConsumer().topic(topicName)
+ .subscriptionName("my-subscriber-name").subscriptionType(SubscriptionType.Exclusive).subscribe();
+ Assert.fail("Server should have failed the TLS handshake since client didn't .");
+ } catch (Exception ex) {
+ // OK
+ }
+
+ // Test 2 - Using TLS on https - sending certs
+ internalSetUpForClient(true, pulsar.getWebServiceAddressTls());
+ try {
+ pulsarClient.newConsumer().topic(topicName)
+ .subscriptionName("my-subscriber-name").subscriptionType(SubscriptionType.Exclusive).subscribe();
+ } catch (Exception ex) {
+ Assert.fail("Should not fail since certs are sent.");
+ }
+ }
+
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeyStoreTlsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeyStoreTlsTest.java
index 0f9993d..68fcd8a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeyStoreTlsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeyStoreTlsTest.java
@@ -26,6 +26,7 @@ import org.apache.pulsar.common.util.keystoretls.KeyStoreSSLContext;
import org.apache.pulsar.common.util.keystoretls.SSLContextValidatorEngine;
import org.testng.annotations.Test;
+@Test(groups = "broker-impl")
public class KeyStoreTlsTest {
protected final String BROKER_KEYSTORE_FILE_PATH =
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChecksumTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChecksumTest.java
index f57fe6e..3eed343 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChecksumTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChecksumTest.java
@@ -50,6 +50,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "broker-impl")
public class MessageChecksumTest extends BrokerTestBase {
private static final Logger log = LoggerFactory.getLogger(MessageChecksumTest.class);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java
index 0f8da89..aa977b9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java
@@ -61,6 +61,7 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+@Test(groups = "broker-impl")
public class MessageChunkingTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(MessageChunkingTest.class);
@@ -346,7 +347,7 @@ public class MessageChunkingTest extends ProducerConsumerBase {
TypedMessageBuilderImpl<byte[]> msg = (TypedMessageBuilderImpl<byte[]>) producer.newMessage().value("message-1".getBytes());
ByteBuf payload = Unpooled.wrappedBuffer(msg.getContent());
- MessageMetadata msgMetadata = ((TypedMessageBuilderImpl<byte[]>) msg).getMetadataBuilder();
+ MessageMetadata msgMetadata = msg.getMetadataBuilder();
msgMetadata.setProducerName("test").setSequenceId(1).setPublishTime(10L)
.setUuid("123").setNumChunksFromMsg(2).setChunkId(0).setTotalChunkMsgSize(100);
ByteBufPair cmd = Commands.newSend(producerId, 1, 1, ChecksumType.Crc32c, msgMetadata, payload);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageIdTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageIdTest.java
index 7162592..1ec734b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageIdTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageIdTest.java
@@ -42,6 +42,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test
public class MessageIdTest extends BrokerTestBase {
private static final Logger log = LoggerFactory.getLogger(MessageIdTest.class);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageParserTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageParserTest.java
index 2298d2c..fe2e8f7 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageParserTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageParserTest.java
@@ -45,6 +45,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "broker-impl")
public class MessageParserTest extends MockedPulsarServiceBaseTest {
@BeforeMethod
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessagePublishThrottlingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessagePublishThrottlingTest.java
index 7df325f..281445c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessagePublishThrottlingTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessagePublishThrottlingTest.java
@@ -41,6 +41,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test
public class MessagePublishThrottlingTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(MessagePublishThrottlingTest.class);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageRedeliveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageRedeliveryTest.java
index c257686..6840db9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageRedeliveryTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageRedeliveryTest.java
@@ -52,6 +52,7 @@ import com.google.common.collect.Sets;
import io.netty.util.concurrent.DefaultThreadFactory;
+@Test(groups = "broker-impl")
public class MessageRedeliveryTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(MessageRedeliveryTest.class);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MultiTopicsReaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MultiTopicsReaderTest.java
index 6359701..a12e9a6 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MultiTopicsReaderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MultiTopicsReaderTest.java
@@ -27,7 +27,6 @@ import com.google.common.collect.Sets;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -56,6 +55,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "flaky")
public class MultiTopicsReaderTest extends MockedPulsarServiceBaseTest {
private static final String subscription = "reader-multi-topics-sub";
@@ -241,9 +241,7 @@ public class MultiTopicsReaderTest extends MockedPulsarServiceBaseTest {
public void testMultiReaderSeek() throws Exception {
String topic = "persistent://my-property/my-ns/testKeyHashRangeReader";
admin.topics().createPartitionedTopic(topic, 3);
- Set<String> ids = publishMessages(topic,100,false);
- List<String> idList = new ArrayList<>(ids);
- Collections.sort(idList);
+ publishMessages(topic,100,false);
}
@Test(timeOut = 10000)
@@ -321,7 +319,7 @@ public class MultiTopicsReaderTest extends MockedPulsarServiceBaseTest {
assertTrue(expectedMessages > 0);
assertEquals(receivedMessages.size(), expectedMessages);
for (String receivedMessage : receivedMessages) {
- assertTrue(Integer.valueOf(receivedMessage) <= StickyKeyConsumerSelector.DEFAULT_RANGE_SIZE / 2);
+ assertTrue(Integer.parseInt(receivedMessage) <= StickyKeyConsumerSelector.DEFAULT_RANGE_SIZE / 2);
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java
index 366695a..5eb43af 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java
@@ -41,6 +41,7 @@ import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@Slf4j
+@Test(groups = "broker-impl")
public class NegativeAcksTest extends ProducerConsumerBase {
@Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImplTest.java
index 1df4d0a..55649ab 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImplTest.java
@@ -52,6 +52,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "broker-impl")
public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
private static final long testTimeout = 90000; // 1.5 min
private static final Logger log = LoggerFactory.getLogger(PatternTopicsConsumerImplTest.class);
@@ -71,6 +72,7 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
public void cleanup() throws Exception {
super.internalCleanup();
}
+
@Test(timeOut = testTimeout)
public void testPatternTopicsSubscribeWithBuilderFail() throws Exception {
String key = "PatternTopicsSubscribeWithBuilderFail";
@@ -419,7 +421,7 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
}
@Test(timeOut = testTimeout)
- public void testTopicsPatternFilter() throws Exception {
+ public void testTopicsPatternFilter() {
String topicName1 = "persistent://my-property/my-ns/pattern-topic-1";
String topicName2 = "persistent://my-property/my-ns/pattern-topic-2";
String topicName3 = "persistent://my-property/my-ns/hello-3";
@@ -438,7 +440,7 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
}
@Test(timeOut = testTimeout)
- public void testTopicsListMinus() throws Exception {
+ public void testTopicsListMinus() {
String topicName1 = "persistent://my-property/my-ns/pattern-topic-1";
String topicName2 = "persistent://my-property/my-ns/pattern-topic-2";
String topicName3 = "persistent://my-property/my-ns/pattern-topic-3";
@@ -677,7 +679,7 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
}
@Test(timeOut = testTimeout)
- public void testAutoUnbubscribePatternConsumer() throws Exception {
+ public void testAutoUnsubscribePatternConsumer() throws Exception {
String key = "AutoUnsubscribePatternConsumer";
String subscriptionName = "my-ex-subscription-" + key;
String topicName1 = "persistent://my-property/my-ns/pattern-topic-1-" + key;
@@ -782,7 +784,7 @@ public class PatternTopicsConsumerImplTest extends ProducerConsumerBase {
producer3.close();
}
- @Test()
+ @Test
public void testTopicDeletion() throws Exception {
String baseTopicName = "persistent://my-property/my-ns/pattern-topic-" + System.currentTimeMillis();
Pattern pattern = Pattern.compile(baseTopicName + ".*");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PerMessageUnAcknowledgedRedeliveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PerMessageUnAcknowledgedRedeliveryTest.java
index 9eb559c..9a49460 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PerMessageUnAcknowledgedRedeliveryTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PerMessageUnAcknowledgedRedeliveryTest.java
@@ -36,6 +36,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "broker-impl")
public class PerMessageUnAcknowledgedRedeliveryTest extends BrokerTestBase {
private static final long testTimeout = 90000; // 1.5 min
private static final Logger log = LoggerFactory.getLogger(PerMessageUnAcknowledgedRedeliveryTest.class);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
index 99020a2..f801d7a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
@@ -39,8 +39,8 @@ import java.util.concurrent.ExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+@Test(groups = "broker-impl")
public class ProducerSemaphoreTest extends ProducerConsumerBase {
- private static final Logger log = LoggerFactory.getLogger(ProducerSemaphoreTest.class);
@Override
@BeforeMethod
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarMultiHostClientTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarMultiHostClientTest.java
index ba29a04..04c73f4 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarMultiHostClientTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PulsarMultiHostClientTest.java
@@ -37,6 +37,7 @@ import java.util.concurrent.TimeUnit;
import static org.testng.Assert.fail;
+@Test(groups = "broker-impl")
public class PulsarMultiHostClientTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(PulsarMultiHostClientTest.class);
@@ -55,7 +56,7 @@ public class PulsarMultiHostClientTest extends ProducerConsumerBase {
}
@Test
- public void testGetPartitionedTopicMetaData() throws Exception {
+ public void testGetPartitionedTopicMetaData() {
log.info("-- Starting {} test --", methodName);
final String topicName = "persistent://my-property/my-ns/my-topic1";
@@ -137,7 +138,7 @@ public class PulsarMultiHostClientTest extends ProducerConsumerBase {
}
Message<byte[]> msg = null;
- Set<String> messageSet = new HashSet();
+ Set<String> messageSet = new HashSet<>();
for (int i = 0; i < 5; i++) {
msg = consumer.receive(5, TimeUnit.SECONDS);
String receivedMessage = new String(msg.getData());
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawMessageSerDeserTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawMessageSerDeserTest.java
index ee4dc9c..9ea7b01 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawMessageSerDeserTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawMessageSerDeserTest.java
@@ -18,7 +18,6 @@
*/
package org.apache.pulsar.client.impl;
-
import org.apache.pulsar.client.api.RawMessage;
import org.apache.pulsar.common.api.proto.MessageIdData;
import org.slf4j.Logger;
@@ -30,11 +29,12 @@ import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import lombok.Cleanup;
+@Test(groups = "broker-impl")
public class RawMessageSerDeserTest {
static final Logger log = LoggerFactory.getLogger(RawMessageSerDeserTest.class);
@Test
- public void testSerializationAndDeserialization() throws Exception {
+ public void testSerializationAndDeserialization() {
int payload = 0xbeefcafe;
ByteBuf headersAndPayload = Unpooled.buffer(4);
headersAndPayload.writeInt(payload);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java
index 78ee802..06cac9f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java
@@ -48,8 +48,9 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "flaky")
public class RawReaderTest extends MockedPulsarServiceBaseTest {
- private static final int BATCH_MAX_MESSAGES = 10;
+
private static final String subscription = "foobar-sub";
@BeforeMethod
@@ -97,7 +98,7 @@ public class RawReaderTest extends MockedPulsarServiceBaseTest {
return keys;
}
- public static String extractKey(RawMessage m) throws Exception {
+ public static String extractKey(RawMessage m) {
ByteBuf headersAndPayload = m.getHeadersAndPayload();
MessageMetadata msgMetadata = Commands.parseMessageMetadata(headersAndPayload);
return msgMetadata.getPartitionKey();
@@ -357,7 +358,7 @@ public class RawReaderTest extends MockedPulsarServiceBaseTest {
PersistentTopic topicRef = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topic).get();
ManagedLedger ledger = topicRef.getManagedLedger();
for (int i = 0; i < 30; i++) {
- if (ledger.openCursor(subscription).getProperties().get("foobar") == Long.valueOf(0xdeadbeefdecaL)) {
+ if (ledger.openCursor(subscription).getProperties().get("foobar").equals(0xdeadbeefdecaL)) {
break;
}
Thread.sleep(100);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
index 13fc898..d771d7a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
@@ -57,6 +57,7 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@Slf4j
+@Test(groups = "broker-impl")
public class ReaderTest extends MockedPulsarServiceBaseTest {
private static final String subscription = "reader-sub";
@@ -144,7 +145,7 @@ public class ReaderTest extends MockedPulsarServiceBaseTest {
Assert.assertTrue(keys.remove(reader.readNext().getKey()));
}
// start from latest with start message inclusive should only read the last message in batch
- Assert.assertTrue(keys.size() == 9);
+ assertEquals(keys.size(), 9);
Assert.assertFalse(keys.contains("key9"));
Assert.assertFalse(reader.hasMessageAvailable());
}
@@ -394,7 +395,7 @@ public class ReaderTest extends MockedPulsarServiceBaseTest {
assertEquals(receivedMessages.size(), expectedMessages);
for (String receivedMessage : receivedMessages) {
log.info("Receive message {}", receivedMessage);
- assertTrue(Integer.valueOf(receivedMessage) <= StickyKeyConsumerSelector.DEFAULT_RANGE_SIZE / 2);
+ assertTrue(Integer.parseInt(receivedMessage) <= StickyKeyConsumerSelector.DEFAULT_RANGE_SIZE / 2);
}
}
@@ -405,7 +406,7 @@ public class ReaderTest extends MockedPulsarServiceBaseTest {
doTestReaderSubName(false);
}
- public void doTestReaderSubName(boolean setPrefix) throws Exception {
+ private void doTestReaderSubName(boolean setPrefix) throws Exception {
final String topic = "persistent://my-property/my-ns/testReaderSubName" + System.currentTimeMillis();
final String subName = "my-sub-name";
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SchemaDeleteTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SchemaDeleteTest.java
index 652b425..a12b000 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SchemaDeleteTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SchemaDeleteTest.java
@@ -34,10 +34,9 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "broker-impl")
public class SchemaDeleteTest extends MockedPulsarServiceBaseTest {
- private static final String subscription = "reader-sub";
-
@BeforeMethod
@Override
protected void setup() throws Exception {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SequenceIdWithErrorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SequenceIdWithErrorTest.java
index 62a4ff0..eb38a56 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SequenceIdWithErrorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SequenceIdWithErrorTest.java
@@ -34,6 +34,7 @@ import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.naming.TopicName;
import org.testng.annotations.Test;
+@Test(groups = "broker-impl")
public class SequenceIdWithErrorTest extends BrokerBkEnsemblesTests {
/**
@@ -79,12 +80,12 @@ public class SequenceIdWithErrorTest extends BrokerBkEnsemblesTests {
}
@Test(enabled = false)
- public void testCrashBrokerWithoutCursorLedgerLeak() throws Exception {
+ public void testCrashBrokerWithoutCursorLedgerLeak() {
// Ignore test
}
@Test(enabled = false)
- public void testSkipCorruptDataLedger() throws Exception {
+ public void testSkipCorruptDataLedger() {
// Ignore test
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicDoesNotExistsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicDoesNotExistsTest.java
index 3bae6ee..6fe2812 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicDoesNotExistsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicDoesNotExistsTest.java
@@ -16,7 +16,6 @@
* specific language governing permissions and limitations
* under the License.
*/
-
package org.apache.pulsar.client.impl;
import io.netty.util.HashedWheelTimer;
@@ -34,6 +33,7 @@ import java.util.concurrent.TimeUnit;
/**
* Tests for not exists topic.
*/
+@Test(groups = "broker-impl")
public class TopicDoesNotExistsTest extends ProducerConsumerBase {
@Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicFromMessageTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicFromMessageTest.java
index 0586ed6..5ad4b51 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicFromMessageTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicFromMessageTest.java
@@ -27,6 +27,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "broker-impl")
public class TopicFromMessageTest extends ProducerConsumerBase {
private static final long TEST_TIMEOUT = 90000; // 1.5 min
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicPublishThrottlingInitTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicPublishThrottlingInitTest.java
index ca4c69e..3ffa7bb 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicPublishThrottlingInitTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicPublishThrottlingInitTest.java
@@ -31,6 +31,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "broker-impl")
public class TopicPublishThrottlingInitTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(TopicPublishThrottlingInitTest.class);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java
index 0634b71..15abac6 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java
@@ -71,6 +71,7 @@ import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
@SuppressWarnings({ "unchecked", "rawtypes" })
+@Test(groups = "broker-impl")
public class TopicsConsumerImplTest extends ProducerConsumerBase {
private static final long testTimeout = 90000; // 1.5 min
private static final Logger log = LoggerFactory.getLogger(TopicsConsumerImplTest.class);
@@ -606,7 +607,6 @@ public class TopicsConsumerImplTest extends ProducerConsumerBase {
producer3.close();
}
-
@Test
public void testResubscribeSameTopic() throws Exception {
final String localTopicName = "TopicsConsumerResubscribeSameTopicTest";
@@ -753,8 +753,6 @@ public class TopicsConsumerImplTest extends ProducerConsumerBase {
.subscribe();
assertTrue(consumer instanceof MultiTopicsConsumerImpl);
- MultiTopicsConsumerImpl topicsConsumer = (MultiTopicsConsumerImpl) consumer;
-
// 3. producer publish messages
for (int i = 0; i < totalMessages; i++) {
producer1.send((messagePredicate + "producer1-" + i).getBytes());
@@ -1156,7 +1154,7 @@ public class TopicsConsumerImplTest extends ProducerConsumerBase {
try {
client.newConsumer(Schema.STRING)
.subscriptionName("subName")
- .topics(Lists.<String>newArrayList(topic0, topic1))
+ .topics(Lists.newArrayList(topic0, topic1))
.receiverQueueSize(2)
.subscriptionType(SubscriptionType.Shared)
.ackTimeout(365, TimeUnit.DAYS)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
index f76c716..a18869f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
@@ -39,14 +39,12 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
-import org.apache.pulsar.broker.TransactionMetadataStoreService;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.broker.transaction.TransactionTestBase;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
-import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerBuilder;
import org.apache.pulsar.client.api.PulsarClient;
@@ -74,11 +72,11 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
-
/**
* End to end transaction test.
*/
@Slf4j
+@Test(groups = "flaky")
public class TransactionEndToEndTest extends TransactionTestBase {
private final static int TOPIC_PARTITION = 3;
@@ -485,7 +483,7 @@ public class TransactionEndToEndTest extends TransactionTestBase {
txnCumulativeAckTest(false, 1, SubscriptionType.Failover);
}
- public void txnCumulativeAckTest(boolean batchEnable, int maxBatchSize, SubscriptionType subscriptionType)
+ private void txnCumulativeAckTest(boolean batchEnable, int maxBatchSize, SubscriptionType subscriptionType)
throws Exception {
String normalTopic = NAMESPACE1 + "/normal-topic";
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/UnAcknowledgedMessagesTimeoutTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/UnAcknowledgedMessagesTimeoutTest.java
index 5f6aae7..dd19b77 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/UnAcknowledgedMessagesTimeoutTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/UnAcknowledgedMessagesTimeoutTest.java
@@ -43,6 +43,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "broker-impl")
public class UnAcknowledgedMessagesTimeoutTest extends BrokerTestBase {
private static final Logger log = LoggerFactory.getLogger(UnAcknowledgedMessagesTimeoutTest.class);
private final long ackTimeOutMillis = TimeUnit.SECONDS.toMillis(2);
@@ -346,7 +347,7 @@ public class UnAcknowledgedMessagesTimeoutTest extends BrokerTestBase {
}
@Test
- public void testAckTimeoutMinValue() throws PulsarClientException {
+ public void testAckTimeoutMinValue() {
try {
pulsarClient.newConsumer().ackTimeout(999, TimeUnit.MILLISECONDS);
Assert.fail("Exception should have been thrown since the set timeout is less than min timeout.");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ZeroQueueSizeTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ZeroQueueSizeTest.java
index ef5a7f3..481bec1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ZeroQueueSizeTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ZeroQueueSizeTest.java
@@ -53,6 +53,7 @@ import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
+@Test(groups = "broker-impl")
public class ZeroQueueSizeTest extends BrokerTestBase {
private static final Logger log = LoggerFactory.getLogger(ZeroQueueSizeTest.class);
private final int totalMessages = 10;
@@ -80,8 +81,8 @@ public class ZeroQueueSizeTest extends BrokerTestBase {
}
@Test(expectedExceptions = PulsarClientException.InvalidConfigurationException.class)
- public void zeroQueueSizeReceieveAsyncInCompatibility() throws PulsarClientException {
- String key = "zeroQueueSizeReceieveAsyncInCompatibility";
+ public void zeroQueueSizeReceiveAsyncInCompatibility() throws PulsarClientException {
+ String key = "zeroQueueSizeReceiveAsyncInCompatibility";
final String topicName = "persistent://prop/use/ns-abc/topic-" + key;
final String subscriptionName = "my-ex-subscription-" + key;
@@ -100,7 +101,7 @@ public class ZeroQueueSizeTest extends BrokerTestBase {
pulsarClient.newConsumer().topic(topicName).subscriptionName(subscriptionName).receiverQueueSize(0).subscribe();
}
- @Test()
+ @Test
public void zeroQueueSizeNormalConsumer() throws PulsarClientException {
String key = "nonZeroQueueSizeNormalConsumer";
@@ -137,7 +138,7 @@ public class ZeroQueueSizeTest extends BrokerTestBase {
}
}
- @Test()
+ @Test
public void zeroQueueSizeConsumerListener() throws Exception {
String key = "zeroQueueSizeConsumerListener";
@@ -181,7 +182,7 @@ public class ZeroQueueSizeTest extends BrokerTestBase {
}
}
- @Test()
+ @Test
public void zeroQueueSizeSharedSubscription() throws PulsarClientException {
String key = "zeroQueueSizeSharedSubscription";
@@ -222,7 +223,7 @@ public class ZeroQueueSizeTest extends BrokerTestBase {
}
}
- @Test()
+ @Test
public void zeroQueueSizeFailoverSubscription() throws PulsarClientException {
String key = "zeroQueueSizeFailoverSubscription";
@@ -277,7 +278,7 @@ public class ZeroQueueSizeTest extends BrokerTestBase {
}
}
- @Test()
+ @Test
public void testFailedZeroQueueSizeBatchMessage() throws PulsarClientException {
int batchMessageDelayMs = 100;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/common/naming/NamespaceBundleTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/common/naming/NamespaceBundleTest.java
index 5b84b7e..fac10ac 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/common/naming/NamespaceBundleTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/common/naming/NamespaceBundleTest.java
@@ -39,6 +39,7 @@ import com.google.common.collect.BoundType;
import com.google.common.collect.Range;
import com.google.common.hash.Hashing;
+@Test(groups = "broker-naming")
public class NamespaceBundleTest {
private final NamespaceBundleFactory factory = getNamespaceBundleFactory();
@@ -127,7 +128,7 @@ public class NamespaceBundleTest {
}
@Test
- public void testGetBundle() throws Exception {
+ public void testGetBundle() {
NamespaceBundle bundle = factory.getBundle(NamespaceName.get("pulsar/use/ns1"),
Range.range(0L, BoundType.CLOSED, 0xffffffffL, BoundType.CLOSED));
assertNotNull(bundle);
@@ -139,7 +140,7 @@ public class NamespaceBundleTest {
}
@Test
- public void testCompareTo() throws Exception {
+ public void testCompareTo() {
NamespaceBundle bundle = factory.getBundle(NamespaceName.get("pulsar/use/ns1"),
Range.range(0l, BoundType.CLOSED, 0x40000000L, BoundType.OPEN));
NamespaceBundle bundle2 = factory.getBundle(NamespaceName.get("pulsar/use/ns1"),
@@ -186,7 +187,7 @@ public class NamespaceBundleTest {
}
@Test
- public void testIncludes() throws Exception {
+ public void testIncludes() {
TopicName topicName = TopicName.get("persistent://pulsar/use/ns1/topic-1");
Long hashKey = factory.getLongHashCode(topicName.toString());
Long upper = Math.max(hashKey + 1, NamespaceBundles.FULL_UPPER_BOUND);
@@ -204,7 +205,7 @@ public class NamespaceBundleTest {
}
@Test
- public void testToString() throws Exception {
+ public void testToString() {
NamespaceBundle bundle0 = factory.getBundle(NamespaceName.get("pulsar/use/ns1"),
Range.range(0l, BoundType.CLOSED, 0x10000000L, BoundType.OPEN));
assertEquals(bundle0.toString(), "pulsar/use/ns1/0x00000000_0x10000000");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/common/naming/NamespaceBundlesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/common/naming/NamespaceBundlesTest.java
index 2a2521d..32da8e7 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/common/naming/NamespaceBundlesTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/common/naming/NamespaceBundlesTest.java
@@ -37,11 +37,6 @@ import java.util.concurrent.CompletableFuture;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.cache.LocalZooKeeperCacheService;
-import org.apache.pulsar.common.naming.TopicName;
-import org.apache.pulsar.common.naming.NamespaceBundle;
-import org.apache.pulsar.common.naming.NamespaceBundleFactory;
-import org.apache.pulsar.common.naming.NamespaceBundles;
-import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.policies.data.LocalPolicies;
import org.apache.pulsar.zookeeper.ZooKeeperDataCache;
import org.slf4j.Logger;
@@ -54,6 +49,7 @@ import com.google.common.collect.Range;
import com.google.common.collect.Sets;
import com.google.common.hash.Hashing;
+@Test(groups = "broker-naming")
public class NamespaceBundlesTest {
private final NamespaceBundleFactory factory = getNamespaceBundleFactory();
@@ -97,10 +93,10 @@ public class NamespaceBundlesTest {
// OK, expected
}
- partitions.add(0l);
- partitions.add(0x10000000l);
- partitions.add(0x40000000l);
- partitions.add(0xffffffffl);
+ partitions.add(0L);
+ partitions.add(0x10000000L);
+ partitions.add(0x40000000L);
+ partitions.add(0xffffffffL);
NamespaceBundles bundles = new NamespaceBundles(NamespaceName.get("pulsar/use/ns2"), partitions, factory);
Field partitionField = NamespaceBundles.class.getDeclaredField("partitions");
Field nsField = NamespaceBundles.class.getDeclaredField("nsname");
@@ -116,11 +112,11 @@ public class NamespaceBundlesTest {
ArrayList<NamespaceBundle> bundleList = (ArrayList<NamespaceBundle>) bundlesField.get(bundles);
assertEquals(bundleList.size(), 3);
assertEquals(bundleList.get(0),
- factory.getBundle(nsFld, Range.range(0l, BoundType.CLOSED, 0x10000000l, BoundType.OPEN)));
+ factory.getBundle(nsFld, Range.range(0L, BoundType.CLOSED, 0x10000000L, BoundType.OPEN)));
assertEquals(bundleList.get(1),
- factory.getBundle(nsFld, Range.range(0x10000000l, BoundType.CLOSED, 0x40000000l, BoundType.OPEN)));
+ factory.getBundle(nsFld, Range.range(0x10000000L, BoundType.CLOSED, 0x40000000L, BoundType.OPEN)));
assertEquals(bundleList.get(2),
- factory.getBundle(nsFld, Range.range(0x40000000l, BoundType.CLOSED, 0xffffffffl, BoundType.CLOSED)));
+ factory.getBundle(nsFld, Range.range(0x40000000L, BoundType.CLOSED, 0xffffffffL, BoundType.CLOSED)));
}
@SuppressWarnings("unchecked")
@@ -137,12 +133,12 @@ public class NamespaceBundlesTest {
@Test
public void testFindBundle() throws Exception {
SortedSet<Long> partitions = Sets.newTreeSet();
- partitions.add(0l);
- partitions.add(0x40000000l);
- partitions.add(0xa0000000l);
- partitions.add(0xb0000000l);
- partitions.add(0xc0000000l);
- partitions.add(0xffffffffl);
+ partitions.add(0L);
+ partitions.add(0x40000000L);
+ partitions.add(0xa0000000L);
+ partitions.add(0xb0000000L);
+ partitions.add(0xc0000000L);
+ partitions.add(0xffffffffL);
NamespaceBundles bundles = new NamespaceBundles(NamespaceName.get("pulsar/global/ns1"), partitions, factory);
TopicName topicName = TopicName.get("persistent://pulsar/global/ns1/topic-1");
NamespaceBundle bundle = bundles.findBundle(topicName);
@@ -175,7 +171,7 @@ public class NamespaceBundlesTest {
}
@Test
- public void testsplitBundles() throws Exception {
+ public void testSplitBundles() throws Exception {
NamespaceName nsname = NamespaceName.get("pulsar/global/ns1");
TopicName topicName = TopicName.get("persistent://pulsar/global/ns1/topic-1");
NamespaceBundles bundles = factory.getBundles(nsname);
@@ -195,8 +191,13 @@ public class NamespaceBundlesTest {
NamespaceBundles bundles2 = splitBundles.getLeft();
NamespaceBundle testChildBundle = bundles2.getBundles().get(0);
- Pair<NamespaceBundles, List<NamespaceBundle>> splitChildBundles = splitBundlesUtilFactory(utilityFactory,
- nsname, bundles2, testChildBundle, numberSplitBundles);
+ Pair<NamespaceBundles, List<NamespaceBundle>> splitChildBundles =
+ splitBundlesUtilFactory(
+ utilityFactory,
+ nsname,
+ bundles2,
+ testChildBundle,
+ numberSplitBundles);
// existing_no_bundles(4) +
// additional_new_split_bundle(4) -
// parent_target_bundle(1)
@@ -207,8 +208,13 @@ public class NamespaceBundlesTest {
// (3) split in 3: second bundle from above split bundles
NamespaceBundle testChildBundl2 = bundles2.getBundles().get(1);
- Pair<NamespaceBundles, List<NamespaceBundle>> splitChildBundles2 = splitBundlesUtilFactory(utilityFactory,
- nsname, bundles2, testChildBundl2, 3);
+ Pair<NamespaceBundles, List<NamespaceBundle>> splitChildBundles2 =
+ splitBundlesUtilFactory(
+ utilityFactory,
+ nsname,
+ bundles2,
+ testChildBundl2,
+ 3);
// existing_no_bundles(4) +
// additional_new_split_bundle(3) -
// parent_target_bundle(1)
@@ -227,7 +233,7 @@ public class NamespaceBundlesTest {
// (1) split : [0x00000000,0xffffffff] => [0x00000000_0x7fffffff,0x7fffffff_0xffffffff]
Pair<NamespaceBundles, List<NamespaceBundle>> splitBundles = factory.splitBundles(bundle, NO_BUNDLES, null);
assertNotNull(splitBundles);
- assertBundleDivideInTwo(bundle, splitBundles.getRight(), NO_BUNDLES);
+ assertBundleDivideInTwo(bundle, splitBundles.getRight());
// (2) split: [0x00000000,0x7fffffff] => [0x00000000_0x3fffffff,0x3fffffff_0x7fffffff],
// [0x7fffffff,0xffffffff] => [0x7fffffff_0xbfffffff,0xbfffffff_0xffffffff]
@@ -293,8 +299,11 @@ public class NamespaceBundlesTest {
return utilityFactory.splitBundles(targetBundle, numBundles, null);
}
- private void assertBundles(NamespaceBundleFactory utilityFactory, NamespaceName nsname, NamespaceBundle bundle,
- Pair<NamespaceBundles, List<NamespaceBundle>> splitBundles, int numBundles) throws Exception {
+ private void assertBundles(NamespaceBundleFactory utilityFactory,
+ NamespaceName nsname,
+ NamespaceBundle bundle,
+ Pair<NamespaceBundles, List<NamespaceBundle>> splitBundles,
+ int numBundles) throws Exception {
NamespaceBundle bundle1 = splitBundles.getRight().get(0);
NamespaceBundle bundle2 = splitBundles.getRight().get(1);
@@ -302,15 +311,16 @@ public class NamespaceBundlesTest {
NamespaceBundles nspaceBundles = splitBundles.getLeft();
Pair<NamespaceBundles, List<NamespaceBundle>> bundle1Split = splitBundlesUtilFactory(utilityFactory, nsname,
nspaceBundles, bundle1, numBundles);
- assertBundleDivideInTwo(bundle1, bundle1Split.getRight(), numBundles);
+ assertBundleDivideInTwo(bundle1, bundle1Split.getRight());
Pair<NamespaceBundles, List<NamespaceBundle>> bundle2Split = splitBundlesUtilFactory(utilityFactory, nsname,
nspaceBundles, bundle2, numBundles);
- assertBundleDivideInTwo(bundle2, bundle2Split.getRight(), numBundles);
+ assertBundleDivideInTwo(bundle2, bundle2Split.getRight());
}
- private void assertBundleDivideInTwo(NamespaceBundle bundle, List<NamespaceBundle> bundles, int numBundles) {
+ private void assertBundleDivideInTwo(NamespaceBundle bundle,
+ List<NamespaceBundle> bundles) {
assertEquals(bundles.size(), 2);
String[] range = bundle.getBundleRange().split("_");
long lower = Long.decode(range[0]);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/common/naming/ServiceConfigurationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/common/naming/ServiceConfigurationTest.java
index bc9e4bd..6b7df2c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/common/naming/ServiceConfigurationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/common/naming/ServiceConfigurationTest.java
@@ -35,10 +35,7 @@ import org.apache.pulsar.common.configuration.PulsarConfigurationLoader;
import org.apache.pulsar.common.policies.data.InactiveTopicDeleteMode;
import org.testng.annotations.Test;
-/**
- *
- *
- */
+@Test(groups = "broker-naming")
public class ServiceConfigurationTest {
final String fileName = "configurations/pulsar_broker_test.conf"; // test-resource file
@@ -89,7 +86,7 @@ public class ServiceConfigurationTest {
@Test(expectedExceptions = IllegalArgumentException.class)
public void testInitFailure() throws Exception {
final String zookeeperServer = "localhost:2184";
- InputStream newStream = updateProp(zookeeperServer, String.valueOf("invalid-string"), null);
+ InputStream newStream = updateProp(zookeeperServer, "invalid-string", null);
PulsarConfigurationLoader.create(newStream, ServiceConfiguration.class);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
index 0f67157..959a395 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
@@ -61,6 +61,7 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+@Test(groups = "broker-compaction")
public class CompactedTopicTest extends MockedPulsarServiceBaseTest {
private final Random r = new Random(0);
@@ -107,7 +108,7 @@ public class CompactedTopicTest extends MockedPulsarServiceBaseTest {
AtomicLong entryIds = new AtomicLong(0L);
CompletableFuture.allOf(
IntStream.range(0, count).mapToObj((i) -> {
- List<MessageIdData> idsInGap = new ArrayList<MessageIdData>();
+ List<MessageIdData> idsInGap = new ArrayList<>();
if (r.nextInt(10) == 1) {
long delta = r.nextInt(10) + 1;
idsInGap.add(new MessageIdData()
@@ -194,14 +195,13 @@ public class CompactedTopicTest extends MockedPulsarServiceBaseTest {
for (Pair<MessageIdData, Long> p : positions) {
PositionImpl pos = new PositionImpl(p.getLeft().getLedgerId(), p.getLeft().getEntryId());
Long got = CompactedTopicImpl.findStartPoint(pos, lastEntryId, cache).get();
- Assert.assertEquals(got, Long.valueOf(p.getRight()));
+ Assert.assertEquals(got, p.getRight());
}
// Check ids we know are in the gaps of the compacted ledger
for (Pair<MessageIdData, Long> gap : idsInGaps) {
PositionImpl pos = new PositionImpl(gap.getLeft().getLedgerId(), gap.getLeft().getEntryId());
- Assert.assertEquals(CompactedTopicImpl.findStartPoint(pos, lastEntryId, cache).get(),
- Long.valueOf(gap.getRight()));
+ Assert.assertEquals(CompactedTopicImpl.findStartPoint(pos, lastEntryId, cache).get(), gap.getRight());
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java
index a7e4abb..6c44121 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java
@@ -73,6 +73,7 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+@Test(groups = "flaky")
public class CompactionTest extends MockedPulsarServiceBaseTest {
private ScheduledExecutorService compactionScheduler;
private BookKeeper bk;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactorTest.java
index 5c5edd4..3405136 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactorTest.java
@@ -52,6 +52,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "broker-compaction")
public class CompactorTest extends MockedPulsarServiceBaseTest {
private ScheduledExecutorService compactionScheduler;
@@ -75,7 +76,6 @@ public class CompactorTest extends MockedPulsarServiceBaseTest {
@Override
public void cleanup() throws Exception {
super.internalCleanup();
-
compactionScheduler.shutdownNow();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/discovery/service/web/DiscoveryServiceWebTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/discovery/service/web/DiscoveryServiceWebTest.java
index 3feedbe..dcd4b8c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/discovery/service/web/DiscoveryServiceWebTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/discovery/service/web/DiscoveryServiceWebTest.java
@@ -52,9 +52,10 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "broker-discovery")
public class DiscoveryServiceWebTest extends ProducerConsumerBase {
- private Client client = ClientBuilder.newClient(new ClientConfig().register(LoggingFeature.class));
+ private final Client client = ClientBuilder.newClient(new ClientConfig().register(LoggingFeature.class));
@BeforeMethod
@Override
@@ -116,11 +117,11 @@ public class DiscoveryServiceWebTest extends ProducerConsumerBase {
WebTarget webTarget = client.target(url);
Invocation.Builder invocationBuilder = webTarget.request(MediaType.APPLICATION_JSON);
if (HttpMethod.PUT.equals(method)) {
- response = (Response) invocationBuilder.put(Entity.entity(data, MediaType.APPLICATION_JSON));
+ response = invocationBuilder.put(Entity.entity(data, MediaType.APPLICATION_JSON));
} else if (HttpMethod.GET.equals(method)) {
- response = (Response) invocationBuilder.get();
+ response = invocationBuilder.get();
} else if (HttpMethod.POST.equals(method)) {
- response = (Response) invocationBuilder.post(Entity.entity(data, MediaType.APPLICATION_JSON));
+ response = invocationBuilder.post(Entity.entity(data, MediaType.APPLICATION_JSON));
} else {
fail("Unsupported http method");
}
@@ -130,8 +131,7 @@ public class DiscoveryServiceWebTest extends ProducerConsumerBase {
}
JsonObject jsonObject = new Gson().fromJson(response.readEntity(String.class), JsonObject.class);
- String serviceResponse = jsonObject.get("reason").getAsString();
- return serviceResponse;
+ return jsonObject.get("reason").getAsString();
}
static class DiscoveryZooKeeperClientFactoryImpl implements ZooKeeperClientFactory {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java
index 3b0628b..9d7a796 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java
@@ -82,6 +82,7 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+@Test(groups = "functions-worker")
public class PulsarFunctionE2ESecurityTest {
LocalBookkeeperEnsemble bkEnsemble;
@@ -259,7 +260,12 @@ public class PulsarFunctionE2ESecurityTest {
return workerService;
}
- protected static FunctionConfig createFunctionConfig(String tenant, String namespace, String functionName, String sourceTopic, String sinkTopic, String subscriptionName) {
+ protected static FunctionConfig createFunctionConfig(String tenant,
+ String namespace,
+ String functionName,
+ String sourceTopic,
+ String sinkTopic,
+ String subscriptionName) {
FunctionConfig functionConfig = new FunctionConfig();
functionConfig.setTenant(tenant);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java
index 7bb15bc..f24a520 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java
@@ -22,6 +22,7 @@ import static org.apache.commons.lang3.StringUtils.isNotBlank;
import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.retryStrategically;
import static org.apache.pulsar.functions.utils.functioncache.FunctionCacheEntry.JAVA_INSTANCE_JAR_PROPERTY;
import static org.mockito.Mockito.spy;
+
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertNotNull;
@@ -29,6 +30,7 @@ import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
+
import java.io.File;
import java.io.IOException;
import java.lang.reflect.Method;
@@ -94,8 +96,8 @@ import org.testng.annotations.Test;
/**
* Test Pulsar sink on function
- *
*/
+@Test(groups = { "flaky" })
public class PulsarFunctionLocalRunTest {
LocalBookkeeperEnsemble bkEnsemble;
@@ -343,7 +345,12 @@ public class PulsarFunctionLocalRunTest {
return workerConfig;
}
- protected static FunctionConfig createFunctionConfig(String tenant, String namespace, String functionName, String sourceTopic, String sinkTopic, String subscriptionName) {
+ protected static FunctionConfig createFunctionConfig(String tenant,
+ String namespace,
+ String functionName,
+ String sourceTopic,
+ String sinkTopic,
+ String subscriptionName) {
FunctionConfig functionConfig = new FunctionConfig();
functionConfig.setTenant(tenant);
@@ -361,7 +368,10 @@ public class PulsarFunctionLocalRunTest {
return functionConfig;
}
- private static SourceConfig createSourceConfig(String tenant, String namespace, String functionName, String sinkTopic) {
+ private static SourceConfig createSourceConfig(String tenant,
+ String namespace,
+ String functionName,
+ String sinkTopic) {
SourceConfig sourceConfig = new SourceConfig();
sourceConfig.setTenant(tenant);
sourceConfig.setNamespace(namespace);
@@ -372,7 +382,11 @@ public class PulsarFunctionLocalRunTest {
return sourceConfig;
}
- private static SinkConfig createSinkConfig(String tenant, String namespace, String functionName, String sourceTopic, String subName) {
+ private static SinkConfig createSinkConfig(String tenant,
+ String namespace,
+ String functionName,
+ String sourceTopic,
+ String subName) {
SinkConfig sinkConfig = new SinkConfig();
sinkConfig.setTenant(tenant);
sinkConfig.setNamespace(namespace);
@@ -501,7 +515,7 @@ public class PulsarFunctionLocalRunTest {
}
}
- public void testAvroFunctionLocalRun(String jarFilePathUrl) throws Exception {
+ private void testAvroFunctionLocalRun(String jarFilePathUrl) throws Exception {
final String namespacePortion = "io";
final String replNamespace = tenant + "/" + namespacePortion;
@@ -615,7 +629,7 @@ public class PulsarFunctionLocalRunTest {
producer.newMessage().property(propertyKey, propertyValue).value(avroTestObjectClass.newInstance()).send();
Message<GenericRecord> msg = consumer.receive(2, TimeUnit.SECONDS);
- assertEquals(msg, null);
+ Assert.assertNull(msg);
producer.close();
consumer.close();
@@ -747,7 +761,6 @@ public class PulsarFunctionLocalRunTest {
testPulsarSourceLocalRun(fileServer.getUrl("/pulsar-io-data-generator.nar"));
}
-
private void testPulsarSinkLocalRun(String jarFilePathUrl) throws Exception {
final String namespacePortion = "io";
final String replNamespace = tenant + "/" + namespacePortion;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java
index ce2cfaf..0bf0886 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java
@@ -83,9 +83,9 @@ import java.util.concurrent.TimeUnit;
/**
* Test Pulsar function state
- *
*/
@Slf4j
+@Test(groups = "functions-worker")
public class PulsarFunctionPublishTest {
LocalBookkeeperEnsemble bkEnsemble;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTlsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTlsTest.java
index 2d571c6..54c24a7 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTlsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTlsTest.java
@@ -18,7 +18,6 @@
*/
package org.apache.pulsar.functions.worker;
-import static org.testng.Assert.assertEquals;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Sets;
@@ -51,7 +50,10 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+import static org.testng.Assert.assertEquals;
+
@Slf4j
+@Test(groups = "functions-worker")
public class PulsarFunctionTlsTest {
protected static final int BROKER_COUNT = 2;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java
index 45601aa..c824b79 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java
@@ -64,6 +64,7 @@ import org.testng.annotations.Test;
*
*/
@Slf4j
+@Test(groups = "functions-worker")
public class PulsarWorkerAssignmentTest {
LocalBookkeeperEnsemble bkEnsemble;
@@ -291,7 +292,9 @@ public class PulsarWorkerAssignmentTest {
functionsWorkerService.stop();
functionsWorkerService = new PulsarWorkerService();
functionsWorkerService.init(workerConfig, dlUri, false);
- functionsWorkerService.start(new AuthenticationService(PulsarConfigurationLoader.convertFrom(workerConfig)), null, ErrorNotifier.getDefaultImpl());
+ functionsWorkerService.start(new AuthenticationService(PulsarConfigurationLoader.convertFrom(workerConfig)),
+ null,
+ ErrorNotifier.getDefaultImpl());
final FunctionRuntimeManager runtimeManager2 = functionsWorkerService.getFunctionRuntimeManager();
retryStrategically((test) -> {
try {
@@ -313,8 +316,12 @@ public class PulsarWorkerAssignmentTest {
}
}
- protected static FunctionConfig createFunctionConfig(String tenant, String namespace,
- String functionName, String sourceTopic, String sinkTopic, String subscriptionName) {
+ protected static FunctionConfig createFunctionConfig(String tenant,
+ String namespace,
+ String functionName,
+ String sourceTopic,
+ String sinkTopic,
+ String subscriptionName) {
final String sourceTopicPattern = String.format("persistent://%s/%s/%s", tenant, namespace, sourceTopic);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java
index 253c029..938f721 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java
@@ -48,6 +48,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "stats")
public class PulsarBrokerStatsClientTest extends ProducerConsumerBase {
@BeforeMethod
@@ -91,7 +92,7 @@ public class PulsarBrokerStatsClientTest extends ProducerConsumerBase {
assertTrue(client.getApiException(new ServerErrorException(500)) instanceof ServerSideErrorException);
assertTrue(client.getApiException(new ServerErrorException(503)) instanceof PulsarAdminException);
- log.info("Client: ", client);
+ log.info("Client: -- {}", client);
admin.close();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthenticationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthenticationTest.java
index 8ccf97c..300fdbd 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthenticationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthenticationTest.java
@@ -55,6 +55,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "websocket")
public class ProxyAuthenticationTest extends ProducerConsumerBase {
private ProxyServer proxyServer;
@@ -92,7 +93,7 @@ public class ProxyAuthenticationTest extends ProducerConsumerBase {
}
@AfterMethod(alwaysRun = true)
- protected void cleanup() throws Exception {
+ public void cleanup() throws Exception {
ExecutorService executor = newFixedThreadPool(1);
try {
executor.submit(() -> {
@@ -117,10 +118,9 @@ public class ProxyAuthenticationTest extends ProducerConsumerBase {
proxyServer.stop();
}
log.info("Finished Cleaning Up Test setup");
-
}
- public void checkSocket() throws Exception {
+ private void checkSocket() throws Exception {
final String topic = "my-property/my-ns/my-topic1";
final String consumerUri = "ws://localhost:" + proxyServer.getListenPortHTTP().get() + "/ws/v2/consumer/persistent/" + topic + "/my-sub";
final String producerUri = "ws://localhost:" + proxyServer.getListenPortHTTP().get() + "/ws/v2/producer/persistent/" + topic;
@@ -149,18 +149,18 @@ public class ProxyAuthenticationTest extends ProducerConsumerBase {
Assert.assertEquals(produceSocket.getBuffer(), consumeSocket.getBuffer());
}
- @Test(timeOut=10000)
+ @Test(timeOut = 10000)
public void authenticatedSocketTest() throws Exception {
checkSocket();
}
- @Test(timeOut=10000)
+ @Test(timeOut = 10000)
public void anonymousSocketTest() throws Exception {
checkSocket();
}
- @Test(timeOut=10000)
- public void unauthenticatedSocketTest() throws Exception{
+ @Test(timeOut = 10000)
+ public void unauthenticatedSocketTest() {
Exception exception = null;
try {
checkSocket();
@@ -170,7 +170,7 @@ public class ProxyAuthenticationTest extends ProducerConsumerBase {
Assert.assertTrue(exception instanceof java.util.concurrent.ExecutionException);
}
- @Test(timeOut=10000)
+ @Test(timeOut = 10000)
public void statsTest() throws Exception {
final String topic = "persistent/my-property/my-ns/my-topic2";
final String consumerUri = "ws://localhost:" + proxyServer.getListenPortHTTP().get() + "/ws/v2/consumer/" + topic + "/my-sub";
@@ -221,7 +221,7 @@ public class ProxyAuthenticationTest extends ProducerConsumerBase {
private void verifyResponseStatus(Client client, String url) {
WebTarget webTarget = client.target(url);
Invocation.Builder invocationBuilder = webTarget.request(MediaType.APPLICATION_JSON);
- Response response = (Response) invocationBuilder.get();
+ Response response = invocationBuilder.get();
Assert.assertEquals(response.getStatus(), 200);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthorizationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthorizationTest.java
index 5947313..96ee825 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthorizationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthorizationTest.java
@@ -44,6 +44,7 @@ import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
+@Test(groups = "websocket")
public class ProxyAuthorizationTest extends MockedPulsarServiceBaseTest {
private WebSocketService service;
private final String configClusterName = "c1";
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyConfigurationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyConfigurationTest.java
index 2190040..3848d6e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyConfigurationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyConfigurationTest.java
@@ -36,6 +36,7 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+@Test(groups = "websocket")
public class ProxyConfigurationTest extends ProducerConsumerBase {
private WebSocketProxyConfiguration config;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java
index 8f433d9..fac22bf 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java
@@ -80,6 +80,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "websocket")
public class ProxyPublishConsumeTest extends ProducerConsumerBase {
protected String methodName;
@@ -299,9 +300,9 @@ public class ProxyPublishConsumeTest extends ProducerConsumerBase {
}
@Test(timeOut = 10000)
- public void emptySubcriptionConsumerTest() throws Exception {
+ public void emptySubscriptionConsumerTest() {
- // Empty subcription name
+ // Empty subscription name
final String consumerUri = "ws://localhost:" + proxyServer.getListenPortHTTP().get()
+ "/ws/v2/consumer/persistent/my-property/my-ns/my-topic2/?subscriptionType=Exclusive";
URI consumeUri = URI.create(consumerUri);
@@ -778,7 +779,7 @@ public class ProxyPublishConsumeTest extends ProducerConsumerBase {
String statUrl = baseUrl + "metrics";
WebTarget webTarget = client.target(statUrl);
Invocation.Builder invocationBuilder = webTarget.request(MediaType.APPLICATION_JSON);
- Response response = (Response) invocationBuilder.get();
+ Response response = invocationBuilder.get();
String responseStr = response.readEntity(String.class);
final Gson gson = new Gson();
List<Metrics> data = gson.fromJson(responseStr, new TypeToken<List<Metrics>>() {
@@ -787,7 +788,7 @@ public class ProxyPublishConsumeTest extends ProducerConsumerBase {
// re-generate metrics
service.getProxyStats().generate();
invocationBuilder = webTarget.request(MediaType.APPLICATION_JSON);
- response = (Response) invocationBuilder.get();
+ response = invocationBuilder.get();
responseStr = response.readEntity(String.class);
data = gson.fromJson(responseStr, new TypeToken<List<Metrics>>() {
}.getType());
@@ -799,7 +800,7 @@ public class ProxyPublishConsumeTest extends ProducerConsumerBase {
String statUrl = baseUrl + "stats";
WebTarget webTarget = client.target(statUrl);
Invocation.Builder invocationBuilder = webTarget.request(MediaType.APPLICATION_JSON);
- Response response = (Response) invocationBuilder.get();
+ Response response = invocationBuilder.get();
String responseStr = response.readEntity(String.class);
final Gson gson = new Gson();
final Map<String, ProxyTopicStat> data = gson.fromJson(responseStr,
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTlsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTlsTest.java
index 1f0a11c..9ba131c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTlsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTlsTest.java
@@ -50,6 +50,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "websocket")
public class ProxyPublishConsumeTlsTest extends TlsProducerConsumerBase {
protected String methodName;
@@ -95,7 +96,7 @@ public class ProxyPublishConsumeTlsTest extends TlsProducerConsumerBase {
}
@Test(timeOut = 30000)
- public void socketTest() throws InterruptedException, GeneralSecurityException {
+ public void socketTest() throws GeneralSecurityException {
String consumerUri =
"wss://localhost:" + proxyServer.getListenPortHTTPS().get() + "/ws/consumer/persistent/my-property/use/my-ns/my-topic/my-sub";
String producerUri = "wss://localhost:" + proxyServer.getListenPortHTTPS().get() + "/ws/producer/persistent/my-property/use/my-ns/my-topic/";
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeWithoutZKTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeWithoutZKTest.java
index 9c4e2c8..2f6a623 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeWithoutZKTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeWithoutZKTest.java
@@ -46,6 +46,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "websocket")
public class ProxyPublishConsumeWithoutZKTest extends ProducerConsumerBase {
protected String methodName;
private ProxyServer proxyServer;
@@ -80,7 +81,7 @@ public class ProxyPublishConsumeWithoutZKTest extends ProducerConsumerBase {
log.info("Finished Cleaning Up Test setup");
}
- @Test(timeOut=30000)
+ @Test(timeOut = 30000)
public void socketTest() throws Exception {
String consumerUri = "ws://localhost:" + proxyServer.getListenPortHTTP().get() + "/ws/v2/consumer/persistent/my-property/my-ns/my-topic/my-sub";
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerSocket.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerSocket.java
index e1ddaae..bf3a779 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerSocket.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerSocket.java
@@ -24,7 +24,6 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
-import org.apache.commons.lang3.StringUtils;
import org.eclipse.jetty.websocket.api.RemoteEndpoint;
import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose;
@@ -49,7 +48,7 @@ public class SimpleConsumerSocket {
public SimpleConsumerSocket() {
this.closeLatch = new CountDownLatch(1);
- consumerBuffer = new ArrayList<String>();
+ consumerBuffer = new ArrayList<>();
}
public boolean awaitClose(int duration, TimeUnit unit) throws InterruptedException {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleProducerSocket.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleProducerSocket.java
index db8e5bf..149e46b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleProducerSocket.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleProducerSocket.java
@@ -49,7 +49,7 @@ public class SimpleProducerSocket {
public SimpleProducerSocket() {
this.closeLatch = new CountDownLatch(1);
- producerBuffer = new ArrayList<String>();
+ producerBuffer = new ArrayList<>();
}
private static String getTestJsonPayload(int index) throws JsonProcessingException {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/v1/V1_ProxyAuthenticationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/v1/V1_ProxyAuthenticationTest.java
index 4653741..0ccb4b5 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/v1/V1_ProxyAuthenticationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/v1/V1_ProxyAuthenticationTest.java
@@ -57,6 +57,7 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+@Test(groups = "websocket")
public class V1_ProxyAuthenticationTest extends V1_ProducerConsumerBase {
private ProxyServer proxyServer;
@@ -94,7 +95,7 @@ public class V1_ProxyAuthenticationTest extends V1_ProducerConsumerBase {
}
@AfterMethod(alwaysRun = true)
- protected void cleanup() throws Exception {
+ public void cleanup() throws Exception {
ExecutorService executor = newFixedThreadPool(1);
try {
executor.submit(() -> {
@@ -122,7 +123,7 @@ public class V1_ProxyAuthenticationTest extends V1_ProducerConsumerBase {
}
- public void socketTest() throws Exception {
+ private void socketTest() throws Exception {
final String topic = "prop/use/my-ns/my-topic1";
final String consumerUri = "ws://localhost:" + proxyServer.getListenPortHTTP().get() + "/ws/consumer/persistent/" + topic + "/my-sub";
final String producerUri = "ws://localhost:" + proxyServer.getListenPortHTTP().get() + "/ws/producer/persistent/" + topic;
@@ -151,18 +152,18 @@ public class V1_ProxyAuthenticationTest extends V1_ProducerConsumerBase {
Assert.assertEquals(produceSocket.getBuffer(), consumeSocket.getBuffer());
}
- @Test(timeOut=10000)
+ @Test(timeOut = 10000)
public void authenticatedSocketTest() throws Exception {
socketTest();
}
- @Test(timeOut=10000)
+ @Test(timeOut = 10000)
public void anonymousSocketTest() throws Exception {
socketTest();
}
- @Test(timeOut=10000)
- public void unauthenticatedSocketTest() throws Exception{
+ @Test(timeOut = 10000)
+ public void unauthenticatedSocketTest() {
Exception exception = null;
try {
socketTest();
@@ -172,7 +173,7 @@ public class V1_ProxyAuthenticationTest extends V1_ProducerConsumerBase {
Assert.assertTrue(exception instanceof java.util.concurrent.ExecutionException);
}
- @Test(timeOut=10000)
+ @Test(timeOut = 10000)
public void statsTest() throws Exception {
final String topic = "prop/use/my-ns/my-topic2";
final String consumerUri = "ws://localhost:" + proxyServer.getListenPortHTTP().get() + "/ws/consumer/persistent/" + topic + "/my-sub";
@@ -223,7 +224,7 @@ public class V1_ProxyAuthenticationTest extends V1_ProducerConsumerBase {
private void verifyResponseStatus(Client client, String url) {
WebTarget webTarget = client.target(url);
Invocation.Builder invocationBuilder = webTarget.request(MediaType.APPLICATION_JSON);
- Response response = (Response) invocationBuilder.get();
+ Response response = invocationBuilder.get();
Assert.assertEquals(response.getStatus(), 200);
}