Fix integer overflow with max limit & batchsize (#1810)

Eliminate unnecessary killCursors command when batchSize == limit
can overflow if using Integer.MAX_VALUE for both.

JAVA-5667
JAVA-5970

Author: rozza

driver-core/src/main/com/mongodb/internal/operation/FindOperation.java

@@ -390,7 +390,7 @@ private BsonDocument getCommand(final OperationContext operationContext, final i
                 commandDocument.put("limit", new BsonInt32(Math.abs(batchSize)));
             } else if (batchSize != 0) {
                 int effectiveBatchSize = Math.abs(batchSize);
-                if (effectiveBatchSize == limit) {
+                if (effectiveBatchSize == limit && effectiveBatchSize < Integer.MAX_VALUE) {
                     // avoid an open cursor on server side when batchSize and limit are equal
                     effectiveBatchSize++;
                 }

driver-core/src/test/unit/com/mongodb/internal/operation/FindOperationUnitSpecification.groovy

@@ -102,6 +102,27 @@ class FindOperationUnitSpecification extends OperationUnitSpecification {
         version << [[3, 2, 0]] * 10 + [[3, 4, 0]] * 10
     }
 
+    def 'should find with correct command with effective batch size'() {
+        when:
+        def operation = new FindOperation<BsonDocument>(namespace, new BsonDocumentCodec())
+                .batchSize(batchSize)
+                .limit(limit)
+
+        def expectedCommand = new BsonDocument('find', new BsonString(namespace.getCollectionName()))
+                .append('batchSize', new BsonInt32(commandBatchSize))
+                .append('limit', new BsonInt32(commandLimit))
+
+        then:
+        testOperation(operation, [7, 0, 0], expectedCommand, async, commandResult)
+
+        where:
+        async << [true, true, false, false]
+        batchSize << [10, Integer.MAX_VALUE] * 2
+        limit << [10, Integer.MAX_VALUE] * 2
+        commandLimit << [10, Integer.MAX_VALUE] * 2
+        commandBatchSize << [11, Integer.MAX_VALUE] * 2
+    }
+
     def 'should use the readPreference to set secondaryOk for commands'() {
         when:
         def operation = new FindOperation<Document>(namespace, new DocumentCodec())