From ea22638ef9ee6a4f14c89eb60d0305e506860cc9 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Mon, 7 Oct 2019 14:43:59 -0700 Subject: [PATCH 1/3] Increase the retries count for PulsarFunctionStateTest --- .../pulsar/functions/worker/PulsarFunctionStateTest.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionStateTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionStateTest.java index d0b85c3a87fb8..41b550f886921 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionStateTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionStateTest.java @@ -311,7 +311,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); @@ -327,7 +327,7 @@ public void testPulsarFunctionState() throws Exception { } catch (PulsarAdminException e) { return false; } - }, 5, 150); + }, 50, 150); retryStrategically((test) -> { try { @@ -336,7 +336,7 @@ public void testPulsarFunctionState() throws Exception { } catch (PulsarAdminException e) { return false; } - }, 5, 150); + }, 50, 150); FunctionState state = admin.functions().getFunctionState(tenant, namespacePortion, functionName, "foo"); assertEquals(state.getNumberValue().intValue(), 5); @@ -375,7 +375,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); From e486f4d298821f1912c5858d118119d2776dbab2 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Tue, 8 Oct 2019 16:27:43 -0700 Subject: [PATCH 2/3] Increased retries in similar tests backoffs --- .../broker/service/BatchMessageTest.java | 2 +- .../service/ReplicatorGlobalNSTest.java | 4 +- .../worker/PulsarFunctionE2ESecurityTest.java | 18 ++++---- .../worker/PulsarFunctionLocalRunTest.java | 4 +- .../worker/PulsarFunctionPublishTest.java | 8 ++-- .../worker/PulsarWorkerAssignmentTest.java | 10 ++--- .../pulsar/io/PulsarFunctionE2ETest.java | 42 +++++++++---------- 7 files changed, 44 insertions(+), 44 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java index 9fe65cb06ed91..c0373bc53b4dc 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java @@ -665,7 +665,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(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorGlobalNSTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorGlobalNSTest.java index da48dd9474aa1..aa5a843afcd0c 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorGlobalNSTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorGlobalNSTest.java @@ -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()); @@ -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)); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java index 3782286fab7fe..fd93759e0c5cb 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java @@ -213,7 +213,7 @@ private WorkerService createPulsarFunctionWorker(ServiceConfiguration config) { System.setProperty(JAVA_INSTANCE_JAR_PROPERTY, FutureUtil.class.getProtectionDomain().getCodeSource().getLocation().getPath()); - + workerConfig = new WorkerConfig(); workerConfig.setPulsarFunctionsNamespace(pulsarFunctionsNamespace); workerConfig.setSchedulerClassName( @@ -323,7 +323,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.functions().getFunctionStatus(TENANT, NAMESPACE, functionName).getNumRunning(), 1); assertEquals(admin1.topics().getStats(sourceTopic).subscriptions.size(), 1); @@ -344,7 +344,7 @@ public void testAuthorizationWithAnonymousUser() throws Exception { } catch (PulsarAdminException e) { return false; } - }, 5, 150); + }, 50, 150); Message msg = consumer.receive(5, TimeUnit.SECONDS); String receivedPropertyValue = msg.getProperty(propertyKey); @@ -376,7 +376,7 @@ public void testAuthorizationWithAnonymousUser() throws Exception { } catch (PulsarAdminException e) { return false; } - }, 5, 150); + }, 50, 150); assertEquals(admin1.functions().getFunctionStatus(TENANT, NAMESPACE, functionName).getNumRunning(), 2); @@ -513,7 +513,7 @@ public void testAuthorizationWithAnonymousUser() throws Exception { } catch (PulsarAdminException e) { return false; } - }, 5, 150); + }, 50, 150); // make sure subscriptions are cleanup assertEquals(admin1.topics().getStats(sourceTopic).subscriptions.size(), 0); @@ -595,7 +595,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.functions().getFunctionStatus(TENANT, NAMESPACE, functionName).getNumRunning(), 1); assertEquals(admin1.topics().getStats(sourceTopic).subscriptions.size(), 1); @@ -616,7 +616,7 @@ public void testAuthorization() throws Exception { } catch (PulsarAdminException e) { return false; } - }, 5, 150); + }, 50, 150); Message msg = consumer.receive(5, TimeUnit.SECONDS); String receivedPropertyValue = msg.getProperty(propertyKey); @@ -648,7 +648,7 @@ public void testAuthorization() throws Exception { } catch (PulsarAdminException e) { return false; } - }, 5, 150); + }, 50, 150); assertEquals(admin1.functions().getFunctionStatus(TENANT, NAMESPACE, functionName).getNumRunning(), 2); @@ -786,7 +786,7 @@ public void testAuthorization() throws Exception { } catch (PulsarAdminException e) { return false; } - }, 5, 150); + }, 50, 150); // make sure subscriptions are cleanup assertEquals(admin1.topics().getStats(sourceTopic).subscriptions.size(), 0); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java index 8c2525e6c048a..103b7c5eac139 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java @@ -419,7 +419,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 @@ -437,7 +437,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 msg = consumer.receive(5, TimeUnit.SECONDS); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java index efa06957111dc..e4ee809f9b2b0 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java @@ -315,7 +315,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); @@ -331,7 +331,7 @@ public void testPulsarFunctionState() throws Exception { } catch (PulsarAdminException e) { return false; } - }, 5, 150); + }, 50, 150); retryStrategically((test) -> { try { @@ -340,7 +340,7 @@ public void testPulsarFunctionState() throws Exception { } catch (PulsarAdminException e) { return false; } - }, 5, 150); + }, 50, 150); for (int i = 0; i < 5; i++) { Message msg = consumer.receive(5, TimeUnit.SECONDS); @@ -364,7 +364,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); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java index 27a7d6981f07f..1dd9f41b849dd 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java @@ -195,7 +195,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); @@ -212,7 +212,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); @@ -252,7 +252,7 @@ public void testFunctionAssignmentsWithRestart() throws Exception { } catch (Exception e) { return false; } - }, 5, 150); + }, 50, 150); // Validate registered assignments Map assignments = runtimeManager.getCurrentAssignments().values().iterator().next(); @@ -281,7 +281,7 @@ public void testFunctionAssignmentsWithRestart() throws Exception { } catch (Exception e) { return false; } - }, 5, 150); + }, 50, 150); // Validate registered assignments assignments = runtimeManager.getCurrentAssignments().values().iterator().next(); @@ -300,7 +300,7 @@ public void testFunctionAssignmentsWithRestart() throws Exception { } catch (Exception e) { return false; } - }, 5, 150); + }, 50, 150); // Validate registered assignments assignments = runtimeManager2.getCurrentAssignments().values().iterator().next(); 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 d8f9cecd07634..fab5b81332a2e 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 @@ -312,7 +312,7 @@ private WorkerService createPulsarFunctionWorker(ServiceConfiguration config) { System.setProperty(JAVA_INSTANCE_JAR_PROPERTY, FutureUtil.class.getProtectionDomain().getCodeSource().getLocation().getPath()); - + workerConfig = new WorkerConfig(); workerConfig.setPulsarFunctionsNamespace(pulsarFunctionsNamespace); workerConfig.setSchedulerClassName( @@ -447,7 +447,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); @@ -463,7 +463,7 @@ private void testE2EPulsarFunction(String jarFilePathUrl) throws Exception { } catch (PulsarAdminException e) { return false; } - }, 5, 150); + }, 50, 150); Message msg = consumer.receive(5, TimeUnit.SECONDS); String receivedPropertyValue = msg.getProperty(propertyKey); @@ -483,7 +483,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); @@ -713,7 +713,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); @@ -908,7 +908,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); @@ -1063,7 +1063,7 @@ public void testPulsarFunctionStats() throws Exception { // get stats after producing functionStats = functionRuntimeManager.getFunctionStats(tenant, namespacePortion, functionName, null); - + functionStatsFromAdmin = admin.functions().getFunctionStats(tenant, namespacePortion, functionName); @@ -1200,7 +1200,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); @@ -1244,7 +1244,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); @@ -1287,7 +1287,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); @@ -1351,7 +1351,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); @@ -1366,7 +1366,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); @@ -1381,7 +1381,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); @@ -1424,7 +1424,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) -> { @@ -1433,7 +1433,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); @@ -1447,7 +1447,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; @@ -1490,7 +1490,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); @@ -1506,7 +1506,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); @@ -1517,7 +1517,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 @@ -1531,7 +1531,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 @@ -1543,7 +1543,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); From b006cccb9424befc9ec5aecafcecc12c4cc11318 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Tue, 11 Feb 2020 11:18:55 -0800 Subject: [PATCH 3/3] Fixed merge conflict --- .../worker/PulsarFunctionE2ESecurityTest.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java index 8d6d3819217df..ebe79b13c0531 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionE2ESecurityTest.java @@ -332,7 +332,7 @@ public void testAuthorizationWithAnonymousUser() throws Exception { } catch (PulsarAdminException e) { return false; } - }, 50, 150); + }, 50, 150)); // validate pulsar sink consumer has started on the topic assertEquals(admin1.topics().getStats(sourceTopic).subscriptions.size(), 1); @@ -384,7 +384,7 @@ public void testAuthorizationWithAnonymousUser() throws Exception { } catch (PulsarAdminException e) { return false; } - }, 50, 150); + }, 50, 150)); // test getFunctionInfo try { @@ -603,7 +603,7 @@ public void testAuthorization() throws Exception { } catch (PulsarAdminException e) { return false; } - }, 50, 150); + }, 50, 150)); // validate pulsar sink consumer has started on the topic assertEquals(admin1.topics().getStats(sourceTopic).subscriptions.size(), 1); @@ -655,7 +655,7 @@ public void testAuthorization() throws Exception { } catch (PulsarAdminException e) { return false; } - }, 50, 150); + }, 50, 150)); // test getFunctionInfo try { @@ -796,7 +796,7 @@ public void testAuthorization() throws Exception { } catch (PulsarAdminException e) { return false; } - }, 50, 150); + }, 50, 150)); } } }