diff --git a/driver-core/src/test/unit/com/mongodb/event/TestServerMonitorListener.java b/driver-core/src/test/unit/com/mongodb/event/TestServerMonitorListener.java index b009b5094f0..27651c316ea 100644 --- a/driver-core/src/test/unit/com/mongodb/event/TestServerMonitorListener.java +++ b/driver-core/src/test/unit/com/mongodb/event/TestServerMonitorListener.java @@ -53,6 +53,16 @@ public TestServerMonitorListener(final Iterable 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 void waitForEvents(final Class type, final Predicate matcher, final int count, final Duration duration) + public void waitForEvents(final Class type, final Predicate matcher, final long count, final Duration duration) throws InterruptedException, TimeoutException { assertTrue(listenable(type)); long remainingNanos = duration.toNanos(); diff --git a/driver-sync/src/test/functional/com/mongodb/client/AbstractSessionsProseTest.java b/driver-sync/src/test/functional/com/mongodb/client/AbstractSessionsProseTest.java index 4d4dbead8e6..3682bd64ff0 100644 --- a/driver-sync/src/test/functional/com/mongodb/client/AbstractSessionsProseTest.java +++ b/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 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)); + } }