Skip to content

Increase the retries count for PulsarFunctionStateTest #5328

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merlimat merged 6 commits into from
Feb 12, 2020
Merged

Increase the retries count for PulsarFunctionStateTest #5328

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
ea22638
Increase the retries count for PulsarFunctionStateTest
merlimat Oct 7, 2019
e486f4d
Increased retries in similar tests backoffs
merlimat Oct 8, 2019
58ab44b
Merge remote-tracking branch 'apache/master' into fix-state-test
merlimat Oct 24, 2019
57c5b99
Merge branch 'master' into fix-state-test
merlimat Feb 10, 2020
b006ccc
Fixed merge conflict
merlimat Feb 11, 2020
7e14e9f
Merge remote-tracking branch 'apache/master' into fix-state-test
merlimat Feb 11, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -721,7 +721,7 @@ public void testConcurrentBatchMessageAck(BatcherBuilder builder) throws Excepti
PersistentDispatcherMultipleConsumers dispatcher = (PersistentDispatcherMultipleConsumers) topic
.getSubscription(subscriptionName).getDispatcher();
// check strategically to let ack-message receive by broker
retryStrategically((test) -> dispatcher.getConsumers().get(0).getUnackedMessages() == 0, 5, 150);
retryStrategically((test) -> dispatcher.getConsumers().get(0).getUnackedMessages() == 0, 50, 150);
assertEquals(dispatcher.getConsumers().get(0).getUnackedMessages(), 0);

executor.shutdown();
Expand Down
4 changes: 2 additions & 2 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorGlobalNSTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,7 @@ public void testRemoveLocalClusterOnGlobalNamespace() throws Exception {
admin1.namespaces().setNamespaceReplicationClusters(namespace, Sets.newHashSet("r2", "r3"));

MockedPulsarServiceBaseTest
.retryStrategically((test) -> !pulsar1.getBrokerService().getTopics().containsKey(topicName), 5, 150);
.retryStrategically((test) -> !pulsar1.getBrokerService().getTopics().containsKey(topicName), 50, 150);

Assert.assertFalse(pulsar1.getBrokerService().getTopics().containsKey(topicName));
Assert.assertFalse(producer1.isConnected());
Expand Down Expand Up @@ -118,7 +118,7 @@ public void testForcefullyTopicDeletion() throws Exception {
admin1.topics().delete(topicName, true);

MockedPulsarServiceBaseTest
.retryStrategically((test) -> !pulsar1.getBrokerService().getTopics().containsKey(topicName), 5, 150);
.retryStrategically((test) -> !pulsar1.getBrokerService().getTopics().containsKey(topicName), 50, 150);

Assert.assertFalse(pulsar1.getBrokerService().getTopics().containsKey(topicName));

Expand Down
14 changes: 7 additions & 7 deletions ...roker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -332,7 +332,7 @@ public void testAuthorizationWithAnonymousUser() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150));
}, 50, 150));
// validate pulsar sink consumer has started on the topic
assertEquals(admin1.topics().getStats(sourceTopic).subscriptions.size(), 1);

Expand All @@ -352,7 +352,7 @@ public void testAuthorizationWithAnonymousUser() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);

Message<String> msg = consumer.receive(5, TimeUnit.SECONDS);
String receivedPropertyValue = msg.getProperty(propertyKey);
Expand Down Expand Up @@ -384,7 +384,7 @@ public void testAuthorizationWithAnonymousUser() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150));
}, 50, 150));

// test getFunctionInfo
try {
Expand Down Expand Up @@ -603,7 +603,7 @@ public void testAuthorization() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150));
}, 50, 150));
// validate pulsar sink consumer has started on the topic
assertEquals(admin1.topics().getStats(sourceTopic).subscriptions.size(), 1);

Expand All @@ -623,7 +623,7 @@ public void testAuthorization() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);

Message<String> msg = consumer.receive(5, TimeUnit.SECONDS);
String receivedPropertyValue = msg.getProperty(propertyKey);
Expand Down Expand Up @@ -655,7 +655,7 @@ public void testAuthorization() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150));
}, 50, 150));

// test getFunctionInfo
try {
Expand Down Expand Up @@ -796,7 +796,7 @@ public void testAuthorization() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150));
}, 50, 150));
}
}
}
4 changes: 2 additions & 2 deletions ...r-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -412,7 +412,7 @@ private void testE2EPulsarFunctionLocalRun(String jarFilePathUrl) throws Excepti
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);
// validate pulsar sink consumer has started on the topic
TopicStats stats = admin.topics().getStats(sourceTopic);
assertTrue(stats.subscriptions.get(subscriptionName) != null
Expand All @@ -430,7 +430,7 @@ private void testE2EPulsarFunctionLocalRun(String jarFilePathUrl) throws Excepti
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);

for (int i = 0; i < totalMsgs; i++) {
Message<String> msg = consumer.receive(5, TimeUnit.SECONDS);
Expand Down
8 changes: 4 additions & 4 deletions ...ar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -304,7 +304,7 @@ public void testPulsarFunctionState() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);
// validate pulsar sink consumer has started on the topic
assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 1);

Expand All @@ -320,7 +320,7 @@ public void testPulsarFunctionState() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);

retryStrategically((test) -> {
try {
Expand All @@ -329,7 +329,7 @@ public void testPulsarFunctionState() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);

for (int i = 0; i < 5; i++) {
Message<String> msg = consumer.receive(5, TimeUnit.SECONDS);
Expand All @@ -353,7 +353,7 @@ public void testPulsarFunctionState() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);

// make sure subscriptions are cleanup
assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 0);
Expand Down
10 changes: 5 additions & 5 deletions ...r-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -193,7 +193,7 @@ public void testFunctionAssignments() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);
// validate 2 instances have been started
assertEquals(admin.topics().getStats(sinkTopic).subscriptions.size(), 1);
assertEquals(admin.topics().getStats(sinkTopic).subscriptions.values().iterator().next().consumers.size(), 2);
Expand All @@ -210,7 +210,7 @@ public void testFunctionAssignments() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);
// validate pulsar sink consumer has started on the topic
log.info("admin.topics().getStats(sinkTopic): {}", new Gson().toJson(admin.topics().getStats(sinkTopic)));
assertEquals(admin.topics().getStats(sinkTopic).subscriptions.values().iterator().next().consumers.size(), 1);
Expand Down Expand Up @@ -250,7 +250,7 @@ public void testFunctionAssignmentsWithRestart() throws Exception {
} catch (Exception e) {
return false;
}
}, 5, 150);
}, 50, 150);

// Validate registered assignments
Map<String, Assignment> assignments = runtimeManager.getCurrentAssignments().values().iterator().next();
Expand Down Expand Up @@ -279,7 +279,7 @@ public void testFunctionAssignmentsWithRestart() throws Exception {
} catch (Exception e) {
return false;
}
}, 5, 150);
}, 50, 150);

// Validate registered assignments
assignments = runtimeManager.getCurrentAssignments().values().iterator().next();
Expand All @@ -298,7 +298,7 @@ public void testFunctionAssignmentsWithRestart() throws Exception {
} catch (Exception e) {
return false;
}
}, 5, 150);
}, 50, 150);

// Validate registered assignments
assignments = runtimeManager2.getCurrentAssignments().values().iterator().next();
Expand Down
38 changes: 19 additions & 19 deletions pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -441,7 +441,7 @@ private void testE2EPulsarFunction(String jarFilePathUrl) throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);
// validate pulsar sink consumer has started on the topic
assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 1);

Expand All @@ -457,7 +457,7 @@ private void testE2EPulsarFunction(String jarFilePathUrl) throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);

Message<String> msg = consumer.receive(5, TimeUnit.SECONDS);
String receivedPropertyValue = msg.getProperty(propertyKey);
Expand All @@ -477,7 +477,7 @@ private void testE2EPulsarFunction(String jarFilePathUrl) throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);

// make sure subscriptions are cleanup
assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 0);
Expand Down Expand Up @@ -708,7 +708,7 @@ private void testPulsarSinkStats(String jarFilePathUrl) throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);

// make sure subscriptions are cleanup
assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 0);
Expand Down Expand Up @@ -905,7 +905,7 @@ public void testPulsarFunctionStats() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);
// validate pulsar sink consumer has started on the topic
assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 1);

Expand Down Expand Up @@ -1197,7 +1197,7 @@ public void testPulsarFunctionStats() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);

// make sure subscriptions are cleanup
assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 0);
Expand Down Expand Up @@ -1241,7 +1241,7 @@ public void testPulsarFunctionStatus() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);
// validate pulsar sink consumer has started on the topic
assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 1);

Expand Down Expand Up @@ -1284,7 +1284,7 @@ public void testPulsarFunctionStatus() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);

// make sure subscriptions are cleanup
assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 0);
Expand Down Expand Up @@ -1348,7 +1348,7 @@ public void testFunctionStopAndRestartApi() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);

SubscriptionStats subStats = admin.topics().getStats(sourceTopic).subscriptions.get(subscriptionName);
assertEquals(subStats.consumers.size(), 1);
Expand All @@ -1363,7 +1363,7 @@ public void testFunctionStopAndRestartApi() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);

subStats = admin.topics().getStats(sourceTopic).subscriptions.get(subscriptionName);
assertEquals(subStats.consumers.size(), 0);
Expand All @@ -1378,7 +1378,7 @@ public void testFunctionStopAndRestartApi() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);

subStats = admin.topics().getStats(sourceTopic).subscriptions.get(subscriptionName);
assertEquals(subStats.consumers.size(), 1);
Expand Down Expand Up @@ -1421,7 +1421,7 @@ public void testFunctionAutomaticSubCleanup() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);
assertFalse(admin.functions().getFunction(tenant, namespacePortion, functionName).getCleanupSubscription());

retryStrategically((test) -> {
Expand All @@ -1430,7 +1430,7 @@ public void testFunctionAutomaticSubCleanup() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);
// validate pulsar source consumer has started on the topic
assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 1);

Expand All @@ -1444,7 +1444,7 @@ public void testFunctionAutomaticSubCleanup() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);
assertTrue(admin.functions().getFunction(tenant, namespacePortion, functionName).getCleanupSubscription());

int totalMsgs = 10;
Expand Down Expand Up @@ -1487,7 +1487,7 @@ public void testFunctionAutomaticSubCleanup() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);

// make sure subscriptions are cleanup
assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 0);
Expand All @@ -1503,7 +1503,7 @@ public void testFunctionAutomaticSubCleanup() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);
// validate pulsar source consumer has started on the topic
assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 1);

Expand All @@ -1514,7 +1514,7 @@ public void testFunctionAutomaticSubCleanup() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);
assertFalse(admin.functions().getFunction(tenant, namespacePortion, functionName).getCleanupSubscription());

// test update another config and making sure that subscription cleanup remains unchanged
Expand All @@ -1528,7 +1528,7 @@ public void testFunctionAutomaticSubCleanup() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);
assertFalse(admin.functions().getFunction(tenant, namespacePortion, functionName).getCleanupSubscription());

// delete functions
Expand All @@ -1540,7 +1540,7 @@ public void testFunctionAutomaticSubCleanup() throws Exception {
} catch (PulsarAdminException e) {
return false;
}
}, 5, 150);
}, 50, 150);

// make sure subscriptions are cleanup
assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 1);
Expand Down