Add Client Metadata Update Support.

.github/CODEOWNERS

@@ -0,0 +1,2 @@
+# Listing code owners is required by DRIVERS-3098
+* @mongodb/dbx-java

.gitmodules

@@ -1,4 +1,3 @@
 [submodule "specifications"]
 	path = driver-core/src/test/resources/specifications
-	url = https://github.com/vbabanin/specifications
-	branch = DRIVERS-2985
+	url = https://github.com/mongodb/specifications

driver-core/src/main/com/mongodb/MongoStalePrimaryException.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.mongodb;
+
+/**
+ * Exception thrown when a replica set primary is identified as a stale primary during Server Discovery and Monitoring (SDAM).
+ * This occurs when a new primary is discovered, causing the previously known primary to be marked stale, typically during network
+ * partitions or elections.
+ *
+ * @since 5.6
+ */
+public class MongoStalePrimaryException extends MongoException {
+
+    /**
+     * Construct an instance.
+     *
+     * @param message the exception message.
+     */
+    public MongoStalePrimaryException(final String message) {
+        super(message);
+    }
+}

driver-core/src/main/com/mongodb/event/ClusterClosedEvent.java

@@ -23,6 +23,8 @@
 /**
  * A cluster closed event.
  *
+ * <p>This event is synonymous with TopologyClosedEvent</p>
+ *
  * @since 3.3
  */
 public final class ClusterClosedEvent {

driver-core/src/main/com/mongodb/event/ClusterDescriptionChangedEvent.java

@@ -24,6 +24,8 @@
 /**
  * An event signifying that the cluster description has changed.
  *
+ * <p>This event is synonymous with TopologyDescriptionChangedEvent</p>
+ *
  * @since 3.3
  */
 public final class ClusterDescriptionChangedEvent {

driver-core/src/main/com/mongodb/event/ClusterOpeningEvent.java

@@ -23,6 +23,8 @@
 /**
  * A cluster opening event.
  *
+ * <p>This event is synonymous with TopologyOpeningEvent</p>
+ *
  * @since 3.3
  */
 public final class ClusterOpeningEvent {

driver-core/src/main/com/mongodb/internal/connection/AbstractMultiServerCluster.java

@@ -17,6 +17,7 @@
 package com.mongodb.internal.connection;
 
 import com.mongodb.MongoException;
+import com.mongodb.MongoStalePrimaryException;
 import com.mongodb.ServerAddress;
 import com.mongodb.connection.ClusterDescription;
 import com.mongodb.connection.ClusterId;
@@ -269,7 +270,7 @@ private boolean handleReplicaSetMemberChanged(final ServerDescription newDescrip
         }
 
         if (isStalePrimary(newDescription)) {
-            invalidatePotentialPrimary(newDescription);
+            invalidatePotentialPrimary(newDescription, new MongoStalePrimaryException("Primary marked stale due to electionId/setVersion mismatch"));
             return false;
         }
 
@@ -300,12 +301,13 @@ private boolean isStalePrimary(final ServerDescription description) {
         }
      }
 
-    private void invalidatePotentialPrimary(final ServerDescription newDescription) {
+    private void invalidatePotentialPrimary(final ServerDescription newDescription, final MongoStalePrimaryException cause) {
         LOGGER.info(format("Invalidating potential primary %s whose (set version, election id) tuple of (%d, %s) "
                         + "is less than one already seen of (%d, %s)",
                 newDescription.getAddress(), newDescription.getSetVersion(), newDescription.getElectionId(),
                 maxSetVersion, maxElectionId));
-        addressToServerTupleMap.get(newDescription.getAddress()).server.resetToConnecting();
+
+        addressToServerTupleMap.get(newDescription.getAddress()).server.resetToConnecting(cause);
     }
 
     /**
@@ -380,7 +382,7 @@ private void invalidateOldPrimaries(final ServerAddress newPrimary) {
                 if (LOGGER.isInfoEnabled()) {
                     LOGGER.info(format("Rediscovering type of existing primary %s", serverTuple.description.getAddress()));
                 }
-                serverTuple.server.invalidate();
+                serverTuple.server.invalidate(new MongoStalePrimaryException("Primary marked stale due to discovery of newer primary"));
             }
         }
     }

driver-core/src/main/com/mongodb/internal/connection/BaseCluster.java

@@ -27,7 +27,6 @@
 import com.mongodb.connection.ClusterDescription;
 import com.mongodb.connection.ClusterId;
 import com.mongodb.connection.ClusterSettings;
-import com.mongodb.connection.ClusterType;
 import com.mongodb.connection.ServerDescription;
 import com.mongodb.event.ClusterClosedEvent;
 import com.mongodb.event.ClusterDescriptionChangedEvent;
@@ -50,7 +49,6 @@
 import com.mongodb.selector.CompositeServerSelector;
 import com.mongodb.selector.ServerSelector;
 
-import java.util.Collections;
 import java.util.Deque;
 import java.util.Iterator;
 import java.util.List;
@@ -64,6 +62,7 @@
 import static com.mongodb.assertions.Assertions.assertNotNull;
 import static com.mongodb.assertions.Assertions.isTrue;
 import static com.mongodb.assertions.Assertions.notNull;
+import static com.mongodb.connection.ClusterType.UNKNOWN;
 import static com.mongodb.connection.ServerDescription.MAX_DRIVER_WIRE_VERSION;
 import static com.mongodb.connection.ServerDescription.MIN_DRIVER_SERVER_VERSION;
 import static com.mongodb.connection.ServerDescription.MIN_DRIVER_WIRE_VERSION;
@@ -72,6 +71,7 @@
 import static com.mongodb.internal.connection.EventHelper.wouldDescriptionsGenerateEquivalentEvents;
 import static com.mongodb.internal.event.EventListenerHelper.singleClusterListener;
 import static com.mongodb.internal.logging.LogMessage.Component.SERVER_SELECTION;
+import static com.mongodb.internal.logging.LogMessage.Component.TOPOLOGY;
 import static com.mongodb.internal.logging.LogMessage.Entry.Name.FAILURE;
 import static com.mongodb.internal.logging.LogMessage.Entry.Name.OPERATION;
 import static com.mongodb.internal.logging.LogMessage.Entry.Name.OPERATION_ID;
@@ -80,11 +80,16 @@
 import static com.mongodb.internal.logging.LogMessage.Entry.Name.SERVER_HOST;
 import static com.mongodb.internal.logging.LogMessage.Entry.Name.SERVER_PORT;
 import static com.mongodb.internal.logging.LogMessage.Entry.Name.TOPOLOGY_DESCRIPTION;
+import static com.mongodb.internal.logging.LogMessage.Entry.Name.TOPOLOGY_ID;
+import static com.mongodb.internal.logging.LogMessage.Entry.Name.TOPOLOGY_NEW_DESCRIPTION;
+import static com.mongodb.internal.logging.LogMessage.Entry.Name.TOPOLOGY_PREVIOUS_DESCRIPTION;
 import static com.mongodb.internal.logging.LogMessage.Level.DEBUG;
 import static com.mongodb.internal.logging.LogMessage.Level.INFO;
 import static com.mongodb.internal.time.Timeout.ZeroSemantics.ZERO_DURATION_MEANS_EXPIRED;
 import static java.lang.String.format;
 import static java.util.Arrays.asList;
+import static java.util.Collections.emptyList;
+import static java.util.Collections.singletonList;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.NANOSECONDS;
 import static java.util.stream.Collectors.toList;
@@ -115,8 +120,10 @@ abstract class BaseCluster implements Cluster {
         this.settings = notNull("settings", settings);
         this.serverFactory = notNull("serverFactory", serverFactory);
         this.clusterListener = singleClusterListener(settings);
-        this.clusterListener.clusterOpening(new ClusterOpeningEvent(clusterId));
-        this.description = new ClusterDescription(settings.getMode(), ClusterType.UNKNOWN, Collections.emptyList(),
+        ClusterOpeningEvent clusterOpeningEvent = new ClusterOpeningEvent(clusterId);
+        this.clusterListener.clusterOpening(clusterOpeningEvent);
+        logTopologyOpening(clusterId, clusterOpeningEvent);
+        this.description = new ClusterDescription(settings.getMode(), UNKNOWN, emptyList(),
                 settings, serverFactory.getSettings());
         this.clientMetadata = clientMetadata;
     }
@@ -220,7 +227,11 @@ public void close() {
         if (!isClosed()) {
             isClosed = true;
             phase.get().countDown();
-            clusterListener.clusterClosed(new ClusterClosedEvent(clusterId));
+            fireChangeEvent(new ClusterDescription(settings.getMode(), UNKNOWN, emptyList(), settings, serverFactory.getSettings()),
+                    description);
+            ClusterClosedEvent clusterClosedEvent = new ClusterClosedEvent(clusterId);
+            clusterListener.clusterClosed(clusterClosedEvent);
+            logTopologyClosedEvent(clusterId, clusterClosedEvent);
             stopWaitQueueHandler();
         }
     }
@@ -247,8 +258,9 @@ protected void updateDescription(final ClusterDescription newDescription) {
      */
     protected void fireChangeEvent(final ClusterDescription newDescription, final ClusterDescription previousDescription) {
         if (!wouldDescriptionsGenerateEquivalentEvents(newDescription, previousDescription)) {
-             clusterListener.clusterDescriptionChanged(
-                     new ClusterDescriptionChangedEvent(getClusterId(), newDescription, previousDescription));
+            ClusterDescriptionChangedEvent changedEvent = new ClusterDescriptionChangedEvent(getClusterId(), newDescription, previousDescription);
+            clusterListener.clusterDescriptionChanged(changedEvent);
+            logTopologyDescriptionChanged(getClusterId(),  changedEvent);
         }
     }
 
@@ -629,4 +641,43 @@ static void logServerSelectionSucceeded(
                             + " Selector: {}, topology description: {}"));
         }
     }
+
+    static void logTopologyOpening(
+            final ClusterId clusterId,
+            final ClusterOpeningEvent clusterOpeningEvent) {
+        if (STRUCTURED_LOGGER.isRequired(DEBUG, clusterId)) {
+            STRUCTURED_LOGGER.log(new LogMessage(
+                    TOPOLOGY, DEBUG, "Starting topology monitoring", clusterId,
+                    singletonList(new Entry(TOPOLOGY_ID, clusterId)),
+                    "Starting monitoring for topology with ID {}"));
+        }
+    }
+
+    static void logTopologyDescriptionChanged(
+            final ClusterId clusterId,
+            final ClusterDescriptionChangedEvent clusterDescriptionChangedEvent) {
+        if (STRUCTURED_LOGGER.isRequired(DEBUG, clusterId)) {
+            STRUCTURED_LOGGER.log(new LogMessage(
+                    TOPOLOGY, DEBUG, "Topology description changed", clusterId,
+                    asList(
+                            new Entry(TOPOLOGY_ID, clusterId),
+                            new Entry(TOPOLOGY_PREVIOUS_DESCRIPTION,
+                                    clusterDescriptionChangedEvent.getPreviousDescription().getShortDescription()),
+                            new Entry(TOPOLOGY_NEW_DESCRIPTION,
+                                    clusterDescriptionChangedEvent.getNewDescription().getShortDescription())),
+                    "Description changed for topology with ID {}. Previous description: {}. New description: {}"));
+        }
+    }
+
+    static void logTopologyClosedEvent(
+            final ClusterId clusterId,
+            final ClusterClosedEvent clusterClosedEvent) {
+        if (STRUCTURED_LOGGER.isRequired(DEBUG, clusterId)) {
+            STRUCTURED_LOGGER.log(new LogMessage(
+                    TOPOLOGY, DEBUG, "Stopped topology monitoring", clusterId,
+                    singletonList(new Entry(TOPOLOGY_ID, clusterId)),
+                    "Stopped monitoring for topology with ID {}"));
+        }
+    }
+
 }

driver-core/src/main/com/mongodb/internal/connection/ClusterableServer.java

@@ -16,6 +16,8 @@
 
 package com.mongodb.internal.connection;
 
+import com.mongodb.MongoException;
+
 import java.util.List;
 
 import static java.util.Arrays.asList;
@@ -30,13 +32,13 @@ interface ClusterableServer extends Server {
     /**
      * Reset server description to connecting state
      */
-    void resetToConnecting();
+    void resetToConnecting(MongoException cause);
 
     /**
      * Invalidate the description of this server.  Implementation of this method should not block, but rather trigger an asynchronous
      * attempt to connect with the server in order to determine its current status.
      */
-    void invalidate();
+    void invalidate(MongoException cause);
 
     /**
      * <p>Closes the server.  Instances that have been closed will no longer be available for use.</p>

driver-core/src/main/com/mongodb/internal/connection/DefaultConnectionPool.java

@@ -430,7 +430,7 @@ void doMaintenance() {
                         try {
                             sdamProvider.optional().ifPresent(sdam -> {
                                 if (!silentlyComplete.test(actualException)) {
-                                    sdam.handleExceptionBeforeHandshake(SdamIssue.specific(actualException, sdam.context(newConnection)));
+                                    sdam.handleExceptionBeforeHandshake(SdamIssue.of(actualException, sdam.context(newConnection)));
                                 }
                             });
                         } catch (Exception suppressed) {

driver-core/src/main/com/mongodb/internal/connection/DefaultSdamServerDescriptionManager.java

@@ -56,7 +56,7 @@ final class DefaultSdamServerDescriptionManager implements SdamServerDescription
     }
 
     @Override
-    public void update(final ServerDescription candidateDescription) {
+    public void monitorUpdate(final ServerDescription candidateDescription) {
         cluster.withLock(() -> {
             if (TopologyVersionHelper.newer(description.getTopologyVersion(), candidateDescription.getTopologyVersion())) {
                 return;
@@ -82,6 +82,18 @@ public void update(final ServerDescription candidateDescription) {
         });
     }
 
+    @Override
+    public void updateToUnknown(final ServerDescription candidateDescription) {
+        assertTrue(candidateDescription.getType() == UNKNOWN);
+        cluster.withLock(() -> {
+            if (TopologyVersionHelper.newer(description.getTopologyVersion(), candidateDescription.getTopologyVersion())) {
+                return;
+            }
+
+            updateDescription(candidateDescription);
+        });
+    }
+
     @Override
     public void handleExceptionBeforeHandshake(final SdamIssue sdamIssue) {
         cluster.withLock(() -> handleException(sdamIssue, true));
@@ -128,7 +140,7 @@ private void handleException(final SdamIssue sdamIssue, final boolean beforeHand
             updateDescription(sdamIssue.serverDescription());
             connectionPool.invalidate(sdamIssue.exception().orElse(null));
             serverMonitor.cancelCurrentCheck();
-        } else if (sdamIssue.relatedToWriteConcern() || !sdamIssue.specific()) {
+        } else if (sdamIssue.relatedToWriteConcern() || sdamIssue.relatedToStalePrimary()) {
             updateDescription(sdamIssue.serverDescription());
             serverMonitor.connect();
         }

driver-core/src/main/com/mongodb/internal/connection/DefaultServer.java

@@ -96,7 +96,7 @@ public Connection getConnection(final OperationContext operationContext) {
             try {
                 operationEnd();
                 if (e instanceof MongoException) {
-                    sdam.handleExceptionBeforeHandshake(SdamIssue.specific(e, exceptionContext));
+                    sdam.handleExceptionBeforeHandshake(SdamIssue.of(e, exceptionContext));
                 }
             } catch (Exception suppressed) {
                 e.addSuppressed(suppressed);
@@ -118,7 +118,7 @@ public void getConnectionAsync(final OperationContext operationContext, final Si
             if (t != null) {
                 try {
                     operationEnd();
-                    sdam.handleExceptionBeforeHandshake(SdamIssue.specific(t, exceptionContext));
+                    sdam.handleExceptionBeforeHandshake(SdamIssue.of(t, exceptionContext));
                 } catch (Exception suppressed) {
                     t.addSuppressed(suppressed);
                 } finally {
@@ -150,14 +150,14 @@ private void operationEnd() {
     }
 
     @Override
-    public void resetToConnecting() {
-        sdam.update(unknownConnectingServerDescription(serverId, null));
+    public void resetToConnecting(final MongoException cause) {
+        sdam.updateToUnknown(unknownConnectingServerDescription(serverId, cause));
     }
 
     @Override
-    public void invalidate() {
+    public void invalidate(final MongoException cause) {
         if (!isClosed()) {
-            sdam.handleExceptionAfterHandshake(SdamIssue.unspecified(sdam.context()));
+            sdam.handleExceptionAfterHandshake(SdamIssue.of(cause, sdam.context()));
         }
     }
 
@@ -208,7 +208,7 @@ public <T> T execute(final CommandProtocol<T> protocol, final InternalConnection
                         .execute(connection);
             } catch (MongoException e) {
                 try {
-                    sdam.handleExceptionAfterHandshake(SdamIssue.specific(e, sdam.context(connection)));
+                    sdam.handleExceptionAfterHandshake(SdamIssue.of(e, sdam.context(connection)));
                 } catch (Exception suppressed) {
                     e.addSuppressed(suppressed);
                 }
@@ -231,7 +231,7 @@ public <T> void executeAsync(final CommandProtocol<T> protocol, final InternalCo
                     .executeAsync(connection, errorHandlingCallback((result, t) -> {
                 if (t != null) {
                     try {
-                        sdam.handleExceptionAfterHandshake(SdamIssue.specific(t, sdam.context(connection)));
+                        sdam.handleExceptionAfterHandshake(SdamIssue.of(t, sdam.context(connection)));
                     } catch (Exception suppressed) {
                         t.addSuppressed(suppressed);
                     } finally {

driver-core/src/main/com/mongodb/internal/connection/DefaultServerMonitor.java

@@ -198,7 +198,7 @@ public void run() {
                     }
 
                     logStateChange(previousServerDescription, currentServerDescription);
-                    sdamProvider.get().update(currentServerDescription);
+                    sdamProvider.get().monitorUpdate(currentServerDescription);
 
                     if ((shouldStreamResponses && currentServerDescription.getType() != UNKNOWN)
                             || (connection != null && connection.hasMoreToCome())