Skip to content

Commit 4f3f24d

Browse files
vbabaninvbabanin
authored
Include error messages for stale primary (#1714)
- Add electionId/setVersion mismatch error messages to ServerDescription . - Add stale primary error messages to ServerDescription. - Rename update method to updateToUnknown for clarity. JAVA-5697
1 parent a7a3df5 commit 4f3f24d

19 files changed

+201
-55
lines changed

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

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
/*
2+
* Copyright 2008-present MongoDB, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.mongodb;
18+
19+
/**
20+
* Exception thrown when a replica set primary is identified as a stale primary during Server Discovery and Monitoring (SDAM).
21+
* This occurs when a new primary is discovered, causing the previously known primary to be marked stale, typically during network
22+
* partitions or elections.
23+
*
24+
* @since 5.6
25+
*/
26+
public class MongoStalePrimaryException extends MongoException {
27+
28+
/**
29+
* Construct an instance.
30+
*
31+
* @param message the exception message.
32+
*/
33+
public MongoStalePrimaryException(final String message) {
34+
super(message);
35+
}
36+
}

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

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
package com.mongodb.internal.connection;
1818

1919
import com.mongodb.MongoException;
20+
import com.mongodb.MongoStalePrimaryException;
2021
import com.mongodb.ServerAddress;
2122
import com.mongodb.connection.ClusterDescription;
2223
import com.mongodb.connection.ClusterId;
@@ -266,7 +267,7 @@ private boolean handleReplicaSetMemberChanged(final ServerDescription newDescrip
266267
}
267268

268269
if (isStalePrimary(newDescription)) {
269-
invalidatePotentialPrimary(newDescription);
270+
invalidatePotentialPrimary(newDescription, new MongoStalePrimaryException("Primary marked stale due to electionId/setVersion mismatch"));
270271
return false;
271272
}
272273

@@ -297,12 +298,13 @@ private boolean isStalePrimary(final ServerDescription description) {
297298
}
298299
}
299300

300-
private void invalidatePotentialPrimary(final ServerDescription newDescription) {
301+
private void invalidatePotentialPrimary(final ServerDescription newDescription, final MongoStalePrimaryException cause) {
301302
LOGGER.info(format("Invalidating potential primary %s whose (set version, election id) tuple of (%d, %s) "
302303
+ "is less than one already seen of (%d, %s)",
303304
newDescription.getAddress(), newDescription.getSetVersion(), newDescription.getElectionId(),
304305
maxSetVersion, maxElectionId));
305-
addressToServerTupleMap.get(newDescription.getAddress()).server.resetToConnecting();
306+
307+
addressToServerTupleMap.get(newDescription.getAddress()).server.resetToConnecting(cause);
306308
}
307309

308310
/**
@@ -377,7 +379,7 @@ private void invalidateOldPrimaries(final ServerAddress newPrimary) {
377379
if (LOGGER.isInfoEnabled()) {
378380
LOGGER.info(format("Rediscovering type of existing primary %s", serverTuple.description.getAddress()));
379381
}
380-
serverTuple.server.invalidate();
382+
serverTuple.server.invalidate(new MongoStalePrimaryException("Primary marked stale due to discovery of newer primary"));
381383
}
382384
}
383385
}

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

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@
1616

1717
package com.mongodb.internal.connection;
1818

19+
import com.mongodb.MongoException;
20+
1921
import java.util.List;
2022

2123
import static java.util.Arrays.asList;
@@ -30,13 +32,13 @@ interface ClusterableServer extends Server {
3032
/**
3133
* Reset server description to connecting state
3234
*/
33-
void resetToConnecting();
35+
void resetToConnecting(MongoException cause);
3436

3537
/**
3638
* Invalidate the description of this server. Implementation of this method should not block, but rather trigger an asynchronous
3739
* attempt to connect with the server in order to determine its current status.
3840
*/
39-
void invalidate();
41+
void invalidate(MongoException cause);
4042

4143
/**
4244
* <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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -430,7 +430,7 @@ void doMaintenance() {
430430
try {
431431
sdamProvider.optional().ifPresent(sdam -> {
432432
if (!silentlyComplete.test(actualException)) {
433-
sdam.handleExceptionBeforeHandshake(SdamIssue.specific(actualException, sdam.context(newConnection)));
433+
sdam.handleExceptionBeforeHandshake(SdamIssue.of(actualException, sdam.context(newConnection)));
434434
}
435435
});
436436
} catch (Exception suppressed) {

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

Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ final class DefaultSdamServerDescriptionManager implements SdamServerDescription
5656
}
5757

5858
@Override
59-
public void update(final ServerDescription candidateDescription) {
59+
public void monitorUpdate(final ServerDescription candidateDescription) {
6060
cluster.withLock(() -> {
6161
if (TopologyVersionHelper.newer(description.getTopologyVersion(), candidateDescription.getTopologyVersion())) {
6262
return;
@@ -82,6 +82,18 @@ public void update(final ServerDescription candidateDescription) {
8282
});
8383
}
8484

85+
@Override
86+
public void updateToUnknown(final ServerDescription candidateDescription) {
87+
assertTrue(candidateDescription.getType() == UNKNOWN);
88+
cluster.withLock(() -> {
89+
if (TopologyVersionHelper.newer(description.getTopologyVersion(), candidateDescription.getTopologyVersion())) {
90+
return;
91+
}
92+
93+
updateDescription(candidateDescription);
94+
});
95+
}
96+
8597
@Override
8698
public void handleExceptionBeforeHandshake(final SdamIssue sdamIssue) {
8799
cluster.withLock(() -> handleException(sdamIssue, true));
@@ -128,7 +140,7 @@ private void handleException(final SdamIssue sdamIssue, final boolean beforeHand
128140
updateDescription(sdamIssue.serverDescription());
129141
connectionPool.invalidate(sdamIssue.exception().orElse(null));
130142
serverMonitor.cancelCurrentCheck();
131-
} else if (sdamIssue.relatedToWriteConcern() || !sdamIssue.specific()) {
143+
} else if (sdamIssue.relatedToWriteConcern() || sdamIssue.relatedToStalePrimary()) {
132144
updateDescription(sdamIssue.serverDescription());
133145
serverMonitor.connect();
134146
}

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

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -96,7 +96,7 @@ public Connection getConnection(final OperationContext operationContext) {
9696
try {
9797
operationEnd();
9898
if (e instanceof MongoException) {
99-
sdam.handleExceptionBeforeHandshake(SdamIssue.specific(e, exceptionContext));
99+
sdam.handleExceptionBeforeHandshake(SdamIssue.of(e, exceptionContext));
100100
}
101101
} catch (Exception suppressed) {
102102
e.addSuppressed(suppressed);
@@ -118,7 +118,7 @@ public void getConnectionAsync(final OperationContext operationContext, final Si
118118
if (t != null) {
119119
try {
120120
operationEnd();
121-
sdam.handleExceptionBeforeHandshake(SdamIssue.specific(t, exceptionContext));
121+
sdam.handleExceptionBeforeHandshake(SdamIssue.of(t, exceptionContext));
122122
} catch (Exception suppressed) {
123123
t.addSuppressed(suppressed);
124124
} finally {
@@ -150,14 +150,14 @@ private void operationEnd() {
150150
}
151151

152152
@Override
153-
public void resetToConnecting() {
154-
sdam.update(unknownConnectingServerDescription(serverId, null));
153+
public void resetToConnecting(final MongoException cause) {
154+
sdam.updateToUnknown(unknownConnectingServerDescription(serverId, cause));
155155
}
156156

157157
@Override
158-
public void invalidate() {
158+
public void invalidate(final MongoException cause) {
159159
if (!isClosed()) {
160-
sdam.handleExceptionAfterHandshake(SdamIssue.unspecified(sdam.context()));
160+
sdam.handleExceptionAfterHandshake(SdamIssue.of(cause, sdam.context()));
161161
}
162162
}
163163

@@ -208,7 +208,7 @@ public <T> T execute(final CommandProtocol<T> protocol, final InternalConnection
208208
.execute(connection);
209209
} catch (MongoException e) {
210210
try {
211-
sdam.handleExceptionAfterHandshake(SdamIssue.specific(e, sdam.context(connection)));
211+
sdam.handleExceptionAfterHandshake(SdamIssue.of(e, sdam.context(connection)));
212212
} catch (Exception suppressed) {
213213
e.addSuppressed(suppressed);
214214
}
@@ -231,7 +231,7 @@ public <T> void executeAsync(final CommandProtocol<T> protocol, final InternalCo
231231
.executeAsync(connection, errorHandlingCallback((result, t) -> {
232232
if (t != null) {
233233
try {
234-
sdam.handleExceptionAfterHandshake(SdamIssue.specific(t, sdam.context(connection)));
234+
sdam.handleExceptionAfterHandshake(SdamIssue.of(t, sdam.context(connection)));
235235
} catch (Exception suppressed) {
236236
t.addSuppressed(suppressed);
237237
} finally {

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -198,7 +198,7 @@ public void run() {
198198
}
199199

200200
logStateChange(previousServerDescription, currentServerDescription);
201-
sdamProvider.get().update(currentServerDescription);
201+
sdamProvider.get().monitorUpdate(currentServerDescription);
202202

203203
if ((shouldStreamResponses && currentServerDescription.getType() != UNKNOWN)
204204
|| (connection != null && connection.hasMoreToCome())

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -80,12 +80,12 @@ public LoadBalancedServer(final ServerId serverId, final ConnectionPool connecti
8080
}
8181

8282
@Override
83-
public void resetToConnecting() {
83+
public void resetToConnecting(final MongoException cause) {
8484
// no op
8585
}
8686

8787
@Override
88-
public void invalidate() {
88+
public void invalidate(final MongoException cause) {
8989
// no op
9090
}
9191

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

Lines changed: 17 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import com.mongodb.MongoSecurityException;
2323
import com.mongodb.MongoSocketException;
2424
import com.mongodb.MongoSocketReadTimeoutException;
25+
import com.mongodb.MongoStalePrimaryException;
2526
import com.mongodb.annotations.Immutable;
2627
import com.mongodb.annotations.ThreadSafe;
2728
import com.mongodb.connection.ServerDescription;
@@ -43,12 +44,21 @@
4344
*/
4445
@ThreadSafe
4546
interface SdamServerDescriptionManager {
47+
/**
48+
* Receives candidate {@link ServerDescription} from the monitoring activity.
49+
*
50+
* @param candidateDescription A {@link ServerDescription} that may or may not be applied
51+
* {@linkplain TopologyVersionHelper#newer(TopologyVersion, TopologyVersion) depending on}
52+
* its {@link ServerDescription#getTopologyVersion() topology version}.
53+
*/
54+
void monitorUpdate(ServerDescription candidateDescription);
55+
4656
/**
4757
* @param candidateDescription A {@link ServerDescription} that may or may not be applied
4858
* {@linkplain TopologyVersionHelper#newer(TopologyVersion, TopologyVersion) depending on}
4959
* its {@link ServerDescription#getTopologyVersion() topology version}.
5060
*/
51-
void update(ServerDescription candidateDescription);
61+
void updateToUnknown(ServerDescription candidateDescription);
5262

5363
void handleExceptionBeforeHandshake(SdamIssue sdamIssue);
5464

@@ -84,34 +94,17 @@ private SdamIssue(@Nullable final Throwable exception, final Context context) {
8494
this.context = assertNotNull(context);
8595
}
8696

87-
/**
88-
* @see #unspecified(Context)
89-
*/
90-
static SdamIssue specific(final Throwable exception, final Context context) {
97+
static SdamIssue of(final Throwable exception, final Context context) {
9198
return new SdamIssue(assertNotNull(exception), assertNotNull(context));
9299
}
93100

94101
/**
95-
* @see #specific(Throwable, Context)
96-
*/
97-
static SdamIssue unspecified(final Context context) {
98-
return new SdamIssue(null, assertNotNull(context));
99-
}
100-
101-
/**
102-
* @return An exception if and only if this {@link SdamIssue} is {@linkplain #specific()}.
102+
* @return An exception that caused this {@link SdamIssue}.
103103
*/
104104
Optional<Throwable> exception() {
105105
return Optional.ofNullable(exception);
106106
}
107107

108-
/**
109-
* @return {@code true} if and only if this {@link SdamIssue} has an exception {@linkplain #specific(Throwable, Context) specified}.
110-
*/
111-
boolean specific() {
112-
return exception != null;
113-
}
114-
115108
ServerDescription serverDescription() {
116109
return unknownConnectingServerDescription(context.serverId(), exception);
117110
}
@@ -127,6 +120,10 @@ boolean relatedToStateChange() {
127120
return exception instanceof MongoNotPrimaryException || exception instanceof MongoNodeIsRecoveringException;
128121
}
129122

123+
boolean relatedToStalePrimary() {
124+
return exception instanceof MongoStalePrimaryException;
125+
}
126+
130127
/**
131128
* Represents a subset of {@link #relatedToStateChange()}.
132129
*

‎driver-core/src/test/functional/com/mongodb/internal/connection/ServerMonitorSpecification.groovy

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -194,7 +194,14 @@ class ServerMonitorSpecification extends OperationFunctionalSpecification {
194194
def initializeServerMonitor(ServerAddress address) {
195195
SdamServerDescriptionManager sdam = new SdamServerDescriptionManager() {
196196
@Override
197-
void update(final ServerDescription candidateDescription) {
197+
void monitorUpdate(final ServerDescription candidateDescription) {
198+
assert candidateDescription != null
199+
newDescription = candidateDescription
200+
latch.countDown()
201+
}
202+
203+
@Override
204+
void updateToUnknown(final ServerDescription candidateDescription) {
198205
assert candidateDescription != null
199206
newDescription = candidateDescription
200207
latch.countDown()

0 commit comments

Comments
 (0)