msg2 = consumer2.receive(5, TimeUnit.SECONDS);
if (msg2 != null) {
fail("msg should have not been replicated to remote cluster");
}
@@ -857,7 +860,7 @@ public void testReplicatedCluster() throws Exception {
*
* 1. Create global topic with 4 partitions
* 2. Update partition with 8 partitions
- * 3. Create producer on the partition topic which loads all new partitions
+ * 3. Create producer on the partition topic which loads all new partitions.
* 4. Check subscriptions are created on all new partitions.
*
* @throws Exception
@@ -868,7 +871,7 @@ public void testUpdateGlobalTopicPartition() throws Exception {
final String cluster1 = pulsar1.getConfig().getClusterName();
final String cluster2 = pulsar2.getConfig().getClusterName();
- final String namespace = "pulsar/global/ns3";
+ final String namespace = "pulsar/global/ns3" + randomName(16);
final String topicName = "persistent://" + namespace + "/topic1";
int startPartitions = 4;
int newPartitions = 8;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTlsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTlsTest.java
index 3ad22e9a134fb..5cf88275a3a38 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTlsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTlsTest.java
@@ -33,7 +33,7 @@
public class ReplicatorTlsTest extends ReplicatorTestBase {
@Override
- @BeforeClass(timeOut = 300000)
+ @BeforeClass(timeOut = 900000)
void setup() throws Exception {
config1.setBrokerClientTlsEnabled(true);
config2.setBrokerClientTlsEnabled(true);
@@ -42,7 +42,7 @@ void setup() throws Exception {
}
@Override
- @AfterClass(timeOut = 300000)
+ @AfterClass(timeOut = 900000)
void shutdown() throws Exception {
super.shutdown();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
index ba4cf5bee2815..8aa7163493309 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
@@ -212,7 +212,7 @@ public void teardown() throws Exception {
executor.shutdownNow();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testConnectCommand() throws Exception {
resetChannel();
assertTrue(channel.isActive());
@@ -245,7 +245,7 @@ private static ByteBuf newConnect(AuthMethod authMethod, String authData, int pr
*
* @throws Exception
*/
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testConnectCommandWithEnum() throws Exception {
resetChannel();
assertTrue(channel.isActive());
@@ -260,7 +260,7 @@ public void testConnectCommandWithEnum() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testConnectCommandWithProtocolVersion() throws Exception {
resetChannel();
assertTrue(channel.isActive());
@@ -276,7 +276,7 @@ public void testConnectCommandWithProtocolVersion() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testKeepAlive() throws Exception {
resetChannel();
assertTrue(channel.isActive());
@@ -301,7 +301,7 @@ public void testKeepAlive() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testKeepAliveNotEnforcedWithOlderClients() throws Exception {
resetChannel();
assertTrue(channel.isActive());
@@ -326,7 +326,7 @@ public void testKeepAliveNotEnforcedWithOlderClients() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testKeepAliveBeforeHandshake() throws Exception {
resetChannel();
assertTrue(channel.isActive());
@@ -345,7 +345,7 @@ public void testKeepAliveBeforeHandshake() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testConnectCommandWithAuthenticationPositive() throws Exception {
AuthenticationService authenticationService = mock(AuthenticationService.class);
AuthenticationProvider authenticationProvider = mock(AuthenticationProvider.class);
@@ -380,7 +380,7 @@ public void testConnectCommandWithAuthenticationPositive() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testConnectCommandWithAuthenticationNegative() throws Exception {
AuthenticationService authenticationService = mock(AuthenticationService.class);
doReturn(authenticationService).when(brokerService).getAuthenticationService();
@@ -400,7 +400,7 @@ public void testConnectCommandWithAuthenticationNegative() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testProducerCommand() throws Exception {
resetChannel();
setChannelConnected();
@@ -428,7 +428,7 @@ public void testProducerCommand() throws Exception {
assertEquals(topicRef.getProducers().size(), 0);
}
- @Test(timeOut = 5000)
+ @Test(timeOut = 50000)
public void testDuplicateConcurrentProducerCommand() throws Exception {
resetChannel();
setChannelConnected();
@@ -451,7 +451,7 @@ public void testDuplicateConcurrentProducerCommand() throws Exception {
assertEquals(error.getError(), ServerError.ServiceNotReady);
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testProducerOnNotOwnedTopic() throws Exception {
resetChannel();
setChannelConnected();
@@ -475,7 +475,7 @@ public void testProducerOnNotOwnedTopic() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testProducerCommandWithAuthorizationPositive() throws Exception {
AuthorizationService authorizationService = mock(AuthorizationService.class);
doReturn(CompletableFuture.completedFuture(true)).when(authorizationService).canProduceAsync(Mockito.any(),
@@ -500,7 +500,7 @@ public void testProducerCommandWithAuthorizationPositive() throws Exception {
assertEquals(topicRef.getProducers().size(), 0);
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testNonExistentTopic() throws Exception {
ZooKeeperDataCache zkDataCache = mock(ZooKeeperDataCache.class);
ConfigurationCacheService configCacheService = mock(ConfigurationCacheService.class);
@@ -536,7 +536,7 @@ public void testNonExistentTopic() throws Exception {
assertTrue(getResponse() instanceof CommandError);
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testClusterAccess() throws Exception {
svcConfig.setAuthorizationEnabled(true);
AuthorizationService authorizationService = spy(new AuthorizationService(svcConfig, configCacheService));
@@ -565,7 +565,7 @@ public void testClusterAccess() throws Exception {
assertTrue(getResponse() instanceof CommandError);
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testNonExistentTopicSuperUserAccess() throws Exception {
AuthorizationService authorizationService = spy(new AuthorizationService(svcConfig, configCacheService));
doReturn(authorizationService).when(brokerService).getAuthorizationService();
@@ -622,7 +622,7 @@ public void testProducerCommandWithAuthorizationNegative() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testSendCommand() throws Exception {
resetChannel();
setChannelConnected();
@@ -645,7 +645,7 @@ public void testSendCommand() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testUseSameProducerName() throws Exception {
resetChannel();
setChannelConnected();
@@ -665,7 +665,7 @@ public void testUseSameProducerName() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testRecreateSameProducer() throws Exception {
resetChannel();
setChannelConnected();
@@ -699,7 +699,7 @@ public void testRecreateSameProducer() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testSubscribeMultipleTimes() throws Exception {
resetChannel();
setChannelConnected();
@@ -756,7 +756,7 @@ public void testDuplicateConcurrentSubscribeCommand() throws Exception {
assertEquals(error.getError(), ServerError.ServiceNotReady);
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testCreateProducerTimeout() throws Exception {
resetChannel();
setChannelConnected();
@@ -891,7 +891,7 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
channel.finish();
}
- @Test(timeOut = 30000, invocationCount = 1, skipFailedInvocations = true)
+ @Test(timeOut = 90000, invocationCount = 1, skipFailedInvocations = true)
public void testCreateProducerBookieTimeout() throws Exception {
resetChannel();
setChannelConnected();
@@ -961,7 +961,7 @@ public void testCreateProducerBookieTimeout() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testSubscribeTimeout() throws Exception {
resetChannel();
setChannelConnected();
@@ -1035,7 +1035,7 @@ public void testSubscribeTimeout() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testSubscribeBookieTimeout() throws Exception {
resetChannel();
setChannelConnected();
@@ -1120,7 +1120,7 @@ public void testSubscribeBookieTimeout() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testSubscribeCommand() throws Exception {
final String failSubName = "failSub";
@@ -1159,7 +1159,7 @@ public void testSubscribeCommand() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testUnsupportedBatchMsgSubscribeCommand() throws Exception {
final String failSubName = "failSub";
@@ -1192,7 +1192,7 @@ public void testUnsupportedBatchMsgSubscribeCommand() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testSubscribeCommandWithAuthorizationPositive() throws Exception {
AuthorizationService authorizationService = mock(AuthorizationService.class);
doReturn(CompletableFuture.completedFuture(true)).when(authorizationService).canConsumeAsync(Mockito.any(),
@@ -1214,7 +1214,7 @@ public void testSubscribeCommandWithAuthorizationPositive() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testSubscribeCommandWithAuthorizationNegative() throws Exception {
AuthorizationService authorizationService = mock(AuthorizationService.class);
doReturn(CompletableFuture.completedFuture(false)).when(authorizationService).canConsumeAsync(Mockito.any(),
@@ -1235,7 +1235,7 @@ public void testSubscribeCommandWithAuthorizationNegative() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testAckCommand() throws Exception {
resetChannel();
setChannelConnected();
@@ -1258,7 +1258,7 @@ public void testAckCommand() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testFlowCommand() throws Exception {
resetChannel();
setChannelConnected();
@@ -1278,7 +1278,7 @@ public void testFlowCommand() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testProducerSuccessOnEncryptionRequiredTopic() throws Exception {
resetChannel();
setChannelConnected();
@@ -1306,7 +1306,7 @@ public void testProducerSuccessOnEncryptionRequiredTopic() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testProducerFailureOnEncryptionRequiredTopic() throws Exception {
resetChannel();
setChannelConnected();
@@ -1336,7 +1336,7 @@ public void testProducerFailureOnEncryptionRequiredTopic() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testSendSuccessOnEncryptionRequiredTopic() throws Exception {
resetChannel();
setChannelConnected();
@@ -1371,7 +1371,7 @@ public void testSendSuccessOnEncryptionRequiredTopic() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testSendFailureOnEncryptionRequiredTopic() throws Exception {
resetChannel();
setChannelConnected();
@@ -1552,7 +1552,7 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
doReturn(successSubName).when(cursorMock).getName();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testInvalidTopicOnLookup() throws Exception {
resetChannel();
setChannelConnected();
@@ -1572,7 +1572,7 @@ public void testInvalidTopicOnLookup() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testInvalidTopicOnProducer() throws Exception {
resetChannel();
setChannelConnected();
@@ -1593,7 +1593,7 @@ public void testInvalidTopicOnProducer() throws Exception {
channel.finish();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testInvalidTopicOnSubscribe() throws Exception {
resetChannel();
setChannelConnected();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/DelayedDeliveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/DelayedDeliveryTest.java
index 3cd2e255e7920..4440f1b98a3b3 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/DelayedDeliveryTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/DelayedDeliveryTest.java
@@ -88,7 +88,7 @@ public void testDelayedDelivery()
// Failover consumer will receive the messages immediately while
// the shared consumer will get them after the delay
- Message msg = sharedConsumer.receive(100, TimeUnit.MILLISECONDS);
+ Message msg = sharedConsumer.receive(10, TimeUnit.MILLISECONDS);
assertNull(msg);
for (int i = 0; i < 10; i++) {
@@ -194,7 +194,7 @@ public void testEverythingFilteredInMultipleReads()
.send();
}
- Message msg = sharedConsumer.receive(100, TimeUnit.MILLISECONDS);
+ Message msg = sharedConsumer.receive(10, TimeUnit.MILLISECONDS);
assertNull(msg);
Set receivedMsgs = new TreeSet<>();
@@ -256,7 +256,7 @@ public void testDelayedDeliveryWithMultipleConcurrentReadEntries()
producer.flush();
- Message msg = consumer.receive(100, TimeUnit.MILLISECONDS);
+ Message msg = consumer.receive(10, TimeUnit.MILLISECONDS);
assertNull(msg);
Set receivedMsgs = new TreeSet<>();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/PartitionedTopicsSchemaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/PartitionedTopicsSchemaTest.java
index 93e41dd11e7af..69173071b902e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/PartitionedTopicsSchemaTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/PartitionedTopicsSchemaTest.java
@@ -25,6 +25,7 @@
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
import org.apache.pulsar.broker.service.BkEnsemblesTestBase;
import org.apache.pulsar.client.api.Consumer;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ManagedLedgerMetricsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ManagedLedgerMetricsTest.java
index f74f192385210..d67512f6abd92 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ManagedLedgerMetricsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ManagedLedgerMetricsTest.java
@@ -34,8 +34,6 @@
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
-/**
- */
public class ManagedLedgerMetricsTest extends BrokerTestBase {
@BeforeClass
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
index e569cafdf5044..ca13acdb78151 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
@@ -29,8 +29,10 @@
import java.util.Map;
import java.util.HashMap;
import java.util.TreeMap;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import java.util.stream.Collectors;
import org.apache.pulsar.broker.service.BrokerTestBase;
import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsGenerator;
@@ -57,10 +59,21 @@ protected void cleanup() throws Exception {
super.internalCleanup();
}
+ // Perhaps it would be better to import this method from PulsarTestBase in tests.integration
+ public static String randomName(int numChars) {
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < numChars; i++) {
+ sb.append((char) (ThreadLocalRandom.current().nextInt(26) + 'a'));
+ }
+ return sb.toString();
+ }
+
@Test
public void testPerTopicStats() throws Exception {
- Producer p1 = pulsarClient.newProducer().topic("persistent://my-property/use/my-ns/my-topic1").create();
- Producer p2 = pulsarClient.newProducer().topic("persistent://my-property/use/my-ns/my-topic2").create();
+ String randSeed = randomName(16);
+ System.out.println("The randSeed of testPerTopicStats() is: " + randSeed);
+ Producer p1 = pulsarClient.newProducer().topic("persistent://my-property/use/" + randSeed + "/my-topic1").create();
+ Producer p2 = pulsarClient.newProducer().topic("persistent://my-property/use/" + randSeed + "/my-topic2").create();
for (int i = 0; i < 10; i++) {
String message = "my-message-" + i;
p1.send(message.getBytes());
@@ -78,36 +91,95 @@ public void testPerTopicStats() throws Exception {
// There should be 2 metrics with different tags for each topic
List cm = (List) metrics.get("pulsar_storage_write_latency_le_1");
- assertEquals(cm.size(), 2);
- assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
- assertEquals(cm.get(0).tags.get("namespace"), "my-property/use/my-ns");
- assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
- assertEquals(cm.get(1).tags.get("namespace"), "my-property/use/my-ns");
+ List matchingMetrics = cm.stream().filter(t -> t.tags.containsValue("my-property/use/" + randSeed)).collect(Collectors.toList());
+ // I think there's a race that's causing cm.get(0) and cm.get(1) to swap items...
+ int positionOfTopic1;
+ int positionOfTopic2;
+
+ matchingMetrics = cm.stream().filter(t -> t.tags.containsValue("my-property/use/" + randSeed)).collect(Collectors.toList());
+ if(matchingMetrics.size() > 2){
+ System.out.println("matchingMetrics.size() > 2 in testPerTopicStats(). First check. Debug entries: ");
+ matchingMetrics.forEach(t -> t.tags.entrySet().forEach(kv -> System.out.println(kv.getKey() + ":" + kv.getValue())));
+ }
+ if(matchingMetrics.get(0).tags.get("topic").equals("persistent://my-property/use/" + randSeed + "/my-topic1")) {
+ positionOfTopic1 = 0;
+ positionOfTopic2 = 1;
+ } else {
+ positionOfTopic2 = 0;
+ positionOfTopic1 = 1;
+ }
+ assertEquals(matchingMetrics.size(), 2);
+ assertEquals(matchingMetrics.get(positionOfTopic2).tags.get("topic"), "persistent://my-property/use/" + randSeed + "/my-topic2");
+ assertEquals(matchingMetrics.get(positionOfTopic2).tags.get("namespace"), "my-property/use/" + randSeed);
+ assertEquals(matchingMetrics.get(positionOfTopic1).tags.get("topic"), "persistent://my-property/use/" + randSeed + "/my-topic1");
+ assertEquals(matchingMetrics.get(positionOfTopic1).tags.get("namespace"), "my-property/use/" + randSeed);
cm = (List) metrics.get("pulsar_producers_count");
- assertEquals(cm.size(), 3);
- assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
- assertEquals(cm.get(1).tags.get("namespace"), "my-property/use/my-ns");
- assertEquals(cm.get(2).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
- assertEquals(cm.get(2).tags.get("namespace"), "my-property/use/my-ns");
+
+ matchingMetrics = cm.stream().filter(t -> t.tags.containsValue("my-property/use/" + randSeed)).collect(Collectors.toList());
+ if(matchingMetrics.size() > 2){
+ System.out.println("matchingMetrics.size() > 2 in testPerTopicStats(). Second check. Debug entries: ");
+ matchingMetrics.forEach(t -> t.tags.entrySet().forEach(kv -> System.out.println(kv.getKey() + ":" + kv.getValue())));
+ }
+ if(matchingMetrics.get(1).tags.get("topic").equals("persistent://my-property/use/" + randSeed + "/my-topic1")) {
+ positionOfTopic1 = 1;
+ positionOfTopic2 = 0;
+ } else {
+ positionOfTopic2 = 1;
+ positionOfTopic1 = 0;
+ }
+ assertEquals(matchingMetrics.size(), 2);
+ assertEquals(matchingMetrics.get(positionOfTopic2).tags.get("topic"), "persistent://my-property/use/" + randSeed + "/my-topic2");
+ assertEquals(matchingMetrics.get(positionOfTopic2).tags.get("namespace"), "my-property/use/" + randSeed);
+ assertEquals(matchingMetrics.get(positionOfTopic1).tags.get("topic"), "persistent://my-property/use/" + randSeed + "/my-topic1");
+ assertEquals(matchingMetrics.get(positionOfTopic1).tags.get("namespace"), "my-property/use/" + randSeed);
cm = (List) metrics.get("topic_load_times_count");
+ //matchingMetrics = cm.stream().filter(t -> t.tags.containsValue("my-property/use/" + randSeed)).collect(Collectors.toList());
+ if(cm.size() > 1){ // This one doesn't have the entire topic name... Hopefully, that won't cause concurrency problems.
+ System.out.println("matchingMetrics.size() > 2 in testPerTopicStats(). Third check. Debug entries: ");
+ cm.forEach(t -> t.tags.entrySet().forEach(kv -> System.out.println(kv.getKey() + ":" + kv.getValue())));
+ }
assertEquals(cm.size(), 1);
assertEquals(cm.get(0).tags.get("cluster"), "test");
cm = (List) metrics.get("pulsar_in_bytes_total");
- assertEquals(cm.size(), 2);
- assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
- assertEquals(cm.get(0).tags.get("namespace"), "my-property/use/my-ns");
- assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
- assertEquals(cm.get(1).tags.get("namespace"), "my-property/use/my-ns");
+ matchingMetrics = cm.stream().filter(t -> t.tags.containsValue("my-property/use/" + randSeed)).collect(Collectors.toList());
+ if(matchingMetrics.size() > 2){
+ System.out.println("matchingMetrics.size() > 2 in testPerTopicStats(). Fourth check. Debug entries: ");
+ matchingMetrics.forEach(t -> t.tags.entrySet().forEach(kv -> System.out.println(kv.getKey() + ":" + kv.getValue())));
+ }
+ if(matchingMetrics.get(0).tags.get("topic").equals("persistent://my-property/use/" + randSeed + "/my-topic1")) {
+ positionOfTopic1 = 0;
+ positionOfTopic2 = 1;
+ } else {
+ positionOfTopic2 = 0;
+ positionOfTopic1 = 1;
+ }
+ assertEquals(matchingMetrics.size(), 2);
+ assertEquals(matchingMetrics.get(positionOfTopic2).tags.get("topic"), "persistent://my-property/use/" + randSeed + "/my-topic2");
+ assertEquals(matchingMetrics.get(positionOfTopic2).tags.get("namespace"), "my-property/use/" + randSeed);
+ assertEquals(matchingMetrics.get(positionOfTopic1).tags.get("topic"), "persistent://my-property/use/" + randSeed + "/my-topic1");
+ assertEquals(matchingMetrics.get(positionOfTopic1).tags.get("namespace"), "my-property/use/" + randSeed);
cm = (List) metrics.get("pulsar_in_messages_total");
- assertEquals(cm.size(), 2);
- assertEquals(cm.get(0).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic2");
- assertEquals(cm.get(0).tags.get("namespace"), "my-property/use/my-ns");
- assertEquals(cm.get(1).tags.get("topic"), "persistent://my-property/use/my-ns/my-topic1");
- assertEquals(cm.get(1).tags.get("namespace"), "my-property/use/my-ns");
+ matchingMetrics = cm.stream().filter(t -> t.tags.containsValue("my-property/use/" + randSeed)).collect(Collectors.toList());
+ if(matchingMetrics.size() > 2){
+ System.out.println("matchingMetrics.size() > 2 in testPerTopicStats(). Fifth check. Debug entries: ");
+ matchingMetrics.forEach(t -> t.tags.entrySet().forEach(kv -> System.out.println(kv.getKey() + ":" + kv.getValue())));
+ }
+ if(matchingMetrics.get(0).tags.get("topic").equals("persistent://my-property/use/" + randSeed + "/my-topic1")) {
+ positionOfTopic1 = 0;
+ positionOfTopic2 = 1;
+ } else {
+ positionOfTopic2 = 0;
+ positionOfTopic1 = 1;
+ }
+ assertEquals(matchingMetrics.size(), 2);
+ assertEquals(matchingMetrics.get(positionOfTopic2).tags.get("topic"), "persistent://my-property/use/" + randSeed + "/my-topic2");
+ assertEquals(matchingMetrics.get(positionOfTopic2).tags.get("namespace"), "my-property/use/" + randSeed);
+ assertEquals(matchingMetrics.get(positionOfTopic1).tags.get("topic"), "persistent://my-property/use/" + randSeed + "/my-topic1");
+ assertEquals(matchingMetrics.get(positionOfTopic1).tags.get("namespace"), "my-property/use/" + randSeed);
p1.close();
p2.close();
@@ -115,8 +187,10 @@ public void testPerTopicStats() throws Exception {
@Test
public void testPerNamespaceStats() throws Exception {
- Producer p1 = pulsarClient.newProducer().topic("persistent://my-property/use/my-ns/my-topic1").create();
- Producer p2 = pulsarClient.newProducer().topic("persistent://my-property/use/my-ns/my-topic2").create();
+ String randSeed = randomName(16);
+ System.out.println("The randSeed of testPerNamespaceStats() is: " + randSeed);
+ Producer p1 = pulsarClient.newProducer().topic("persistent://my-property/use/" + randSeed + "/my-topic1").create();
+ Producer p2 = pulsarClient.newProducer().topic("persistent://my-property/use/" + randSeed + "/my-topic2").create();
for (int i = 0; i < 10; i++) {
String message = "my-message-" + i;
p1.send(message.getBytes());
@@ -135,15 +209,29 @@ public void testPerNamespaceStats() throws Exception {
// There should be 1 metric aggregated per namespace
List cm = (List) metrics.get("pulsar_storage_write_latency_le_1");
- assertEquals(cm.size(), 1);
- assertNull(cm.get(0).tags.get("topic"));
- assertEquals(cm.get(0).tags.get("namespace"), "my-property/use/my-ns");
+ // Need to filter out Prometheus metrics like the way we're doing it in the Go Functions API...
+ // Maybe there's a workaround for now.
+ List matchingMetrics = cm.stream().filter(t -> t.tags.containsValue("my-property/use/" + randSeed)).collect(Collectors.toList());
+ if(matchingMetrics.size() > 1){
+ System.out.println("matchingMetrics.size() > 1 in testPerNamespaceStats(). First check. Debug entries: ");
+ matchingMetrics.forEach(t -> t.tags.entrySet().forEach(kv -> System.out.println(kv.getKey() + ":" + kv.getValue())));
+ }
+ assertEquals(matchingMetrics.size(), 1);
+ assertNull(matchingMetrics.get(0).tags.get("topic"), "Test");
+ assertEquals(matchingMetrics.get(0).tags.get("namespace"), "my-property/use/" + randSeed);
+ // Need to filter out Prometheus metrics like the way we're doing it in the Go Functions API...
+ // Maybe there's a workaround for now.
cm = (List) metrics.get("pulsar_producers_count");
- assertEquals(cm.size(), 2);
- assertEquals(cm.get(1).value, 2.0);
- assertNull(cm.get(1).tags.get("topic"));
- assertEquals(cm.get(1).tags.get("namespace"), "my-property/use/my-ns");
+ matchingMetrics = cm.stream().filter(t -> t.tags.containsValue("my-property/use/" + randSeed)).collect(Collectors.toList());
+ if(matchingMetrics.size() > 1){
+ System.out.println("matchingMetrics.size() > 1 in testPerNamespaceStats(). Second check. Debug entries: ");
+ matchingMetrics.forEach(t -> t.tags.entrySet().forEach(kv -> System.out.println(kv.getKey() + ":" + kv.getValue())));
+ }
+ assertEquals(matchingMetrics.size(), 1);
+ assertEquals(matchingMetrics.get(0).value, 2.0);
+ assertNull(matchingMetrics.get(0).tags.get("topic"));
+ assertEquals(matchingMetrics.get(0).tags.get("namespace"), "my-property/use/" + randSeed);
p1.close();
p2.close();
@@ -264,6 +352,10 @@ private static Multimap parseMetrics(String metrics) {
}
Matcher matcher = pattern.matcher(line);
+ if(!matcher.matches()){
+ System.out.println("Failing line is: " + line);
+ System.out.println("Failing pattern is: " + pattern.toString());
+ }
assertTrue(matcher.matches());
String name = matcher.group(1);
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 09bfb79ccdcd1..98adeada995e5 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
@@ -24,6 +24,7 @@
import static org.testng.Assert.fail;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import org.testng.annotations.AfterClass;
@@ -46,11 +47,11 @@ protected void cleanup() throws Exception {
@Test
public void testProducerSequenceAfterReconnect() throws Exception {
- String topic = "persistent://my-property/my-ns/testProducerSequenceAfterReconnect";
+ String topic = "persistent://my-property/my-ns/testProducerSequenceAfterReconnect"+ randomName(16);
admin.namespaces().setDeduplicationStatus("my-property/my-ns", true);
ProducerBuilder producerBuilder = pulsarClient.newProducer().topic(topic)
- .producerName("my-producer-name");
+ .producerName("my-producer-name"+ randomName(16));
Producer producer = producerBuilder.create();
assertEquals(producer.getLastSequenceId(), -1L);
@@ -77,11 +78,11 @@ public void testProducerSequenceAfterReconnect() throws Exception {
@Test
public void testProducerSequenceAfterRestart() throws Exception {
- String topic = "persistent://my-property/my-ns/testProducerSequenceAfterRestart";
+ String topic = "persistent://my-property/my-ns/testProducerSequenceAfterRestart"+ randomName(16);
admin.namespaces().setDeduplicationStatus("my-property/my-ns", true);
ProducerBuilder producerBuilder = pulsarClient.newProducer().topic(topic)
- .producerName("my-producer-name");
+ .producerName("my-producer-name"+ randomName(16));
Producer producer = producerBuilder.create();
assertEquals(producer.getLastSequenceId(), -1L);
@@ -109,19 +110,19 @@ public void testProducerSequenceAfterRestart() throws Exception {
producer.close();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testProducerDeduplication() throws Exception {
- String topic = "persistent://my-property/my-ns/testProducerDeduplication";
+ String topic = "persistent://my-property/my-ns/testProducerDeduplication"+ randomName(16);
admin.namespaces().setDeduplicationStatus("my-property/my-ns", true);
// Set infinite timeout
ProducerBuilder producerBuilder = pulsarClient.newProducer().topic(topic)
- .producerName("my-producer-name").sendTimeout(0, TimeUnit.SECONDS);
+ .producerName("my-producer-name" + randomName(16)).sendTimeout(0, TimeUnit.SECONDS);
Producer producer = producerBuilder.create();
assertEquals(producer.getLastSequenceId(), -1L);
- Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-subscription")
+ Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-subscription"+ randomName(16))
.subscribe();
producer.newMessage().value("my-message-0".getBytes()).sequenceId(0).send();
@@ -141,7 +142,7 @@ public void testProducerDeduplication() throws Exception {
}
// No other messages should be received
- Message msg = consumer.receive(1, TimeUnit.SECONDS);
+ Message msg = consumer.receive(10, TimeUnit.MILLISECONDS);
assertNull(msg);
// Kill and restart broker
@@ -154,25 +155,32 @@ public void testProducerDeduplication() throws Exception {
producer.newMessage().value("my-message-1".getBytes()).sequenceId(1).send();
producer.newMessage().value("my-message-2".getBytes()).sequenceId(2).send();
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(10, TimeUnit.MILLISECONDS);
assertNull(msg);
producer.close();
}
+ public static String randomName(int numChars) {
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < numChars; i++) {
+ sb.append((char) (ThreadLocalRandom.current().nextInt(26) + 'a'));
+ }
+ return sb.toString();
+ }
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testProducerDeduplicationWithDiscontinuousSequenceId() throws Exception {
- String topic = "persistent://my-property/my-ns/testProducerDeduplicationWithDiscontinuousSequenceId";
+ String topic = "persistent://my-property/my-ns/testProducerDeduplicationWithDiscontinuousSequenceId"+ randomName(16);
admin.namespaces().setDeduplicationStatus("my-property/my-ns", true);
// Set infinite timeout
ProducerBuilder producerBuilder = pulsarClient.newProducer().topic(topic)
- .producerName("my-producer-name").enableBatching(true).batchingMaxMessages(10).sendTimeout(0, TimeUnit.SECONDS);
+ .producerName("my-producer-name" + randomName(16)).enableBatching(true).batchingMaxMessages(10).sendTimeout(0, TimeUnit.SECONDS);
Producer producer = producerBuilder.create();
assertEquals(producer.getLastSequenceId(), -1L);
- Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-subscription")
+ Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-subscription" + randomName(16))
.subscribe();
producer.newMessage().value("my-message-0".getBytes()).sequenceId(2).sendAsync();
@@ -194,7 +202,7 @@ public void testProducerDeduplicationWithDiscontinuousSequenceId() throws Except
}
// No other messages should be received
- Message msg = consumer.receive(1, TimeUnit.SECONDS);
+ Message msg = consumer.receive(10, TimeUnit.MILLISECONDS);
assertNull(msg);
producer.close();
@@ -209,25 +217,25 @@ public void testProducerDeduplicationWithDiscontinuousSequenceId() throws Except
producer.newMessage().value("my-message-2".getBytes()).sequenceId(4).sendAsync();
producer.flush();
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(100, TimeUnit.MILLISECONDS);
assertNull(msg);
producer.close();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testProducerDeduplicationNonBatchAsync() throws Exception {
- String topic = "persistent://my-property/my-ns/testProducerDeduplicationNonBatchAsync";
+ String topic = "persistent://my-property/my-ns/testProducerDeduplicationNonBatchAsync"+ randomName(16);
admin.namespaces().setDeduplicationStatus("my-property/my-ns", true);
// Set infinite timeout
ProducerBuilder producerBuilder = pulsarClient.newProducer().topic(topic)
- .producerName("my-producer-name").enableBatching(false).sendTimeout(0, TimeUnit.SECONDS);
+ .producerName("my-producer-name" + randomName(16)).enableBatching(false).sendTimeout(0, TimeUnit.SECONDS);
Producer producer = producerBuilder.create();
assertEquals(producer.getLastSequenceId(), -1L);
- Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-subscription")
+ Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-subscription"+ randomName(16))
.subscribe();
producer.newMessage().value("my-message-0".getBytes()).sequenceId(2).sendAsync();
@@ -246,7 +254,7 @@ public void testProducerDeduplicationNonBatchAsync() throws Exception {
}
// No other messages should be received
- Message msg = consumer.receive(1, TimeUnit.SECONDS);
+ Message msg = consumer.receive(10, TimeUnit.MILLISECONDS);
assertNull(msg);
// Kill and restart broker
@@ -259,7 +267,7 @@ public void testProducerDeduplicationNonBatchAsync() throws Exception {
producer.newMessage().value("my-message-1".getBytes()).sequenceId(2).sendAsync();
producer.newMessage().value("my-message-2".getBytes()).sequenceId(4).sendAsync();
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(10, TimeUnit.MILLISECONDS);
assertNull(msg);
producer.close();
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 b1f1ddd6fd849..c069e4c340902 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
@@ -106,7 +106,7 @@ public void testDeadLetterTopic() throws Exception {
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
.subscribe();
- Message checkMessage = checkConsumer.receive(3, TimeUnit.SECONDS);
+ Message checkMessage = checkConsumer.receive(10, TimeUnit.MILLISECONDS);
if (checkMessage != null) {
log.info("check consumer received message : {} {}", checkMessage.getMessageId(), new String(checkMessage.getData()));
}
@@ -288,7 +288,7 @@ public void testDeadLetterWithoutConsumerReceiveImmediately() throws PulsarClien
// Wait a while, message should not be send to DLQ
Thread.sleep(5000L);
- Message msg = consumer.receive(1, TimeUnit.SECONDS);
+ Message msg = consumer.receive(5, TimeUnit.SECONDS);
assertNotNull(msg);
}
}
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 5cb4fc1383dae..7507ac7d18ba4 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
@@ -670,12 +670,12 @@ public void testBrokerSubscriptionRecovery(boolean unloadBundleGracefully) throw
* acking a. as consumer is acking not reached maxUnAckPerDispatcher=20 unack msg => consumes all produced msgs
* 4.Subscription-1 : acks all pending msgs and consume by acking a. broker unblocks all dispatcher and sub-1
* consumes all messages 5. Subscription-2 : it triggers redelivery and acks all messages so, it consumes all
- * produced messages
+ * produced messages.
*
*
* @throws Exception
*/
- @Test(timeOut = 10000)
+ @Test(timeOut = 190000)
public void testBlockBrokerDispatching() throws Exception {
log.info("-- Starting {} test --", methodName);
@@ -743,7 +743,7 @@ public void testBlockBrokerDispatching() throws Exception {
Message msg = null;
Set messages1 = Sets.newHashSet();
for (int j = 0; j < totalProducedMsgs; j++) {
- msg = consumer1Sub1.receive(100, TimeUnit.MILLISECONDS);
+ msg = consumer1Sub1.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages1.add(msg.getMessageId());
} else {
@@ -752,7 +752,7 @@ public void testBlockBrokerDispatching() throws Exception {
// once consumer receives maxUnAckPerBroker-msgs then sleep to give a chance to scheduler to block the
// subscription
if (j == maxUnAckPerBroker) {
- Thread.sleep(200);
+ Thread.sleep(1000);
}
}
// client must receive number of messages = maxUnAckPerbroker rather all produced messages
@@ -764,7 +764,7 @@ public void testBlockBrokerDispatching() throws Exception {
.subscriptionType(SubscriptionType.Shared).acknowledgmentGroupTime(0, TimeUnit.SECONDS).subscribe();
int consumer2Msgs = 0;
for (int j = 0; j < totalProducedMsgs; j++) {
- msg = consumer2Sub1.receive(100, TimeUnit.MILLISECONDS);
+ msg = consumer2Sub1.receive(5, TimeUnit.SECONDS);
if (msg != null) {
consumer2Msgs++;
} else {
@@ -789,7 +789,7 @@ public void testBlockBrokerDispatching() throws Exception {
.subscriptionType(SubscriptionType.Shared).acknowledgmentGroupTime(0, TimeUnit.SECONDS).subscribe();
Set messages2 = Sets.newHashSet();
for (int j = 0; j < totalProducedMsgs; j++) {
- msg = consumerSub2.receive(100, TimeUnit.MILLISECONDS);
+ msg = consumerSub2.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages2.add(msg.getMessageId());
} else {
@@ -806,7 +806,7 @@ public void testBlockBrokerDispatching() throws Exception {
.subscriptionType(SubscriptionType.Shared).acknowledgmentGroupTime(0, TimeUnit.SECONDS).subscribe();
int consumedMsgsSub3 = 0;
for (int j = 0; j < totalProducedMsgs; j++) {
- msg = consumer1Sub3.receive(100, TimeUnit.MILLISECONDS);
+ msg = consumer1Sub3.receive(5, TimeUnit.SECONDS);
if (msg != null) {
consumedMsgsSub3++;
consumer1Sub3.acknowledge(msg);
@@ -822,7 +822,7 @@ public void testBlockBrokerDispatching() throws Exception {
// sleep so, broker receives all ack back to unblock subscription
Thread.sleep(1000);
for (int j = 0; j < totalProducedMsgs; j++) {
- msg = consumer1Sub1.receive(1, TimeUnit.SECONDS);
+ msg = consumer1Sub1.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages1.add(msg.getMessageId());
consumer1Sub1.acknowledge(msg);
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 8ebbca0be23d6..a8e88196d435c 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
@@ -43,7 +43,7 @@ protected void cleanup() throws Exception {
super.internalCleanup();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testRedeliveryCount() throws PulsarClientException {
final String topic = "persistent://my-property/my-ns/redeliveryCount";
@@ -78,7 +78,7 @@ public void testRedeliveryCount() throws PulsarClientException {
consumer.close();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testRedeliveryCountWithPartitionedTopic() throws PulsarClientException, PulsarAdminException {
final String topic = "persistent://my-property/my-ns/redeliveryCount.partitioned";
@@ -117,7 +117,7 @@ public void testRedeliveryCountWithPartitionedTopic() throws PulsarClientExcepti
admin.topics().deletePartitionedTopic(topic);
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 190000)
public void testRedeliveryCountWhenConsumerDisconnected() throws PulsarClientException, InterruptedException {
String topic = "persistent://my-property/my-ns/testRedeliveryCountWhenConsumerDisconnected";
@@ -150,7 +150,7 @@ public void testRedeliveryCountWhenConsumerDisconnected() throws PulsarClientExc
List> receivedMessagesForConsumer1 = new ArrayList<>();
for (int i = 0; i < messages; i++) {
- Message msg = consumer0.receive(1, TimeUnit.SECONDS);
+ Message msg = consumer0.receive(5, TimeUnit.SECONDS);
if (msg != null) {
receivedMessagesForConsumer0.add(msg);
} else {
@@ -159,7 +159,7 @@ public void testRedeliveryCountWhenConsumerDisconnected() throws PulsarClientExc
}
for (int i = 0; i < messages; i++) {
- Message msg = consumer1.receive(1, TimeUnit.SECONDS);
+ Message msg = consumer1.receive(5, TimeUnit.SECONDS);
if (msg != null) {
receivedMessagesForConsumer1.add(msg);
} else {
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 9232642c02f9b..3855390f771b6 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
@@ -119,7 +119,7 @@ public void testNonPersistentTopic(SubscriptionType type) throws Exception {
Message> msg = null;
Set messageSet = Sets.newHashSet();
for (int i = 0; i < totalProduceMsg; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
consumer.acknowledge(msg);
String receivedMessage = new String(msg.getData());
@@ -162,7 +162,7 @@ public void testPartitionedNonPersistentTopic(SubscriptionType type) throws Exce
Message> msg = null;
Set messageSet = Sets.newHashSet();
for (int i = 0; i < totalProduceMsg; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
consumer.acknowledge(msg);
String receivedMessage = new String(msg.getData());
@@ -217,7 +217,7 @@ public void testPartitionedNonPersistentTopicWithTcpLookup(SubscriptionType type
Message> msg = null;
Set messageSet = Sets.newHashSet();
for (int i = 0; i < totalProduceMsg; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
consumer.acknowledge(msg);
String receivedMessage = new String(msg.getData());
@@ -260,7 +260,7 @@ public void testConsumerInternalQueueMaxOut(SubscriptionType type) throws Except
Message> msg = null;
Set messageSet = Sets.newHashSet();
for (int i = 0; i < totalProduceMsg; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
consumer.acknowledge(msg);
String receivedMessage = new String(msg.getData());
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 982005640b55c..be9d2f37687da 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
@@ -69,7 +69,7 @@ protected void cleanup() throws Exception {
executor.shutdown();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testRoundRobinProducer() throws Exception {
log.info("-- Starting {} test --", methodName);
PulsarClient pulsarClient = newPulsarClient(lookupUrl.toString(), 0);// Creates new client connection
@@ -95,7 +95,7 @@ public void testRoundRobinProducer() throws Exception {
Message msg;
Set messageSet = Sets.newHashSet();
for (int i = 0; i < 10; i++) {
- msg = consumer.receive(5, TimeUnit.SECONDS);
+ msg = consumer.receive();
Assert.assertNotNull(msg, "Message should not be null");
consumer.acknowledge(msg);
String receivedMessage = new String(msg.getData());
@@ -112,7 +112,7 @@ public void testRoundRobinProducer() throws Exception {
log.info("-- Exiting {} test --", methodName);
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testPartitionedTopicNameWithSpecialCharacter() throws Exception {
log.info("-- Starting {} test --", methodName);
@@ -130,7 +130,7 @@ public void testPartitionedTopicNameWithSpecialCharacter() throws Exception {
log.info("-- Exiting {} test --", methodName);
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testCustomPartitionProducer() throws Exception {
PulsarClient pulsarClient = newPulsarClient(lookupUrl.toString(), 0);// Creates new client connection
TopicName topicName = null;
@@ -162,7 +162,7 @@ public void testCustomPartitionProducer() throws Exception {
Set messageSet = Sets.newHashSet();
for (int i = 0; i < MESSAGE_COUNT; i++) {
- msg = consumer.receive(5, TimeUnit.SECONDS);
+ msg = consumer.receive();
Assert.assertNotNull(msg, "Message should not be null");
consumer.acknowledge(msg);
String receivedMessage = new String(msg.getData());
@@ -181,7 +181,7 @@ public void testCustomPartitionProducer() throws Exception {
}
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testSinglePartitionProducer() throws Exception {
log.info("-- Starting {} test --", methodName);
PulsarClient pulsarClient = newPulsarClient(lookupUrl.toString(), 0);// Creates new client connection
@@ -207,7 +207,7 @@ public void testSinglePartitionProducer() throws Exception {
Set messageSet = Sets.newHashSet();
for (int i = 0; i < 10; i++) {
- msg = consumer.receive(5, TimeUnit.SECONDS);
+ msg = consumer.receive();
Assert.assertNotNull(msg, "Message should not be null");
consumer.acknowledge(msg);
String receivedMessage = new String(msg.getData());
@@ -225,7 +225,7 @@ public void testSinglePartitionProducer() throws Exception {
log.info("-- Exiting {} test --", methodName);
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testKeyBasedProducer() throws Exception {
log.info("-- Starting {} test --", methodName);
PulsarClient pulsarClient = newPulsarClient(lookupUrl.toString(), 0);// Creates new client connection
@@ -253,7 +253,7 @@ public void testKeyBasedProducer() throws Exception {
Set messageSet = Sets.newHashSet();
for (int i = 0; i < 10; i++) {
- Message msg = consumer.receive(5, TimeUnit.SECONDS);
+ Message msg = consumer.receive();
Assert.assertNotNull(msg, "Message should not be null");
consumer.acknowledge(msg);
String receivedMessage = new String(msg.getData());
@@ -334,7 +334,7 @@ public void testPauseAndResume() throws Exception {
log.info("-- Exiting {} test --", methodName);
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testInvalidSequence() throws Exception {
log.info("-- Starting {} test --", methodName);
@@ -389,7 +389,7 @@ public void testInvalidSequence() throws Exception {
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testSillyUser() throws Exception {
int numPartitions = 4;
@@ -433,7 +433,7 @@ public void testSillyUser() throws Exception {
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testDeletePartitionedTopic() throws Exception {
int numPartitions = 4;
TopicName topicName = TopicName
@@ -455,7 +455,7 @@ public void testDeletePartitionedTopic() throws Exception {
}
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testAsyncPartitionedProducerConsumer() throws Exception {
log.info("-- Starting {} test --", methodName);
@@ -504,7 +504,7 @@ public void testAsyncPartitionedProducerConsumer() throws Exception {
log.info("-- Exiting {} test --", methodName);
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testAsyncPartitionedProducerConsumerQueueSizeOne() throws Exception {
log.info("-- Starting {} test --", methodName);
PulsarClient pulsarClient = newPulsarClient(lookupUrl.toString(), 0);// Creates new client connection
@@ -561,7 +561,7 @@ public void testAsyncPartitionedProducerConsumerQueueSizeOne() throws Exception
*
* @throws Exception
*/
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testFairDistributionForPartitionConsumers() throws Exception {
log.info("-- Starting {} test --", methodName);
PulsarClient pulsarClient = newPulsarClient(lookupUrl.toString(), 0);// Creates new client connection
@@ -748,7 +748,7 @@ public void testMessageIdForSubscribeToSinglePartition() throws Exception {
*
* @throws Exception
*/
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testAutoUpdatePartitionsForProducerConsumer() throws Exception {
log.info("-- Starting {} test --", methodName);
PulsarClient pulsarClient = newPulsarClient(lookupUrl.toString(), 0);// Creates new client connection
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 62b4aec3a4a09..9a77b06a7c4e8 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
@@ -26,11 +26,7 @@
import java.util.List;
import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.pulsar.broker.stats.NamespaceStats;
@@ -64,6 +60,15 @@ protected void cleanup() throws Exception {
super.internalCleanup();
}
+ // Perhaps it would be better to import this method from PulsarTestBase in tests.integration
+ public static String randomName(int numChars) {
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < numChars; i++) {
+ sb.append((char) (ThreadLocalRandom.current().nextInt(26) + 'a'));
+ }
+ return sb.toString();
+ }
+
@DataProvider(name = "batch")
public Object[][] batchMessageDelayMsProvider() {
return new Object[][] { { 0 }, { 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 612b610754361..e3e6d5bb9e80c 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
@@ -536,7 +536,7 @@ public void testSendTimeout(int batchMessageDelayMs) throws Exception {
startBroker();
// We should not have received any message
- Message msg = consumer.receive(3, TimeUnit.SECONDS);
+ Message msg = consumer.receive(10, TimeUnit.MILLISECONDS);
Assert.assertNull(msg);
consumer.close();
log.info("-- Exiting {} test --", methodName);
@@ -750,7 +750,7 @@ public void testConcurrentConsumerReceiveWhileReconnect(int batchMessageDelayMs)
// clear the queue
while (true) {
- Message msg = consumer.receive(1, TimeUnit.SECONDS);
+ Message msg = consumer.receive(7, TimeUnit.SECONDS);
if (msg == null) {
break;
}
@@ -946,7 +946,7 @@ public void testActiveAndInActiveConsumerEntryCacheBehavior() throws Exception {
// sleep for a second: as ledger.updateCursorRateLimit RateLimiter will allow to invoke cursor-update after a
// second
- Thread.sleep(1000);//
+ Thread.sleep(2000); // Needs to be replaced with await() approach.
// produce-consume one more message to trigger : ledger.internalReadFromLedger(..) which updates cursor and
// EntryCache
producer.send("message".getBytes());
@@ -970,7 +970,7 @@ public void testActiveAndInActiveConsumerEntryCacheBehavior() throws Exception {
// sleep for a second: as ledger.updateCursorRateLimit RateLimiter will allow to invoke cursor-update after a
// second
- Thread.sleep(1000);//
+ Thread.sleep(2000);// Needs to be replaced with await() approach.
// produce-consume one more message to trigger : ledger.internalReadFromLedger(..) which updates cursor and
// EntryCache
producer.send("message".getBytes());
@@ -1176,7 +1176,7 @@ public void testSendCallBack() throws Exception {
*
* @throws Exception
*/
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testSharedConsumerAckDifferentConsumer() throws Exception {
log.info("-- Starting {} test --", methodName);
@@ -1301,7 +1301,7 @@ public void testConsumerBlockingWithUnAckedMessages() throws Exception {
Message msg = null;
List> messages = Lists.newArrayList();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(2, TimeUnit.SECONDS);
if (msg != null) {
messages.add(msg);
log.info("Received message: " + new String(msg.getData()));
@@ -1324,7 +1324,7 @@ public void testConsumerBlockingWithUnAckedMessages() throws Exception {
// try to consume remaining messages
int remainingMessages = totalProducedMsgs - messages.size();
for (int i = 0; i < remainingMessages; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(2, TimeUnit.SECONDS);
if (msg != null) {
messages.add(msg);
log.info("Received message: " + new String(msg.getData()));
@@ -1386,7 +1386,7 @@ public void testConsumerBlockingWithUnAckedMessagesMultipleIteration() throws Ex
Message msg = null;
List> messages = Lists.newArrayList();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(2, TimeUnit.SECONDS);
if (msg != null) {
messages.add(msg);
log.info("Received message: " + new String(msg.getData()));
@@ -1461,7 +1461,7 @@ public void testMutlipleSharedConsumerBlockingWithUnAckedMessages() throws Excep
Message msg = null;
List> messages = Lists.newArrayList();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer1.receive(1, TimeUnit.SECONDS);
+ msg = consumer1.receive(2, TimeUnit.SECONDS);
if (msg != null) {
messages.add(msg);
totalReceiveMessages++;
@@ -1476,7 +1476,7 @@ public void testMutlipleSharedConsumerBlockingWithUnAckedMessages() throws Excep
// (3.1) Consumer2 will start consuming messages without ack: it should stop after maxUnackedMessages
messages.clear();
for (int i = 0; i < totalProducedMsgs - maxUnackedMessages; i++) {
- msg = consumer2.receive(1, TimeUnit.SECONDS);
+ msg = consumer2.receive(2, TimeUnit.SECONDS);
if (msg != null) {
messages.add(msg);
totalReceiveMessages++;
@@ -1498,7 +1498,7 @@ public void testMutlipleSharedConsumerBlockingWithUnAckedMessages() throws Excep
// (4) Consumer2 consumer and ack: so it should consume all remaining messages
messages.clear();
for (int i = 0; i < totalProducedMsgs - (2 * maxUnackedMessages); i++) {
- msg = consumer2.receive(1, TimeUnit.SECONDS);
+ msg = consumer2.receive(2, TimeUnit.SECONDS);
if (msg != null) {
messages.add(msg);
totalReceiveMessages++;
@@ -1559,7 +1559,7 @@ public void testShouldNotBlockConsumerIfRedeliverBeforeReceive() throws Exceptio
assertEquals(consumer.numMessagesInQueue(), receiverQueueSize);
for (int i = 0; i < totalProducedMsgs; i++) {
- Message msg = consumer.receive(1, TimeUnit.SECONDS);
+ Message msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
consumer.acknowledge(msg);
totalReceiveMsg++;
@@ -1610,7 +1610,7 @@ public void testUnackBlockRedeliverMessages() throws Exception {
Message msg = null;
List> messages = Lists.newArrayList();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(2, TimeUnit.SECONDS);
if (msg != null) {
messages.add(msg);
totalReceiveMsg++;
@@ -1626,7 +1626,7 @@ public void testUnackBlockRedeliverMessages() throws Exception {
int alreadyConsumedMessages = messages.size();
messages.clear();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
consumer.acknowledge(msg);
totalReceiveMsg++;
@@ -1691,7 +1691,7 @@ public void testUnackedBlockAtBatch(int batchMessageDelayMs) throws Exception {
Message msg = null;
List> messages = Lists.newArrayList();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer1.receive(1, TimeUnit.SECONDS);
+ msg = consumer1.receive(2, TimeUnit.SECONDS);
if (msg != null) {
messages.add(msg);
totalReceiveMessages++;
@@ -1714,7 +1714,7 @@ public void testUnackedBlockAtBatch(int batchMessageDelayMs) throws Exception {
// (3) Consumer consumes and ack: so it should consume all remaining messages
messages.clear();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer1.receive(1, TimeUnit.SECONDS);
+ msg = consumer1.receive(2, TimeUnit.SECONDS);
if (msg != null) {
messages.add(msg);
totalReceiveMessages++;
@@ -1774,7 +1774,7 @@ public void testBlockUnackConsumerAckByDifferentConsumer() throws Exception {
Message msg = null;
List> messages = Lists.newArrayList();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer1.receive(1, TimeUnit.SECONDS);
+ msg = consumer1.receive(2, TimeUnit.SECONDS);
if (msg != null) {
messages.add(msg);
totalReceiveMessages++;
@@ -1797,7 +1797,7 @@ public void testBlockUnackConsumerAckByDifferentConsumer() throws Exception {
// (4) consumer1 will consumer remaining msgs and consumer2 will ack those messages
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer1.receive(1, TimeUnit.SECONDS);
+ msg = consumer1.receive(2, TimeUnit.SECONDS);
if (msg != null) {
totalReceiveMessages++;
consumer2.acknowledge(msg);
@@ -1808,7 +1808,7 @@ public void testBlockUnackConsumerAckByDifferentConsumer() throws Exception {
}
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer2.receive(1, TimeUnit.SECONDS);
+ msg = consumer2.receive(2, TimeUnit.SECONDS);
if (msg != null) {
totalReceiveMessages++;
log.info("Received message: " + new String(msg.getData()));
@@ -1904,7 +1904,7 @@ public void testBlockUnackedConsumerRedeliverySpecificMessagesProduceWithPause()
Message msg = null;
List> messages1 = Lists.newArrayList();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(2, TimeUnit.SECONDS);
if (msg != null) {
messages1.add(msg);
log.info("Received message: " + new String(msg.getData()));
@@ -1925,7 +1925,7 @@ public void testBlockUnackedConsumerRedeliverySpecificMessagesProduceWithPause()
Set messages2 = Sets.newHashSet();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(2, TimeUnit.SECONDS);
if (msg != null) {
messages2.add((MessageIdImpl) msg.getMessageId());
log.info("Received message: " + new String(msg.getData()));
@@ -1993,7 +1993,7 @@ public void testBlockUnackedConsumerRedeliverySpecificMessagesCloseConsumerWhile
Message msg = null;
List> messages1 = Lists.newArrayList();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(2, TimeUnit.SECONDS);
if (msg != null) {
messages1.add(msg);
log.info("Received message: " + new String(msg.getData()));
@@ -2014,7 +2014,7 @@ public void testBlockUnackedConsumerRedeliverySpecificMessagesCloseConsumerWhile
Set messages2 = Sets.newHashSet();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(2, TimeUnit.SECONDS);
if (msg != null) {
messages2.add((MessageIdImpl) msg.getMessageId());
log.info("Received message: " + new String(msg.getData()));
@@ -2271,7 +2271,7 @@ public void testRedeliveryFailOverConsumer() throws Exception {
Message msg = null;
List> messages1 = Lists.newArrayList();
for (int i = 0; i < consumeMsgInParts; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages1.add(msg);
consumer.acknowledge(msg);
@@ -2286,7 +2286,7 @@ public void testRedeliveryFailOverConsumer() throws Exception {
// (1.b) consume second consumeMsgInParts msgs and trigger redeliver
messages1.clear();
for (int i = 0; i < consumeMsgInParts; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages1.add(msg);
consumer.acknowledge(msg);
@@ -2308,7 +2308,7 @@ public void testRedeliveryFailOverConsumer() throws Exception {
int remainingMsgs = (2 * receiverQueueSize) - (2 * consumeMsgInParts);
messages1.clear();
for (int i = 0; i < remainingMsgs; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages1.add(msg);
consumer.acknowledge(msg);
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 7ed88f5c516f2..b3c6ad4b520c8 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
@@ -21,6 +21,7 @@
import com.google.common.collect.Sets;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.pulsar.broker.service.Dispatcher;
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 82aa7d35aa0e4..e08ff688efed3 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
@@ -35,7 +35,7 @@ public class TlsProducerConsumerTest extends TlsProducerConsumerBase {
*
* @throws Exception
*/
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testTlsLargeSizeMessage() throws Exception {
log.info("-- Starting {} test --", methodName);
@@ -69,7 +69,7 @@ public void testTlsLargeSizeMessage() throws Exception {
log.info("-- Exiting {} test --", methodName);
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testTlsClientAuthOverBinaryProtocol() throws Exception {
log.info("-- Starting {} test --", methodName);
@@ -97,7 +97,7 @@ public void testTlsClientAuthOverBinaryProtocol() throws Exception {
}
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testTlsClientAuthOverHTTPProtocol() throws Exception {
log.info("-- Starting {} test --", methodName);
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 a54dcd853c701..742bb8deed3d9 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
@@ -329,7 +329,7 @@ public void testSendTimeout(int batchMessageDelayMs) throws Exception {
startBroker();
// We should not have received any message
- Message msg = consumer.receive(3, TimeUnit.SECONDS);
+ Message msg = consumer.receive(10, TimeUnit.MILLISECONDS);
Assert.assertNull(msg);
consumer.close();
log.info("-- Exiting {} test --", methodName);
@@ -520,7 +520,7 @@ public Void call() throws Exception {
}
barrier.await();
- // there will be 10 threads calling receive() from the same consumer and will block
+ // there will be 10 threads calling receive(5) from the same consumer and will block
Thread.sleep(100);
// we restart the broker to reconnect
@@ -565,7 +565,7 @@ public Void call() throws Exception {
// clear the queue
while (true) {
- Message msg = consumer.receive(1, TimeUnit.SECONDS);
+ Message msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg == null) {
break;
}
@@ -853,7 +853,7 @@ public void testSendCallBack() throws Exception {
*
* @throws Exception
*/
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testSharedConsumerAckDifferentConsumer() throws Exception {
log.info("-- Starting {} test --", methodName);
@@ -985,7 +985,7 @@ public void testConsumerBlockingWithUnAckedMessages() throws Exception {
Messagemsg = null;
List> messages = Lists.newArrayList();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages.add(msg);
log.info("Received message: " + new String(msg.getData()));
@@ -1008,7 +1008,7 @@ public void testConsumerBlockingWithUnAckedMessages() throws Exception {
// try to consume remaining messages
int remainingMessages = totalProducedMsgs - messages.size();
for (int i = 0; i < remainingMessages; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages.add(msg);
log.info("Received message: " + new String(msg.getData()));
@@ -1073,7 +1073,7 @@ public void testConsumerBlockingWithUnAckedMessagesMultipleIteration() throws Ex
Messagemsg = null;
List> messages = Lists.newArrayList();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages.add(msg);
log.info("Received message: " + new String(msg.getData()));
@@ -1154,7 +1154,7 @@ public void testMutlipleSharedConsumerBlockingWithUnAckedMessages() throws Excep
Messagemsg = null;
List> messages = Lists.newArrayList();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer1.receive(1, TimeUnit.SECONDS);
+ msg = consumer1.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages.add(msg);
totalReceiveMessages++;
@@ -1169,7 +1169,7 @@ public void testMutlipleSharedConsumerBlockingWithUnAckedMessages() throws Excep
// (3.1) Consumer2 will start consuming messages without ack: it should stop after maxUnackedMessages
messages.clear();
for (int i = 0; i < totalProducedMsgs - maxUnackedMessages; i++) {
- msg = consumer2.receive(1, TimeUnit.SECONDS);
+ msg = consumer2.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages.add(msg);
totalReceiveMessages++;
@@ -1191,7 +1191,7 @@ public void testMutlipleSharedConsumerBlockingWithUnAckedMessages() throws Excep
// (4) Consumer2 consumer and ack: so it should consume all remaining messages
messages.clear();
for (int i = 0; i < totalProducedMsgs - (2 * maxUnackedMessages); i++) {
- msg = consumer2.receive(1, TimeUnit.SECONDS);
+ msg = consumer2.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages.add(msg);
totalReceiveMessages++;
@@ -1254,7 +1254,7 @@ public void testShouldNotBlockConsumerIfRedeliverBeforeReceive() throws Exceptio
assertEquals(consumer.numMessagesInQueue(), receiverQueueSize);
for (int i = 0; i < totalProducedMsgs; i++) {
- Messagemsg = consumer.receive(1, TimeUnit.SECONDS);
+ Messagemsg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
consumer.acknowledge(msg);
totalReceiveMsg++;
@@ -1309,7 +1309,7 @@ public void testUnackBlockRedeliverMessages() throws Exception {
Messagemsg = null;
List> messages = Lists.newArrayList();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages.add(msg);
totalReceiveMsg++;
@@ -1325,7 +1325,7 @@ public void testUnackBlockRedeliverMessages() throws Exception {
int alreadyConsumedMessages = messages.size();
messages.clear();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
consumer.acknowledge(msg);
totalReceiveMsg++;
@@ -1387,7 +1387,7 @@ public void testUnackedBlockAtBatch(int batchMessageDelayMs) throws Exception {
Messagemsg = null;
List> messages = Lists.newArrayList();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer1.receive(1, TimeUnit.SECONDS);
+ msg = consumer1.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages.add(msg);
totalReceiveMessages++;
@@ -1410,7 +1410,7 @@ public void testUnackedBlockAtBatch(int batchMessageDelayMs) throws Exception {
// (3) Consumer consumes and ack: so it should consume all remaining messages
messages.clear();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer1.receive(1, TimeUnit.SECONDS);
+ msg = consumer1.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages.add(msg);
totalReceiveMessages++;
@@ -1473,7 +1473,7 @@ public void testBlockUnackConsumerAckByDifferentConsumer() throws Exception {
Messagemsg = null;
List> messages = Lists.newArrayList();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer1.receive(1, TimeUnit.SECONDS);
+ msg = consumer1.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages.add(msg);
totalReceiveMessages++;
@@ -1496,7 +1496,7 @@ public void testBlockUnackConsumerAckByDifferentConsumer() throws Exception {
// (4) consumer1 will consumer remaining msgs and consumer2 will ack those messages
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer1.receive(1, TimeUnit.SECONDS);
+ msg = consumer1.receive(5, TimeUnit.SECONDS);
if (msg != null) {
totalReceiveMessages++;
consumer2.acknowledge(msg);
@@ -1507,7 +1507,7 @@ public void testBlockUnackConsumerAckByDifferentConsumer() throws Exception {
}
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer2.receive(1, TimeUnit.SECONDS);
+ msg = consumer2.receive(5, TimeUnit.SECONDS);
if (msg != null) {
totalReceiveMessages++;
log.info("Received message: " + new String(msg.getData()));
@@ -1609,7 +1609,7 @@ public void testBlockUnackedConsumerRedeliverySpecificMessagesProduceWithPause()
Messagemsg = null;
List> messages1 = Lists.newArrayList();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages1.add(msg);
log.info("Received message: " + new String(msg.getData()));
@@ -1630,7 +1630,7 @@ public void testBlockUnackedConsumerRedeliverySpecificMessagesProduceWithPause()
Set messages2 = Sets.newHashSet();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages2.add((MessageIdImpl) msg.getMessageId());
log.info("Received message: " + new String(msg.getData()));
@@ -1705,7 +1705,7 @@ public void testBlockUnackedConsumerRedeliverySpecificMessagesCloseConsumerWhile
Messagemsg = null;
List> messages1 = Lists.newArrayList();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages1.add(msg);
log.info("Received message: " + new String(msg.getData()));
@@ -1726,7 +1726,7 @@ public void testBlockUnackedConsumerRedeliverySpecificMessagesCloseConsumerWhile
Set messages2 = Sets.newHashSet();
for (int i = 0; i < totalProducedMsgs; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages2.add((MessageIdImpl) msg.getMessageId());
log.info("Received message: " + new String(msg.getData()));
@@ -1985,7 +1985,7 @@ public void testRedeliveryFailOverConsumer() throws Exception {
Message msg = null;
List> messages1 = Lists.newArrayList();
for (int i = 0; i < consumeMsgInParts; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages1.add(msg);
consumer.acknowledge(msg);
@@ -2000,7 +2000,7 @@ public void testRedeliveryFailOverConsumer() throws Exception {
// (1.b) consume second consumeMsgInParts msgs and trigger redeliver
messages1.clear();
for (int i = 0; i < consumeMsgInParts; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages1.add(msg);
consumer.acknowledge(msg);
@@ -2022,7 +2022,7 @@ public void testRedeliveryFailOverConsumer() throws Exception {
int remainingMsgs = (2 * receiverQueueSize) - (2 * consumeMsgInParts);
messages1.clear();
for (int i = 0; i < remainingMsgs; i++) {
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
if (msg != null) {
messages1.add(msg);
consumer.acknowledge(msg);
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 15357d318c8d6..16c85be11fdb6 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
@@ -352,7 +352,7 @@ public void testUnsupportedBatchMessageConsumer(SubscriptionType subType) throws
Set messageSet = Sets.newHashSet();
Message msg = null;
for (int i = 0; i < numMessagesPerBatch; i++) {
- msg = consumer1.receive(1, TimeUnit.SECONDS);
+ msg = consumer1.receive(5, TimeUnit.SECONDS);
String receivedMessage = new String(msg.getData());
String expectedMessage = "my-message-" + i;
testMessageOrderAndDuplicates(messageSet, receivedMessage, expectedMessage);
@@ -370,7 +370,7 @@ public void testUnsupportedBatchMessageConsumer(SubscriptionType subType) throws
batchProducer.flush();
// consumer should have not received any message as it should have been disconnected
- msg = consumer1.receive(100, TimeUnit.MILLISECONDS);
+ msg = consumer1.receive(10, TimeUnit.MILLISECONDS);
assertNull(msg);
// subscribe consumer2 with supporting batch version
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 f7020378d5655..2bd28d7fbbf70 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,6 +33,8 @@
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+import java.util.concurrent.TimeUnit;
+
public class CompactedOutBatchMessageTest extends ProducerConsumerBase {
@BeforeMethod
@Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerDedupPermitsUpdate.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerDedupPermitsUpdate.java
index 15070484a0fb8..6e1434d5b2a4f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerDedupPermitsUpdate.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerDedupPermitsUpdate.java
@@ -92,7 +92,7 @@ public void testConsumerDedup(boolean batchingEnabled, int receiverQueueSize) th
// Consumer receives and acks all the messages, though the acks
// are still cached in client lib
for (int i = 0; i < 30; i++) {
- Message msg = consumer.receive();
+ Message msg = consumer.receive(); // Can't use receive with timeout, if the queue size is 0
assertEquals(msg.getValue(), "hello-" + i);
consumer.acknowledge(msg);
}
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 14349f7c4f8f8..12930356e5d4e 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
@@ -343,9 +343,9 @@ public void testChecksumVersionComptability() throws Exception {
((ConsumerImpl) consumer).grabCnx();
// We should only receive msg1
- Message msg = consumer.receive(1, TimeUnit.SECONDS);
+ Message msg = consumer.receive(5, TimeUnit.SECONDS);
assertEquals(new String(msg.getData()), "message-1");
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
assertEquals(new String(msg.getData()), "message-3");
}
@@ -415,9 +415,9 @@ public void testChecksumReconnection() throws Exception {
((ConsumerImpl) consumer).grabCnx();
// We should only receive msg1
- Message msg = consumer.receive(1, TimeUnit.SECONDS);
+ Message msg = consumer.receive(5, TimeUnit.SECONDS);
assertEquals(new String(msg.getData()), "message-1");
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(5, TimeUnit.SECONDS);
assertEquals(new String(msg.getData()), "message-3");
}
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 b14c1cd466090..a1d8dd23ac837 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
@@ -809,7 +809,7 @@ public void testTopicDeletion() throws Exception {
producer2.send("msg-2");
- message = consumer.receive();
+ message = consumer.receive(5, TimeUnit.SECONDS);
assertEquals(message.getValue(), "msg-2");
consumer.acknowledge(message);
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 d435938c33f72..d3026a460a45c 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
@@ -55,7 +55,7 @@ public void cleanup() throws Exception {
super.internalCleanup();
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testProducerSemaphoreAcquireAndRelease() throws PulsarClientException, ExecutionException, InterruptedException {
final int pendingQueueSize = 100;
@@ -122,9 +122,9 @@ public void testProducerSemaphoreAcquireAndRelease() throws PulsarClientExceptio
/**
* We use semaphore to limit the pending send, so we must ensure that the thread of sending message never block
* at the pending message queue. If not, the dead lock might occur. Here is the related issue to describe the
- * dead lock happens {https://github.com/apache/pulsar/issues/5585}
+ * dead lock {https://github.com/apache/pulsar/issues/5585}
*/
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testEnsureNotBlockOnThePendingQueue() throws Exception {
final int pendingQueueSize = 10;
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 b75cfcee0b9d2..e24933597e9ec 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
@@ -139,9 +139,11 @@ public void testReadMessageWithBatchingWithMessageInclusive() throws Exception {
String topic = "persistent://my-property/my-ns/my-reader-topic-with-batching-inclusive";
Set keys = publishMessages(topic, 10, true);
- Reader reader = pulsarClient.newReader().topic(topic).startMessageId(MessageId.latest)
+ Reader reader = pulsarClient.newReader().topic(topic).startMessageId(MessageId.earliest)
.startMessageIdInclusive().readerName(subscription).create();
+ // If we use MessageId.latest, then publishMessages must occur after we create the reader.
+ Assert.assertTrue(reader.hasMessageAvailable());
while (reader.hasMessageAvailable()) {
Assert.assertTrue(keys.remove(reader.readNext().getKey()));
}
@@ -199,7 +201,7 @@ public void testReadFromPartition() throws Exception {
* 1. publish messages which are 5 hour old
* 2. publish messages which are 1 hour old
* 3. Create reader with rollback time 2 hours
- * 4. Reader should be able to read only messages which are only 2 hours old
+ * 4. Reader should be able to read only messages which are only 2 hours old.
*
* @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 b604505ac54d0..df2027ac311a7 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
@@ -21,6 +21,7 @@
import static org.testng.Assert.assertEquals;
import java.util.Collections;
+import java.util.concurrent.TimeUnit;
import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
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 c6883c63e0670..6ade1c59b8cb7 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
@@ -638,7 +638,7 @@ public void testTopicsNameSubscribeWithBuilderFail() throws Exception {
/**
* Test Listener for github issue #2547
*/
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testMultiTopicsMessageListener() throws Exception {
String key = "MultiTopicsMessageListenerTest";
final String subscriptionName = "my-ex-subscription-" + key;
@@ -704,7 +704,7 @@ public void testMultiTopicsMessageListener() throws Exception {
* 4. produce message to xx-partition-2 again, and verify consumer could receive message.
*
*/
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void testTopicAutoUpdatePartitions() throws Exception {
String key = "TestTopicAutoUpdatePartitions";
final String subscriptionName = "my-ex-subscription-" + key;
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 122ce3ea158c8..299bb0cba365a 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
@@ -45,7 +45,7 @@
public class UnAcknowledgedMessagesTimeoutTest extends BrokerTestBase {
private static final Logger log = LoggerFactory.getLogger(UnAcknowledgedMessagesTimeoutTest.class);
- private final long ackTimeOutMillis = TimeUnit.SECONDS.toMillis(2);
+ private final long ackTimeOutMillis = TimeUnit.SECONDS.toMillis(12);
@Override
@BeforeMethod
@@ -209,7 +209,7 @@ public void testSharedSingleAckedPartitionedTopic() throws Exception {
// 5. Check if Messages redelivered again
// Since receive is a blocking call hoping that timeout will kick in
- Thread.sleep((int) (ackTimeOutMillis * 1.1));
+ Thread.sleep((int) (ackTimeOutMillis * 1.1)); // Timeout triggers redelivery of unacked messages.
log.info(key + " Timeout should be triggered now");
messageCount1 = receiveAllMessage(consumer1, true);
messageCount2 += receiveAllMessage(consumer2, false);
@@ -236,7 +236,7 @@ public void testSharedSingleAckedPartitionedTopic() throws Exception {
private static int receiveAllMessage(Consumer> consumer, boolean ackMessages) throws Exception {
int messagesReceived = 0;
- Message> msg = consumer.receive(1, TimeUnit.SECONDS);
+ Message> msg = consumer.receive(2, TimeUnit.SECONDS);
while (msg != null) {
++messagesReceived;
log.info("Consumer received {}", new String(msg.getData()));
@@ -245,7 +245,7 @@ private static int receiveAllMessage(Consumer> consumer, boolean ackMessages)
consumer.acknowledge(msg);
}
- msg = consumer.receive(1, TimeUnit.SECONDS);
+ msg = consumer.receive(2, TimeUnit.SECONDS);
}
return messagesReceived;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java
index c5b605a766a23..ee17012fff279 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java
@@ -489,13 +489,13 @@ private void testE2EPulsarFunction(String jarFilePathUrl) throws Exception {
Assert.assertEquals(foundFiles.length, 0, "Temporary files left over: " + Arrays.asList(foundFiles));
}
- @Test(timeOut = 20000)
+ @Test(timeOut = 40000)
public void testE2EPulsarFunctionWithFile() throws Exception {
String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-functions-api-examples.jar").getFile();
testE2EPulsarFunction(jarFilePathUrl);
}
- @Test(timeOut = 40000)
+ @Test(timeOut = 80000)
public void testE2EPulsarFunctionWithUrl() throws Exception {
String jarFilePathUrl = String.format("http://127.0.0.1:%d/pulsar-functions-api-examples.jar",
fileServer.getAddress().getPort());
@@ -720,13 +720,13 @@ private void testPulsarSinkStats(String jarFilePathUrl) throws Exception {
Assert.assertEquals(foundFiles.length, 0, "Temporary files left over: " + Arrays.asList(foundFiles));
}
- @Test(timeOut = 20000)
+ @Test(timeOut = 40000)
public void testPulsarSinkStatsWithFile() throws Exception {
String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-io-data-generator.nar").getFile();
testPulsarSinkStats(jarFilePathUrl);
}
- @Test(timeOut = 40000)
+ @Test(timeOut = 80000)
public void testPulsarSinkStatsWithUrl() throws Exception {
String jarFilePathUrl = String.format("http://127.0.0.1:%d/pulsar-io-data-generator.nar",
fileServer.getAddress().getPort());
@@ -860,20 +860,20 @@ private void testPulsarSourceStats(String jarFilePathUrl) throws Exception {
Assert.assertEquals(foundFiles.length, 0, "Temporary files left over: " + Arrays.asList(foundFiles));
}
- @Test(timeOut = 20000)
+ @Test(timeOut = 40000)
public void testPulsarSourceStatsWithFile() throws Exception {
String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-io-data-generator.nar").getFile();
testPulsarSourceStats(jarFilePathUrl);
}
- @Test(timeOut = 40000)
+ @Test(timeOut = 80000)
public void testPulsarSourceStatsWithUrl() throws Exception {
String jarFilePathUrl = String.format("http://127.0.0.1:%d/pulsar-io-data-generator.nar",
fileServer.getAddress().getPort());
testPulsarSourceStats(jarFilePathUrl);
}
- @Test(timeOut = 20000)
+ @Test(timeOut = 40000)
public void testPulsarFunctionStats() throws Exception {
final String namespacePortion = "io";
@@ -1209,7 +1209,7 @@ public void testPulsarFunctionStats() throws Exception {
Assert.assertEquals(foundFiles.length, 0, "Temporary files left over: " + Arrays.asList(foundFiles));
}
- @Test(timeOut = 20000)
+ @Test(timeOut = 40000)
public void testPulsarFunctionStatus() throws Exception {
final String namespacePortion = "io";
@@ -1319,7 +1319,7 @@ public void testAuthorization(boolean validRoleName) throws Exception {
}
}
- @Test(timeOut = 20000)
+ @Test(timeOut = 40000)
public void testFunctionStopAndRestartApi() throws Exception {
final String namespacePortion = "io";
@@ -1386,7 +1386,7 @@ public void testFunctionStopAndRestartApi() throws Exception {
producer.close();
}
- @Test(timeOut = 20000)
+ @Test(timeOut = 40000)
public void testFunctionAutomaticSubCleanup() throws Exception {
final String namespacePortion = "io";
final String replNamespace = tenant + "/" + namespacePortion;
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 d4e90329c0133..5481895fb210f 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
@@ -86,7 +86,7 @@ protected void cleanup() throws Exception {
}
- @Test(timeOut = 30000)
+ @Test(timeOut = 90000)
public void socketTest() throws InterruptedException, GeneralSecurityException {
String consumerUri =
"wss://localhost:" + proxyServer.getListenPortHTTPS().get() + "/ws/consumer/persistent/my-property/use/my-ns/my-topic/my-sub";
diff --git a/pulsar-client-cpp/tests/BasicEndToEndTest.cc b/pulsar-client-cpp/tests/BasicEndToEndTest.cc
index 0010c963fdcb3..f73f2dd4d53c6 100644
--- a/pulsar-client-cpp/tests/BasicEndToEndTest.cc
+++ b/pulsar-client-cpp/tests/BasicEndToEndTest.cc
@@ -748,7 +748,7 @@ TEST(BasicEndToEndTest, testSinglePartitionRoutingPolicy) {
for (int i = 0; i < 10; i++) {
Message m;
- consumer.receive(m);
+ consumer.receive(m, 9000);
consumer.acknowledgeCumulative(m);
}
@@ -1990,14 +1990,14 @@ TEST(BasicEndToEndTest, testPatternMultiTopicsConsumerPubSub) {
LOG_INFO("Consuming and acking 300 messages by multiTopicsConsumer");
for (int i = 0; i < 3 * messageNumber; i++) {
Message m;
- ASSERT_EQ(ResultOk, consumer.receive(m, 1000));
+ ASSERT_EQ(ResultOk, consumer.receive(m, 5000));
ASSERT_EQ(ResultOk, consumer.acknowledge(m));
}
LOG_INFO("Consumed and acked 300 messages by multiTopicsConsumer");
- // verify no more to receive, because producer4 not match pattern
+ // verify no more to receive because producer4 not match pattern
Message m;
- ASSERT_EQ(ResultTimeout, consumer.receive(m, 1000));
+ ASSERT_EQ(ResultTimeout, consumer.receive(m, 5000));
ASSERT_EQ(ResultOk, consumer.unsubscribe());
@@ -2085,14 +2085,14 @@ TEST(BasicEndToEndTest, testpatternMultiTopicsHttpConsumerPubSub) {
LOG_INFO("Consuming and acking 300 messages by multiTopicsConsumer");
for (int i = 0; i < 3 * messageNumber; i++) {
Message m;
- ASSERT_EQ(ResultOk, consumer.receive(m, 1000));
+ ASSERT_EQ(ResultOk, consumer.receive(m, 5000));
ASSERT_EQ(ResultOk, consumer.acknowledge(m));
}
LOG_INFO("Consumed and acked 300 messages by multiTopicsConsumer");
// verify no more to receive
Message m;
- ASSERT_EQ(ResultTimeout, consumer.receive(m, 1000));
+ ASSERT_EQ(ResultTimeout, consumer.receive(m, 5000));
ASSERT_EQ(ResultOk, consumer.unsubscribe());
@@ -2230,14 +2230,14 @@ TEST(BasicEndToEndTest, testPatternMultiTopicsConsumerAutoDiscovery) {
LOG_INFO("Consuming and acking 300 messages by pattern topics consumer");
for (int i = 0; i < 3 * messageNumber; i++) {
Message m;
- ASSERT_EQ(ResultOk, consumer.receive(m, 1000));
+ ASSERT_EQ(ResultOk, consumer.receive(m, 5000));
ASSERT_EQ(ResultOk, consumer.acknowledge(m));
}
LOG_INFO("Consumed and acked 300 messages by pattern topics consumer");
// verify no more to receive, because producer4 not match pattern
Message m;
- ASSERT_EQ(ResultTimeout, consumer.receive(m, 1000));
+ ASSERT_EQ(ResultTimeout, consumer.receive(m, 5000));
ASSERT_EQ(ResultOk, consumer.unsubscribe());
diff --git a/pulsar-client-kafka-compat/pulsar-client-kafka_0_8/pom.xml b/pulsar-client-kafka-compat/pulsar-client-kafka_0_8/pom.xml
index 8d35f2d265d33..18f4fe0984690 100644
--- a/pulsar-client-kafka-compat/pulsar-client-kafka_0_8/pom.xml
+++ b/pulsar-client-kafka-compat/pulsar-client-kafka_0_8/pom.xml
@@ -49,6 +49,10 @@
test