avoid clearing server in cache when scanner closed and interrupted (apache#5313)

When the thread running an accumulo scanner is interrupted it may cause
the server the scanner was reading from to be cleared from the cache.
This can be disruptive in the case where the server is fine. This change
makes a narrow exception to clearing the server from the cache for the
case where the scanner was closed and an interrupt was seen.  The reason
this is so narrow is to avoid failing to invalidate the cache in the
case where there is actually a problem with the server.

Author: keith-turner

core/src/main/java/org/apache/accumulo/core/clientImpl/ScannerIterator.java

@@ -124,6 +124,8 @@ public Entry<Key,Value> next() {
   }
 
   void close() {
+    // setting this so that some errors can be ignored
+    scanState.closeInitiated = true;
     // run actual close operation in the background so this does not block.
     context.executeCleanupTask(() -> {
       synchronized (scanState) {

core/src/main/java/org/apache/accumulo/core/clientImpl/ThriftScanner.java

@@ -22,6 +22,7 @@
 import static java.util.concurrent.TimeUnit.SECONDS;
 
 import java.io.IOException;
+import java.io.InterruptedIOException;
 import java.security.SecureRandom;
 import java.time.Duration;
 import java.util.ArrayList;
@@ -212,6 +213,8 @@ public static class ScanState {
 
     Duration busyTimeout;
 
+    volatile boolean closeInitiated = false;
+
     TabletLocation getErrorLocation() {
       return prevLoc;
     }
@@ -508,8 +511,13 @@ public static List<KeyValue> scan(ClientContext context, ScanState scanState, Du
           TraceUtil.setException(child2, e, false);
           sleepMillis = pause(sleepMillis, maxSleepTime, scanState.runOnScanServer);
         } catch (TException e) {
-          TabletLocator.getLocator(context, scanState.tableId).invalidateCache(context,
-              loc.tablet_location);
+          boolean wasInterruptedAfterClose =
+              e.getCause() != null && e.getCause().getClass().equals(InterruptedIOException.class)
+                  && scanState.closeInitiated;
+          if (!wasInterruptedAfterClose) {
+            TabletLocator.getLocator(context, scanState.tableId).invalidateCache(context,
+                loc.tablet_location);
+          }
           error = "Scan failed, thrift error " + e.getClass().getName() + "  " + e.getMessage()
               + " " + scanState.getErrorLocation();
           if (!error.equals(lastError)) {