Fix NPE with unacknowledged retryable bulk writes

JAVA-2922

Author: rozza

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

@@ -234,6 +234,9 @@ static RequestMessage.EncodingMetadata encodeMessageWithMetadata(final RequestMe
     private static final List<Integer> RECOVERING_CODES = asList(11600, 11602, 13436, 189, 91);
     public static MongoException createSpecialException(final BsonDocument response, final ServerAddress serverAddress,
                                                         final String errorMessageFieldName) {
+        if (response == null) {
+            return null;
+        }
         int errorCode = getErrorCode(response);
         String errorMessage = getErrorMessage(response, errorMessageFieldName);
         if (ErrorCategory.fromErrorCode(errorCode) == ErrorCategory.EXECUTION_TIMEOUT) {

driver-core/src/test/functional/com/mongodb/operation/MixedBulkWriteOperationSpecification.groovy

@@ -1025,6 +1025,31 @@ class MixedBulkWriteOperationSpecification extends OperationFunctionalSpecificat
         ].combinations()
     }
 
+    @IgnoreIf({ !serverVersionAtLeast(3, 6) || !isDiscoverableReplicaSet() })
+    def 'should not fail with unacknowledged writes, retryWrites and failPoints'() {
+        given:
+        def testWrites = getTestWrites()
+        getCollectionHelper().insertDocuments(getTestInserts())
+        def operation = new MixedBulkWriteOperation(getNamespace(), testWrites, true, UNACKNOWLEDGED, true)
+
+        when:
+        enableOnPrimaryTransactionalWriteFailPoint(BsonDocument.parse(failPoint))
+        def result = executeWithSession(operation, async)
+
+        then:
+        result == BulkWriteResult.unacknowledged()
+
+        cleanup:
+        disableOnPrimaryTransactionalWriteFailPoint()
+
+        where:
+        [async, failPoint] << [
+                [true, false],
+                ['{mode: {times: 2}, data: {failBeforeCommitExceptionCode : 1}}',
+                 '{mode: {skip: 2}, data: {failBeforeCommitExceptionCode : 1}}']
+        ].combinations()
+    }
+
     @IgnoreIf({ !serverVersionAtLeast(3, 6) })
     def 'should retry if the connection initially fails'() {
         when: