JAVA-2858: Ensure all change stream getMore commands are retried

This commit ensures that change streams are resumable even when watching
a collection that has no activity between failovers.

Author: jyemin

driver-core/src/main/com/mongodb/operation/AsyncChangeStreamBatchCursor.java

@@ -125,7 +125,7 @@ public void onResult(final AsyncBatchCursor<T> result, final Throwable t) {
                 } else {
                     wrapped = ((AsyncChangeStreamBatchCursor<T>) result).getWrapped();
                     binding.release(); // release the new change stream batch cursor's reference to the binding
-                    asyncBlock.apply(wrapped, callback);
+                    resumeableOperation(asyncBlock, callback);
                 }
             }
         });

driver-core/src/main/com/mongodb/operation/ChangeStreamBatchCursor.java

@@ -127,19 +127,19 @@ private List<T> convertResults(final List<RawBsonDocument> rawDocuments) {
     }
 
     <R> R resumeableOperation(final Function<BatchCursor<RawBsonDocument>, R> function) {
-        try {
-            return function.apply(wrapped);
-        } catch (MongoNotPrimaryException e) {
-            // Ignore
-        } catch (MongoCursorNotFoundException w) {
-            // Ignore
-        } catch (MongoSocketException e) {
-            // Ignore
+        while (true) {
+            try {
+                return function.apply(wrapped);
+            } catch (MongoNotPrimaryException e) {
+                // Ignore
+            } catch (MongoCursorNotFoundException w) {
+                // Ignore
+            } catch (MongoSocketException e) {
+                // Ignore
+            }
+            wrapped.close();
+            wrapped = ((ChangeStreamBatchCursor<T>) changeStreamOperation.resumeAfter(resumeToken).execute(binding)).getWrapped();
+            binding.release(); // release the new change stream batch cursor's reference to the binding
         }
-        wrapped.close();
-        wrapped = ((ChangeStreamBatchCursor<T>) changeStreamOperation.resumeAfter(resumeToken).execute(binding)).getWrapped();
-        binding.release(); // release the new change stream batch cursor's reference to the binding
-        return function.apply(wrapped);
     }
-
 }