Skip to content

Remove code that won't execute on supported server releases #1704

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
May 30, 2025
Merged

Remove code that won't execute on supported server releases #1704

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7e8ebcc
Remove code that won't execute on supported server releases
jyemin May 2, 2025
437fb96
Fix unit tests
jyemin May 2, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 4 additions & 41 deletions ...er-core/src/main/com/mongodb/internal/connection/InternalStreamConnectionInitializer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,6 @@
import static com.mongodb.internal.connection.CommandHelper.LEGACY_HELLO;
import static com.mongodb.internal.connection.CommandHelper.executeCommand;
import static com.mongodb.internal.connection.CommandHelper.executeCommandAsync;
import static com.mongodb.internal.connection.CommandHelper.executeCommandWithoutCheckingForFailure;
import static com.mongodb.internal.connection.DefaultAuthenticator.USER_NOT_FOUND_CODE;
import static com.mongodb.internal.connection.DescriptionHelper.createConnectionDescription;
import static com.mongodb.internal.connection.DescriptionHelper.createServerDescription;
Expand Down Expand Up @@ -88,7 +87,8 @@ public InternalConnectionInitializationDescription finishHandshake(final Interna
if (Authenticator.shouldAuthenticate(authenticator, connectionDescription)) {
authenticator.authenticate(internalConnection, connectionDescription, operationContext);
}
return completeConnectionDescriptionInitialization(internalConnection, description, operationContext);

return description;
}

@Override
Expand Down Expand Up @@ -121,14 +121,14 @@ public void finishHandshakeAsync(final InternalConnection internalConnection,
ConnectionDescription connectionDescription = description.getConnectionDescription();

if (!Authenticator.shouldAuthenticate(authenticator, connectionDescription)) {
completeConnectionDescriptionInitializationAsync(internalConnection, description, operationContext, callback);
callback.onResult(description, null);
} else {
authenticator.authenticateAsync(internalConnection, connectionDescription, operationContext,
(result1, t1) -> {
if (t1 != null) {
callback.onResult(null, t1);
} else {
completeConnectionDescriptionInitializationAsync(internalConnection, description, operationContext, callback);
callback.onResult(description, null);
}
});
}
Expand Down Expand Up @@ -203,50 +203,13 @@ private BsonDocument createHelloCommand(final Authenticator authenticator, final
return helloCommandDocument;
}

private InternalConnectionInitializationDescription completeConnectionDescriptionInitialization(
final InternalConnection internalConnection,
final InternalConnectionInitializationDescription description,
final OperationContext operationContext) {

if (description.getConnectionDescription().getConnectionId().getServerValue() != null) {
return description;
}

return applyGetLastErrorResult(executeCommandWithoutCheckingForFailure("admin",
new BsonDocument("getlasterror", new BsonInt32(1)), clusterConnectionMode, serverApi,
internalConnection, operationContext),
description);
}

private void setSpeculativeAuthenticateResponse(final BsonDocument helloResult) {
if (authenticator instanceof SpeculativeAuthenticator) {
((SpeculativeAuthenticator) authenticator).setSpeculativeAuthenticateResponse(
helloResult.getDocument("speculativeAuthenticate", null));
}
}

private void completeConnectionDescriptionInitializationAsync(
final InternalConnection internalConnection,
final InternalConnectionInitializationDescription description,
final OperationContext operationContext,
final SingleResultCallback<InternalConnectionInitializationDescription> callback) {

if (description.getConnectionDescription().getConnectionId().getServerValue() != null) {
callback.onResult(description, null);
return;
}

executeCommandAsync("admin", new BsonDocument("getlasterror", new BsonInt32(1)), clusterConnectionMode, serverApi,
internalConnection, operationContext,
(result, t) -> {
if (t != null) {
callback.onResult(description, null);
} else {
callback.onResult(applyGetLastErrorResult(result, description), null);
}
});
}

private InternalConnectionInitializationDescription applyGetLastErrorResult(
final BsonDocument getLastErrorResult,
final InternalConnectionInitializationDescription description) {
Expand Down
65 changes: 10 additions & 55 deletions ...t/com/mongodb/internal/connection/InternalStreamConnectionInitializerSpecification.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,14 +61,14 @@ class InternalStreamConnectionInitializerSpecification extends Specification {
def initializer = new InternalStreamConnectionInitializer(SINGLE, null, null, [], null)

when:
enqueueSuccessfulReplies(false, null)
enqueueSuccessfulReplies(false, 123)
def description = initializer.startHandshake(internalConnection, operationContext)
description = initializer.finishHandshake(internalConnection, description, operationContext)
def connectionDescription = description.connectionDescription
def serverDescription = description.serverDescription

then:
connectionDescription == getExpectedConnectionDescription(connectionDescription.connectionId.localValue, null)
connectionDescription == getExpectedConnectionDescription(connectionDescription.connectionId.localValue, 123)
serverDescription == getExpectedServerDescription(serverDescription)
}

Expand All @@ -77,7 +77,7 @@ class InternalStreamConnectionInitializerSpecification extends Specification {
def initializer = new InternalStreamConnectionInitializer(SINGLE, null, null, [], null)

when:
enqueueSuccessfulReplies(false, null)
enqueueSuccessfulReplies(false, 123)
def futureCallback = new FutureResultCallback<InternalConnectionInitializationDescription>()
initializer.startHandshakeAsync(internalConnection, operationContext, futureCallback)
def description = futureCallback.get()
Expand All @@ -88,7 +88,7 @@ class InternalStreamConnectionInitializerSpecification extends Specification {
def serverDescription = description.serverDescription

then:
connectionDescription == getExpectedConnectionDescription(connectionDescription.connectionId.localValue, null)
connectionDescription == getExpectedConnectionDescription(connectionDescription.connectionId.localValue, 123)
serverDescription == getExpectedServerDescription(serverDescription)
}

Expand All @@ -106,20 +106,6 @@ class InternalStreamConnectionInitializerSpecification extends Specification {
connectionDescription == getExpectedConnectionDescription(connectionDescription.connectionId.localValue, 123)
}

def 'should create correct description with server connection id from hello'() {
given:
def initializer = new InternalStreamConnectionInitializer(SINGLE, null, null, [], null)

when:
enqueueSuccessfulRepliesWithConnectionIdIsHelloResponse(false, 123)
def internalDescription = initializer.startHandshake(internalConnection, operationContext)
def connectionDescription = initializer.finishHandshake(internalConnection, internalDescription, operationContext)
.connectionDescription

then:
connectionDescription == getExpectedConnectionDescription(connectionDescription.connectionId.localValue, 123)
}

def 'should create correct description with server connection id asynchronously'() {
given:
def initializer = new InternalStreamConnectionInitializer(SINGLE, null, null, [], null)
Expand All @@ -137,31 +123,13 @@ class InternalStreamConnectionInitializerSpecification extends Specification {
connectionDescription == getExpectedConnectionDescription(connectionDescription.connectionId.localValue, 123)
}

def 'should create correct description with server connection id from hello asynchronously'() {
given:
def initializer = new InternalStreamConnectionInitializer(SINGLE, null, null, [], null)

when:
enqueueSuccessfulRepliesWithConnectionIdIsHelloResponse(false, 123)
def futureCallback = new FutureResultCallback<InternalConnectionInitializationDescription>()
initializer.startHandshakeAsync(internalConnection, operationContext, futureCallback)
def description = futureCallback.get()
futureCallback = new FutureResultCallback<InternalConnectionInitializationDescription>()
initializer.finishHandshakeAsync(internalConnection, description, operationContext, futureCallback)
description = futureCallback.get()
def connectionDescription = description.connectionDescription

then:
connectionDescription == getExpectedConnectionDescription(connectionDescription.connectionId.localValue, 123)
}

def 'should authenticate'() {
given:
def firstAuthenticator = Mock(Authenticator)
def initializer = new InternalStreamConnectionInitializer(SINGLE, firstAuthenticator, null, [], null)

when:
enqueueSuccessfulReplies(false, null)
enqueueSuccessfulReplies(false, 123)

def internalDescription = initializer.startHandshake(internalConnection, operationContext)
def connectionDescription = initializer.finishHandshake(internalConnection, internalDescription, operationContext)
Expand All @@ -178,7 +146,7 @@ class InternalStreamConnectionInitializerSpecification extends Specification {
def initializer = new InternalStreamConnectionInitializer(SINGLE, authenticator, null, [], null)

when:
enqueueSuccessfulReplies(false, null)
enqueueSuccessfulReplies(false, 123)

def futureCallback = new FutureResultCallback<InternalConnectionInitializationDescription>()
initializer.startHandshakeAsync(internalConnection, operationContext, futureCallback)
Expand All @@ -198,7 +166,7 @@ class InternalStreamConnectionInitializerSpecification extends Specification {
def initializer = new InternalStreamConnectionInitializer(SINGLE, authenticator, null, [], null)

when:
enqueueSuccessfulReplies(true, null)
enqueueSuccessfulReplies(true, 123)

def internalDescription = initializer.startHandshake(internalConnection, operationContext)
def connectionDescription = initializer.finishHandshake(internalConnection, internalDescription, operationContext)
Expand All @@ -215,7 +183,7 @@ class InternalStreamConnectionInitializerSpecification extends Specification {
def initializer = new InternalStreamConnectionInitializer(SINGLE, authenticator, null, [], null)

when:
enqueueSuccessfulReplies(true, null)
enqueueSuccessfulReplies(true, 123)

def futureCallback = new FutureResultCallback<InternalConnectionInitializationDescription>()
initializer.startHandshakeAsync(internalConnection, operationContext, futureCallback)
Expand All @@ -240,7 +208,7 @@ class InternalStreamConnectionInitializerSpecification extends Specification {
}

when:
enqueueSuccessfulReplies(false, null)
enqueueSuccessfulReplies(false, 123)
if (async) {
def callback = new FutureResultCallback<InternalConnectionInitializationDescription>()
initializer.startHandshakeAsync(internalConnection, operationContext, callback)
Expand Down Expand Up @@ -277,7 +245,7 @@ class InternalStreamConnectionInitializerSpecification extends Specification {
}

when:
enqueueSuccessfulReplies(false, null)
enqueueSuccessfulReplies(false, 123)
if (async) {
def callback = new FutureResultCallback<InternalConnectionInitializationDescription>()
initializer.startHandshakeAsync(internalConnection, operationContext, callback)
Expand Down Expand Up @@ -477,25 +445,12 @@ class InternalStreamConnectionInitializerSpecification extends Specification {
}

def enqueueSuccessfulReplies(final boolean isArbiter, final Integer serverConnectionId) {
internalConnection.enqueueReply(buildSuccessfulReply(
'{ok: 1, ' +
'maxWireVersion: 3' +
(isArbiter ? ', isreplicaset: true, arbiterOnly: true' : '') +
'}'))
internalConnection.enqueueReply(buildSuccessfulReply(
'{ok: 1 ' +
(serverConnectionId == null ? '' : ', connectionId: ' + serverConnectionId) +
'}'))
}

def enqueueSuccessfulRepliesWithConnectionIdIsHelloResponse(final boolean isArbiter, final Integer serverConnectionId) {
internalConnection.enqueueReply(buildSuccessfulReply(
'{ok: 1, ' +
'maxWireVersion: 3,' +
'connectionId: ' + serverConnectionId +
(isArbiter ? ', isreplicaset: true, arbiterOnly: true' : '') +
'}'))
internalConnection.enqueueReply(buildSuccessfulReply('{ok: 1, versionArray : [3, 0, 0]}'))
}

def enqueueSpeculativeAuthenticationResponsesForScramSha256() {
Expand Down