Either CommandFailed or CommandSucceeded event should be published, not both (#793)

This fixes a bug in the synchronous driver in which the driver sends a CommandSucceededEvent,
followed by a CommandFailedEvent (for the same command) in the case were the server replied
successfully but the Decoder configured to decode the response throws an exception.

JAVA-4321

Author: jyemin

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

@@ -315,16 +315,17 @@ public <T> T sendAndReceive(final CommandMessage message, final Decoder<T> decod
 
         try {
             sendCommandMessage(message, bsonOutput, sessionContext);
-            if (message.isResponseExpected()) {
-                return receiveCommandMessageResponse(decoder, commandEventSender, sessionContext, 0);
-            } else {
-                commandEventSender.sendSucceededEventForOneWayCommand();
-                return null;
-            }
         } catch (RuntimeException e) {
             commandEventSender.sendFailedEvent(e);
             throw e;
         }
+
+        if (message.isResponseExpected()) {
+            return receiveCommandMessageResponse(decoder, commandEventSender, sessionContext, 0);
+        } else {
+            commandEventSender.sendSucceededEventForOneWayCommand();
+            return null;
+        }
     }
 
     @Override
@@ -341,7 +342,7 @@ public <T> void send(final CommandMessage message, final Decoder<T> decoder, fin
     @Override
     public <T> T receive(final Decoder<T> decoder, final SessionContext sessionContext) {
         isTrue("Response is expected", hasMoreToCome);
-        return receiveCommandMessageResponse(decoder, null, sessionContext, 0);
+        return receiveCommandMessageResponse(decoder, new NoOpCommandEventSender(), sessionContext, 0);
     }
 
     @Override
@@ -352,7 +353,7 @@ public boolean supportsAdditionalTimeout() {
     @Override
     public <T> T receive(final Decoder<T> decoder, final SessionContext sessionContext, final int additionalTimeout) {
         isTrue("Response is expected", hasMoreToCome);
-        return receiveCommandMessageResponse(decoder, null, sessionContext, additionalTimeout);
+        return receiveCommandMessageResponse(decoder, new NoOpCommandEventSender(), sessionContext, additionalTimeout);
     }
 
     @Override
@@ -392,27 +393,32 @@ private void sendCommandMessage(final CommandMessage message,
     private <T> T receiveCommandMessageResponse(final Decoder<T> decoder,
                                                 final CommandEventSender commandEventSender, final SessionContext sessionContext,
                                                 final int additionalTimeout) {
+        boolean commandSuccessful = false;
         try (ResponseBuffers responseBuffers = receiveMessageWithAdditionalTimeout(additionalTimeout)) {
             updateSessionContext(sessionContext, responseBuffers);
             if (!isCommandOk(responseBuffers)) {
-                throw getCommandFailureException(responseBuffers.getResponseDocument(responseTo, new BsonDocumentCodec()),
-                        description.getServerAddress());
+                throw getCommandFailureException(responseBuffers.getResponseDocument(responseTo,
+                        new BsonDocumentCodec()), description.getServerAddress());
             }
 
-            if (commandEventSender != null) {
-                commandEventSender.sendSucceededEvent(responseBuffers);
-            }
+            commandSuccessful = true;
+            commandEventSender.sendSucceededEvent(responseBuffers);
 
             T commandResult = getCommandResult(decoder, responseBuffers, responseTo);
-
             hasMoreToCome = responseBuffers.getReplyHeader().hasMoreToCome();
             if (hasMoreToCome) {
                 responseTo = responseBuffers.getReplyHeader().getRequestId();
             } else {
                 responseTo = 0;
             }
+
             return commandResult;
-        }
+        } catch (RuntimeException e) {
+            if (!commandSuccessful) {
+                commandEventSender.sendFailedEvent(e);
+            }
+            throw e;
+    }
     }
 
     @Override

driver-core/src/test/unit/com/mongodb/internal/connection/InternalStreamConnectionSpecification.groovy

@@ -16,7 +16,6 @@
 
 package com.mongodb.internal.connection
 
-
 import com.mongodb.MongoCommandException
 import com.mongodb.MongoInternalException
 import com.mongodb.MongoNamespace
@@ -44,10 +43,13 @@ import com.mongodb.internal.session.SessionContext
 import com.mongodb.internal.validator.NoOpFieldNameValidator
 import org.bson.BsonDocument
 import org.bson.BsonInt32
+import org.bson.BsonReader
 import org.bson.BsonString
 import org.bson.ByteBuf
 import org.bson.ByteBufNIO
 import org.bson.codecs.BsonDocumentCodec
+import org.bson.codecs.DecoderContext
+import org.bson.codecs.configuration.CodecConfigurationException
 import spock.lang.Specification
 
 import java.nio.ByteBuffer
@@ -504,6 +506,29 @@ class InternalStreamConnectionSpecification extends Specification {
                         new BsonDocument('ok', new BsonInt32(1)), 1000)])
     }
 
+    def 'should send events for successful command with decoding error'() {
+        given:
+        def connection = getOpenedConnection()
+        def pingCommandDocument = new BsonDocument('ping', new BsonInt32(1))
+        def commandMessage = new CommandMessage(cmdNamespace, pingCommandDocument, fieldNameValidator, primary(), messageSettings, null)
+        stream.getBuffer(1024) >> { new ByteBufNIO(ByteBuffer.wrap(new byte[1024])) }
+        stream.read(16, 0) >> helper.defaultMessageHeader(commandMessage.getId())
+        stream.read(90, 0) >> helper.defaultReply()
+
+        when:
+        connection.sendAndReceive(commandMessage, {
+            BsonReader reader, DecoderContext decoderContext -> throw new CodecConfigurationException('')
+        }, NoOpSessionContext.INSTANCE, IgnorableRequestContext.INSTANCE)
+
+        then:
+        thrown(CodecConfigurationException)
+        commandListener.eventsWereDelivered([
+                new CommandStartedEvent(1, connection.getDescription(), 'admin', 'ping',
+                        pingCommandDocument.append('$db', new BsonString('admin'))),
+                new CommandSucceededEvent(1, connection.getDescription(), 'ping',
+                        new BsonDocument('ok', new BsonInt32(1)), 1000)])
+    }
+
     def 'should extract cluster and operation time into session context'() {
         given:
         def connection = getOpenedConnection()
@@ -747,6 +772,39 @@ class InternalStreamConnectionSpecification extends Specification {
                         new BsonDocument('ok', new BsonInt32(1)), 1000)])
     }
 
+    def 'should send events for successful asynchronous command with decoding error'() {
+        given:
+        def connection = getOpenedConnection()
+        def pingCommandDocument = new BsonDocument('ping', new BsonInt32(1))
+        def commandMessage = new CommandMessage(cmdNamespace, pingCommandDocument, fieldNameValidator, primary(), messageSettings, null)
+        def callback = new FutureResultCallback()
+
+        stream.getBuffer(1024) >> { new ByteBufNIO(ByteBuffer.wrap(new byte[1024])) }
+        stream.writeAsync(_, _) >> { buffers, handler ->
+            handler.completed(null)
+        }
+        stream.readAsync(16, _) >> { numBytes, handler ->
+            handler.completed(helper.defaultMessageHeader(commandMessage.getId()))
+        }
+        stream.readAsync(90, _) >> { numBytes, handler ->
+            handler.completed(helper.defaultReply())
+        }
+
+        when:
+        connection.sendAndReceiveAsync(commandMessage, {
+            BsonReader reader, DecoderContext decoderContext -> throw new CodecConfigurationException('')
+        }, NoOpSessionContext.INSTANCE, IgnorableRequestContext.INSTANCE, callback)
+        callback.get()
+
+        then:
+        thrown(CodecConfigurationException)
+        commandListener.eventsWereDelivered([
+                new CommandStartedEvent(1, connection.getDescription(), 'admin', 'ping',
+                        pingCommandDocument.append('$db', new BsonString('admin'))),
+                new CommandSucceededEvent(1, connection.getDescription(), 'ping',
+                        new BsonDocument('ok', new BsonInt32(1)), 1000)])
+    }
+
 
     def 'should send events for asynchronous command failure with exception writing message'() {
         given: