Declare MongoInterruptedException as a non-retryable error

JAVA-3291

Author: jstewart148

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

@@ -19,6 +19,7 @@
 import com.mongodb.MongoChangeStreamException;
 import com.mongodb.MongoCursorNotFoundException;
 import com.mongodb.MongoException;
+import com.mongodb.MongoInterruptedException;
 import com.mongodb.MongoNotPrimaryException;
 import com.mongodb.MongoSocketException;
 
@@ -32,7 +33,7 @@ final class ChangeStreamBatchCursorHelper {
     private static final List<String> NONRESUMABLE_CHANGE_STREAM_ERROR_LABELS = asList("NonResumableChangeStreamError");
 
     static boolean isRetryableError(final Throwable t) {
-        if (!(t instanceof MongoException) || t instanceof MongoChangeStreamException) {
+        if (!(t instanceof MongoException) || t instanceof MongoChangeStreamException || t instanceof MongoInterruptedException) {
             return false;
         } else if (t instanceof MongoNotPrimaryException || t instanceof MongoCursorNotFoundException
                 || t instanceof MongoSocketException) {

driver-sync/src/test/functional/com/mongodb/client/ChangeStreamProseTest.java

@@ -19,6 +19,7 @@
 import com.mongodb.MongoChangeStreamException;
 import com.mongodb.MongoCommandException;
 import com.mongodb.MongoException;
+import com.mongodb.MongoInterruptedException;
 import com.mongodb.MongoQueryException;
 import com.mongodb.client.internal.MongoChangeStreamCursorImpl;
 import com.mongodb.client.model.Aggregates;
@@ -59,6 +60,43 @@ public void setUp() {
         collection.insertOne(Document.parse("{ _id : 0 }"));
     }
 
+    class ChangeStreamWatcher implements Runnable {
+        private volatile boolean interruptedExceptionOccurred = false;
+        private final MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor;
+
+        ChangeStreamWatcher(final MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor) {
+            this.cursor = cursor;
+        }
+
+        @Override
+        public void run() {
+            try {
+                cursor.next();
+            } catch (final MongoInterruptedException e) {
+                interruptedExceptionOccurred = true;
+            } finally {
+                cursor.close();
+            }
+        }
+
+        public boolean getInterruptedExceptionOccurred() {
+            return interruptedExceptionOccurred;
+        }
+    }
+
+    //
+    // Test that MongoInterruptedException is not retryable so that a thread can be interrupted.
+    //
+    @Test
+    public void testThreadInterrupted() throws InterruptedException {
+        final ChangeStreamWatcher watcher = new ChangeStreamWatcher(collection.watch().cursor());
+        final Thread t = new Thread(watcher);
+        t.start();
+        t.interrupt();
+        t.join();
+        assertTrue(watcher.getInterruptedExceptionOccurred());
+    }
+
     //
     // Test that the ChangeStream continuously tracks the last seen resumeToken.
     //