You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2017/12/29 18:34:26 UTC
[GitHub] rdhabalia closed pull request #1019: Keep strategically-retry logic common for tests
rdhabalia closed pull request #1019: Keep strategically-retry logic common for tests
URL: https://github.com/apache/incubator-pulsar/pull/1019
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
index 6460efe27..2362f44f4 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
@@ -71,6 +71,7 @@
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
+import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.retryStrategically;
public class AdminApiTest2 extends MockedPulsarServiceBaseTest {
@@ -340,12 +341,8 @@ public void testUpdatePersistencePolicyUpdateManagedCursor() throws Exception {
final int newEnsembleSize = 5;
admin.namespaces().setPersistence(namespace, new PersistencePolicies(newEnsembleSize, 3, 3, newThrottleRate));
- for (int i = 0; i < 5; i++) {
- if ((managedLedger.getConfig().getEnsembleSize() != newEnsembleSize
- && cursor.getThrottleMarkDelete() != newThrottleRate) || i == 4) {
- Thread.sleep(200);
- }
- }
+ retryStrategically((test) -> managedLedger.getConfig().getEnsembleSize() != newEnsembleSize
+ && cursor.getThrottleMarkDelete() != newThrottleRate, 5, 200);
// (1) verify cursor.markDelete has been updated
assertEquals(cursor.getThrottleMarkDelete(), newThrottleRate);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
index 29f2ef65f..1de1f1484 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
@@ -28,6 +28,7 @@
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
import java.util.function.Supplier;
import org.apache.bookkeeper.client.BKException;
@@ -245,5 +246,15 @@ public void close() {
}
};
+ public static void retryStrategically(Predicate<Void> predicate, int retryCount, long intSleepTime)
+ throws Exception {
+ for (int i = 0; i < retryCount; i++) {
+ if (predicate.test(null) || i == (retryCount - 1)) {
+ break;
+ }
+ Thread.sleep(intSleepTime + (intSleepTime * i));
+ }
+ }
+
private static final Logger log = LoggerFactory.getLogger(MockedPulsarServiceBaseTest.class);
}
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 7c0a53325..6b9509ca7 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
@@ -655,13 +655,7 @@ public void testConcurrentBatchMessageAck() throws Exception {
PersistentDispatcherMultipleConsumers dispatcher = (PersistentDispatcherMultipleConsumers) topic
.getSubscription(subscriptionName).getDispatcher();
// check strategically to let ack-message receive by broker
- for (int i = 0; i < 5; i++) {
- if (dispatcher.getConsumers().get(0).getUnackedMessages() == 0 || i == 4) {
- break;
- } else {
- Thread.sleep(150);
- }
- }
+ retryStrategically((test) -> dispatcher.getConsumers().get(0).getUnackedMessages() == 0, 5, 150);
assertEquals(dispatcher.getConsumers().get(0).getUnackedMessages(), 0);
executor.shutdown();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java
index e9d7629c5..347ffaf56 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerBkEnsemblesTests.java
@@ -23,7 +23,6 @@
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
-import java.util.function.Predicate;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl;
@@ -52,6 +51,7 @@
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
+import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.retryStrategically;
/**
*/
@@ -206,14 +206,5 @@ public void testCrashBrokerWithoutCursorLedgerLeak() throws Exception {
}
- void retryStrategically(Predicate<Void> predicate, int retryCount, long intSleepTime) throws Exception {
- for (int i = 0; i < retryCount; i++) {
- if (predicate.test(null) || i == (retryCount - 1)) {
- break;
- }
- Thread.sleep(intSleepTime + (intSleepTime * i));
- }
- }
-
private static final Logger LOG = LoggerFactory.getLogger(BrokerBkEnsemblesTests.class);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceThrottlingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceThrottlingTest.java
index 79d3d0344..89f477144 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceThrottlingTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceThrottlingTest.java
@@ -48,6 +48,8 @@
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
+import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.retryStrategically;
+import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.retryStrategically;
/**
*/
@@ -232,13 +234,7 @@ public void testLookupThrottlingForClientByBrokerInternalRetry() throws Exceptio
startBroker();
// wait strategically for all consumers to reconnect
- for (int i = 0; i < 5; i++) {
- if (!areAllConsumersConnected(consumers)) {
- Thread.sleep(1000 + (i * 500));
- } else {
- break;
- }
- }
+ retryStrategically((test) -> areAllConsumersConnected(consumers), 5, 500);
int totalConnectedConsumers = 0;
for (int i = 0; i < consumers.size(); i++) {
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 6f4fd342b..4200d925d 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
@@ -765,16 +765,8 @@ public void testActiveAndInActiveConsumerEntryCacheBehavior() throws Exception {
subscriber2.close();
// retry strategically until broker clean up closed subscribers and invalidate all cache entries
- int retry = 5;
- for (int i = 0; i < retry; i++) {
- if (entryCache.getSize() != 0) {
- if (i != retry - 1) {
- Thread.sleep(100);
- }
- } else {
- break;
- }
- }
+ retryStrategically((test) -> entryCache.getSize() == 0, 5, 100);
+
// Verify: EntryCache should be cleared
assertTrue(entryCache.getSize() == 0);
subscriber1.close();
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 854690b25..32fb343f5 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
@@ -83,6 +83,7 @@
import com.google.common.collect.Sets;
import jersey.repackaged.com.google.common.collect.Lists;
+import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.retryStrategically;
public class BrokerClientIntegrationTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(BrokerClientIntegrationTest.class);
@@ -260,14 +261,8 @@ public void testCloseBrokerService() throws Exception {
OwnershipCache ownershipCache = pulsar.getNamespaceService().getOwnershipCache();
assertTrue(ownershipCache.getOwnedBundles().keySet().isEmpty());
// Strategical retry
- for (int i = 0; i < 5; i++) {
- if (producer1.getClientCnx() != null || consumer1.getClientCnx() != null
- || producer2.getClientCnx() != null) {
- Thread.sleep(100);
- } else {
- break;
- }
- }
+ retryStrategically((test) -> (producer1.getClientCnx() == null && consumer1.getClientCnx() == null
+ && producer2.getClientCnx() == null), 5, 100);
// [2] All clients must be disconnected and in connecting state
// producer1 must not be able to connect again
assertTrue(producer1.getClientCnx() == null);
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services