Add prose test to verify server monitors do not gossip cluster time. (#1766)

JAVA-5546

Author: vbabanin

driver-core/src/test/unit/com/mongodb/event/TestServerMonitorListener.java

@@ -53,6 +53,16 @@ public TestServerMonitorListener(final Iterable<String> listenableEventTypes) {
         events = new ArrayList<>();
     }
 
+    public void reset() {
+        lock.lock();
+        try {
+            events.clear();
+            condition.signalAll();
+        } finally {
+            lock.unlock();
+        }
+    }
+
     public void serverHearbeatStarted(final ServerHeartbeatStartedEvent event) {
         register(event);
     }
@@ -65,7 +75,7 @@ public void serverHeartbeatFailed(final ServerHeartbeatFailedEvent event) {
         register(event);
     }
 
-    public <T> void waitForEvents(final Class<T> type, final Predicate<? super T> matcher, final int count, final Duration duration)
+    public <T> void waitForEvents(final Class<T> type, final Predicate<? super T> matcher, final long count, final Duration duration)
             throws InterruptedException, TimeoutException {
         assertTrue(listenable(type));
         long remainingNanos = duration.toNanos();

driver-sync/src/test/functional/com/mongodb/client/AbstractSessionsProseTest.java

@@ -16,6 +16,7 @@
 
 package com.mongodb.client;
 
+import com.mongodb.ClusterFixture;
 import com.mongodb.MongoClientException;
 import com.mongodb.MongoClientSettings;
 import com.mongodb.MongoCommandException;
@@ -25,6 +26,10 @@
 import com.mongodb.client.model.Updates;
 import com.mongodb.event.CommandListener;
 import com.mongodb.event.CommandStartedEvent;
+import com.mongodb.event.ServerHeartbeatStartedEvent;
+import com.mongodb.event.ServerHeartbeatSucceededEvent;
+import com.mongodb.event.TestServerMonitorListener;
+import com.mongodb.internal.connection.TestCommandListener;
 import org.bson.BsonDocument;
 import org.bson.Document;
 import org.junit.jupiter.api.AfterAll;
@@ -33,22 +38,27 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.time.Duration;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import static com.mongodb.ClusterFixture.getDefaultDatabaseName;
+import static com.mongodb.ClusterFixture.isStandalone;
 import static com.mongodb.client.Fixture.getMongoClientSettingsBuilder;
 import static java.util.Arrays.asList;
 import static java.util.Collections.singletonList;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static org.bson.assertions.Assertions.fail;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
 
 // Prose tests for Sessions specification: https://github.com/mongodb/specifications/tree/master/source/sessions
 // Prose test README: https://github.com/mongodb/specifications/tree/master/source/sessions/tests/README.md
@@ -194,6 +204,61 @@ public void shouldThrowOnExplicitSessionIfConnectionDoesNotSupportSessions() thr
         }
     }
 
+    /* Test 20 from #20-drivers-do-not-gossip-clustertime-on-sdam-commands
+      In this test, we check that the cluster time has not been advanced on client1 through the server monitors, after client2 advanced
+      the cluster time on the deployment/cluster.
+    */
+    @Test
+    public void shouldNotGossipClusterTimeInServerMonitors() throws InterruptedException, TimeoutException {
+        assumeTrue(!isStandalone());
+
+        //given
+        TestServerMonitorListener serverMonitorListener =
+                new TestServerMonitorListener(asList("serverHeartbeatStartedEvent", "serverHeartbeatSucceededEvent",
+                        "serverHeartbeatFailedEvent"));
+        TestCommandListener commandListener = new TestCommandListener();
+        try (MongoClient client1 = getMongoClient(
+                getDirectPrimaryMongoClientSettingsBuilder()
+                        .addCommandListener(commandListener)
+                        .applyToServerSettings(builder -> builder
+                                .heartbeatFrequency(10, MILLISECONDS)
+                                .addServerMonitorListener(serverMonitorListener))
+                        .build());
+             MongoClient client2 = getMongoClient(getDirectPrimaryMongoClientSettingsBuilder()
+                     .build())) {
+
+            Document clusterTime = executePing(client1)
+                    .get("$clusterTime", Document.class);
+
+            //when
+            client2.getDatabase("test")
+                    .getCollection("test")
+                    .insertOne(new Document("advance", "$clusterTime"));
+
+            // wait until the client1 processes the next pair of SDAM heartbeat started + succeeded events.
+            serverMonitorListener.reset();
+            serverMonitorListener.waitForEvents(ServerHeartbeatStartedEvent.class, serverHeartbeatStartedEvent -> true,
+                    1, Duration.ofMillis(20 + ClusterFixture.getPrimaryRTT()));
+            serverMonitorListener.waitForEvents(ServerHeartbeatSucceededEvent.class, serverHeartbeatSucceededEvent -> true,
+                    1, Duration.ofMillis(20 + ClusterFixture.getPrimaryRTT()));
+
+            commandListener.reset();
+            executePing(client1);
+
+            //then
+            List<CommandStartedEvent> pingStartedEvents = commandListener.getCommandStartedEvents("ping");
+            assertEquals(1, pingStartedEvents.size());
+            BsonDocument sentClusterTime = pingStartedEvents.get(0).getCommand().getDocument("$clusterTime");
+
+            assertEquals(clusterTime.toBsonDocument(), sentClusterTime, "Cluster time should not have advanced after the first ping");
+        }
+    }
+
+    private static MongoClientSettings.Builder getDirectPrimaryMongoClientSettingsBuilder() {
+        return getMongoClientSettingsBuilder()
+                .applyToClusterSettings(ClusterFixture::setDirectConnection);
+    }
+
     private static MongoClientSettings.Builder getMongocryptdMongoClientSettingsBuilder() {
         return MongoClientSettings.builder()
                 .applyToClusterSettings(builder ->
@@ -209,5 +274,10 @@ private static Process startMongocryptdProcess() throws IOException {
         processBuilder.redirectOutput(new File("/tmp/mongocryptd.log"));
         return processBuilder.start();
     }
+
+    private static Document executePing(final MongoClient client1) {
+        return client1.getDatabase("admin")
+                .runCommand(new Document("ping", 1));
+    }
 }