Merge pull request #1396 from nats-io/service-flapper

Fix test flappers

Author: scottf

build.gradle

@@ -85,8 +85,8 @@ test {
     }
     retry {
         failOnPassedAfterRetry = false
-        maxFailures = 2
-        maxRetries = 2
+        maxFailures = 4
+        maxRetries = 4
     }
     maxParallelForks = Runtime.runtime.availableProcessors()
     systemProperty 'junit.jupiter.execution.timeout.default', '3m'

src/main/java/io/nats/service/Service.java

@@ -24,9 +24,8 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.ReentrantLock;
 
 import static io.nats.client.support.ApiConstants.*;
@@ -50,19 +49,22 @@ public class Service {
     private final ConcurrentHashMap<String, EndpointContext> serviceContexts;
     private final List<EndpointContext> discoveryContexts;
     private final List<Dispatcher> dInternals;
+    private final AtomicReference<ZonedDateTime> startTimeRef;
+    private final CompletableFuture<Boolean> startedFuture;
     private final PingResponse pingResponse;
     private final InfoResponse infoResponse;
 
     private final ReentrantLock startStopLock;
     private CompletableFuture<Boolean> runningIndicator;
-    private ZonedDateTime started;
 
     Service(ServiceBuilder b) {
         String id = new io.nats.client.NUID().next();
         conn = b.conn;
         drainTimeout = b.drainTimeout;
         dInternals = new ArrayList<>();
         startStopLock = new ReentrantLock();
+        startTimeRef = new AtomicReference<>(DateTimeUtils.DEFAULT_TIME);
+        startedFuture = new CompletableFuture<>();
 
         // build responses first. info needs to be available when adding service endpoints.
         pingResponse = new PingResponse(id, b.name, b.version, b.metadata);
@@ -189,7 +191,8 @@ public CompletableFuture<Boolean> startService() {
                 for (EndpointContext ctx : discoveryContexts) {
                     ctx.start();
                 }
-                started = DateTimeUtils.gmtNow();
+                startTimeRef.set(DateTimeUtils.gmtNow());
+                startedFuture.complete(true);
             }
             return runningIndicator;
         }
@@ -302,7 +305,10 @@ public void stop(boolean drain, Throwable t) {
      * Reset the statistics for the endpoints
      */
     public void reset() {
-        started = DateTimeUtils.gmtNow();
+        if (isStarted()) {
+            // has actually been started if the ref has been set
+            startTimeRef.set(DateTimeUtils.gmtNow());
+        }
         for (EndpointContext c : discoveryContexts) {
             c.reset();
         }
@@ -343,6 +349,33 @@ public String getDescription() {
         return infoResponse.getDescription();
     }
 
+    /**
+     * Get whether the service has full started
+     * @return true if started
+     */
+    public boolean isStarted() {
+        return startedFuture.isDone();
+    }
+
+    /**
+     * Get
+     * @param timeout the maximum time to wait
+     * @param unit the time unit of the timeout argument
+     * @return true if started by the timeout
+     */
+    public boolean isStarted(long timeout, TimeUnit unit) {
+        try {
+            return startedFuture.get(timeout, unit);
+        }
+        catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            return false;
+        }
+        catch (ExecutionException | TimeoutException e) {
+            return false;
+        }
+    }
+
     /**
      * Get the drain timeout setting
      * @return the drain timeout setting
@@ -376,7 +409,8 @@ public StatsResponse getStatsResponse() {
         for (EndpointContext c : serviceContexts.values()) {
             endpointStats.add(c.getEndpointStats());
         }
-        return new StatsResponse(pingResponse, started, endpointStats);
+        // StatsResponse handles a start time of DateTimeUtils.DEFAULT_TIME
+        return new StatsResponse(pingResponse, startTimeRef.get(), endpointStats);
     }
 
     /**
@@ -396,6 +430,7 @@ public String toString() {
         JsonUtils.addField(sb, NAME, infoResponse.getName());
         JsonUtils.addField(sb, VERSION, infoResponse.getVersion());
         JsonUtils.addField(sb, DESCRIPTION, infoResponse.getDescription());
+        JsonUtils.addField(sb, STARTED, startTimeRef.get());
         return endJson(sb).toString();
     }
 }

src/test/java/io/nats/client/impl/SimplificationTests.java

@@ -1728,7 +1728,7 @@ public void testReconnectOverOrdered() throws Exception {
 
             ConsumeOptions consumeOptions = ConsumeOptions.builder()
                 .batchSize(100) // small batch size means more round trips
-                .expiresIn(2000) // idle heartbeat is half of this, alarm time is 3 * ihb
+                .expiresIn(1500) // idle heartbeat is half of this, alarm time is 3 * ihb
                 .build();
 
             OrderedConsumerConfiguration ocConfig = new OrderedConsumerConfiguration().filterSubjects(subject);
@@ -1749,7 +1749,7 @@ public void testReconnectOverOrdered() throws Exception {
         // reconnect and get some more messages
         try (NatsTestServer ignored = new NatsTestServer(port, false, true)) {
             standardConnectionWait(nc);
-            sleep(5000); // long enough to get messages and for the hb alarm to have tripped
+            sleep(6000); // long enough to get messages and for the hb alarm to have tripped
         }
         validateConsumerNameForOrdered(orderedConsumerContext, mcon, null);
         assertNotEquals(firstConsumerName, orderedConsumerContext.getConsumerName());
@@ -1759,10 +1759,10 @@ public void testReconnectOverOrdered() throws Exception {
         assertTrue(count2 > count1);
         assertEquals(count2, nextExpectedSequence.get());
 
-        sleep(4000); // enough delay before reconnect to trip hb alarm again
+        sleep(6000); // enough delay before reconnect to trip hb alarm again
         try (NatsTestServer ignored = new NatsTestServer(port, false, true)) {
             standardConnectionWait(nc);
-            sleep(4000); // long enough to get messages and for the hb alarm to have tripped
+            sleep(6000); // long enough to get messages and for the hb alarm to have tripped
 
             try {
                 nc.jetStreamManagement().deleteStream(stream); // it was a file stream clean it up

src/test/java/io/nats/service/ServiceTests.java

@@ -600,6 +600,9 @@ public void testResponsesFromAllInstances() throws Exception {
                 service1.startService();
                 service2.startService();
 
+                assertTrue(service1.isStarted(1, TimeUnit.SECONDS));
+                assertTrue(service2.isStarted(1, TimeUnit.SECONDS));
+
                 Discovery discovery = new Discovery(clientNc);
 
                 List<PingResponse> prs = discovery.ping();