Restore thread interrupt status in InterruptedException catch blocks

runners/flink/2.0/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/StreamingImpulseSource.java

@@ -73,6 +73,7 @@ public void run(SourceContext<WindowedValue<byte[]>> ctx) {
           Thread.sleep(intervalMillis);
         }
       } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
         LOG.warn("Interrupted while sleeping", e);
       }
     }

runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/SplittableDoFnOperator.java

@@ -187,6 +187,7 @@ public void close() throws Exception {
       }
     } catch (InterruptedException e) {
       LOG.debug("Could not properly await the termination of the scheduled executor service.", e);
+      Thread.currentThread().interrupt();
       executorService.shutdownNow();
       executorService = null;
     }

runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/StreamingImpulseSource.java

@@ -71,6 +71,7 @@ public void run(SourceContext<WindowedValue<byte[]>> ctx) {
           Thread.sleep(intervalMillis);
         }
       } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
         LOG.warn("Interrupted while sleeping", e);
       }
     }

runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/data/GrpcDataService.java

@@ -113,6 +113,9 @@ public StreamObserver<BeamFnApi.Elements> data(
       // incoming messages are sent to the single multiplexer instance.
       return connectedClient.get().getInboundObserver();
     } catch (InterruptedException | ExecutionException e) {
+      if (e instanceof InterruptedException) {
+        Thread.currentThread().interrupt();
+      }
       throw new RuntimeException(e);
     }
   }

runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/environment/DockerCommand.java

@@ -94,6 +94,9 @@ public String runImage(String imageTag, List<String> dockerOpts, List<String> ar
           ImmutableList.<String>builder().add(dockerExecutable).add("pull").add(imageTag).build(),
           pullTimeout);
     } catch (IOException | TimeoutException | InterruptedException e) {
+      if (e instanceof InterruptedException) {
+        Thread.currentThread().interrupt();
+      }
       if (LOG.isDebugEnabled()) {
         LOG.debug("Unable to pull docker image {}", imageTag, e);
       } else {

runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/status/BeamWorkerStatusGrpcService.java

@@ -117,6 +117,9 @@ public StreamObserver<WorkerStatusResponse> workerStatus(
                 return existingClientFuture;
               }
             } catch (IOException | InterruptedException | ExecutionException e) {
+              if (e instanceof InterruptedException) {
+                Thread.currentThread().interrupt();
+              }
               LOG.warn("Error closing worker status client", e);
             }
           }

runners/jet/src/main/java/org/apache/beam/runners/jet/JetPipelineResult.java

@@ -118,6 +118,9 @@ public State waitUntilFinish(Duration duration) {
       completionFuture.get(duration.getMillis(), TimeUnit.MILLISECONDS);
       return State.DONE;
     } catch (InterruptedException | TimeoutException e) {
+      if (e instanceof InterruptedException) {
+        Thread.currentThread().interrupt();
+      }
       return getState(); // job should be RUNNING or STOPPED
     } catch (ExecutionException e) {
       throw new CompletionException(e.getCause());

runners/prism/java/src/main/java/org/apache/beam/runners/prism/PrismExecutor.java

@@ -79,7 +79,7 @@ void stop() {
     try {
       boolean ignored = executorService.awaitTermination(5000L, TimeUnit.MILLISECONDS);
     } catch (InterruptedException ignored) {
-      // Ignore InterruptedException during shutdown.
+      Thread.currentThread().interrupt();
     }
     if (process == null) {
       return;
@@ -91,7 +91,7 @@ void stop() {
     try {
       process.waitFor();
     } catch (InterruptedException ignored) {
-      // Ignore InterruptedException during shutdown.
+      Thread.currentThread().interrupt();
     }
   }
 

sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer.java

@@ -296,6 +296,9 @@ private void forwardToConsumerForInstructionId(String instructionId, BeamFnApi.E
         outboundObserver.onError(e);
         return;
       } catch (ExecutionException | InterruptedException | PoisonedException e) {
+        if (e instanceof InterruptedException) {
+          Thread.currentThread().interrupt();
+        }
         if (e instanceof PoisonedException || e.getCause() instanceof PoisonedException) {
           LOG.debug("Received data for poisoned instruction {}. Dropping input.", instructionId);
           return;

sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/TransformUpgrader.java

@@ -398,7 +398,7 @@ private static int findAvailablePort() throws IOException {
         // Some systems don't free the port for future use immediately.
         Thread.sleep(100);
       } catch (InterruptedException exn) {
-        // ignore
+        Thread.currentThread().interrupt();
       }
     }
   }

sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/RetryHttpRequestInitializer.java

@@ -183,6 +183,9 @@ private boolean backOffWasSuccessful(BackOff backOff) {
         sleeper.sleep(backOffTime);
         return true;
       } catch (InterruptedException | IOException e) {
+        if (e instanceof InterruptedException) {
+          Thread.currentThread().interrupt();
+        }
         return false;
       }
     }

sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/PythonExternalTransform.java

@@ -447,7 +447,9 @@ private boolean isPythonAvailable() {
         new ProcessBuilder(executable, "--version").start().waitFor();
         return true;
       } catch (IOException | InterruptedException exn) {
-        // Ignore.
+        if (exn instanceof InterruptedException) {
+          Thread.currentThread().interrupt();
+        }
       }
     }
     return false;
@@ -459,7 +461,9 @@ private boolean isDockerAvailable() {
       new ProcessBuilder(executable, "--version").start().waitFor();
       return true;
     } catch (IOException | InterruptedException exn) {
-      // Ignore.
+      if (exn instanceof InterruptedException) {
+        Thread.currentThread().interrupt();
+      }
     }
     return false;
   }

sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/PythonService.java

@@ -144,7 +144,9 @@ private String whichPython() {
         new ProcessBuilder(executable, "--version").start().waitFor();
         return executable;
       } catch (IOException | InterruptedException exn) {
-        // Ignore.
+        if (exn instanceof InterruptedException) {
+          Thread.currentThread().interrupt();
+        }
       }
     }
     throw new RuntimeException("Unable to find a suitable Python executable.");
@@ -171,7 +173,7 @@ public static int findAvailablePort() throws IOException {
         // Some systems don't free the port for future use immediately.
         Thread.sleep(100);
       } catch (InterruptedException exn) {
-        // ignore
+        Thread.currentThread().interrupt();
       }
     }
   }

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/FinalizeBundleHandler.java

@@ -118,7 +118,7 @@ private void cleanupThreadBody() {
         }
       }
     } catch (InterruptedException e) {
-      // We're being shutdown.
+      Thread.currentThread().interrupt();
     } finally {
       lock.unlock();
     }

sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/ShardReadersPool.java

@@ -210,6 +210,7 @@ CustomOptional<KinesisRecord> nextRecord() {
       return CustomOptional.of(record);
     } catch (InterruptedException e) {
       LOG.warn("Interrupted while waiting for KinesisRecord from the buffer");
+      Thread.currentThread().interrupt();
       return CustomOptional.absent();
     }
   }

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryStorageStreamSource.java

@@ -431,6 +431,9 @@ public synchronized BigQueryStorageStreamSource<T> getCurrentSource() {
             // StorageClientImpl.
             future.get(30, TimeUnit.SECONDS);
           } catch (TimeoutException | InterruptedException | ExecutionException e) {
+            if (e instanceof InterruptedException) {
+              Thread.currentThread().interrupt();
+            }
             badSplitPointCalls.inc();
             LOG.info(
                 "Split of stream {} abandoned because current position check failed with {}.",

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/DynamicDestinationsHelpers.java

@@ -468,13 +468,17 @@ private MatchTableDynamicDestinations(
             try (DatasetService datasetService = bqServices.getDatasetService(bqOptions)) {
               return datasetService.getTable(tableReference);
             } catch (InterruptedException | IOException e) {
+              if (e instanceof InterruptedException) {
+                Thread.currentThread().interrupt();
+              }
               LOG.info("Failed to get BigQuery table {}", tableReference);
             }
           } catch (Exception e) {
             throw new RuntimeException(e);
           }
         } while (nextBackOff(Sleeper.DEFAULT, backoff));
       } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
         throw new RuntimeException(e);
       }
       return null;

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIO.java

@@ -1349,6 +1349,9 @@ public void importBatch(FinishBundleContext context) throws IOException {
           // longer needed.
           FileSystems.delete(tempDestinations);
         } catch (IOException | InterruptedException e) {
+          if (e instanceof InterruptedException) {
+            Thread.currentThread().interrupt();
+          }
           ResourceId deadLetterResourceId =
               FileSystems.matchNewResource(deadLetterGcsPath.get(), true);
           LOG.warn(

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java

@@ -3000,7 +3000,7 @@ private void writeMutations(Iterable<Mutation> mutationIterable)
             try {
               sleeper.sleep(sleepTimeMsecs);
             } catch (InterruptedException e) {
-              // ignore.
+              Thread.currentThread().interrupt();
             }
           } else {
             // Some other failure: pass up the stack.

sdks/java/io/hadoop-format/src/main/java/org/apache/beam/sdk/io/hadoop/format/HadoopFormatIO.java

@@ -979,6 +979,9 @@ public KV<K, V> getCurrent() {
                   valueCoder,
                   skipValueClone);
         } catch (IOException | InterruptedException e) {
+          if (e instanceof InterruptedException) {
+            Thread.currentThread().interrupt();
+          }
           LOG.error("Unable to read data: ", e);
           throw new IllegalStateException("Unable to read data: " + "{}", e);
         }
@@ -1569,6 +1572,9 @@ private RecordWriter<KeyT, ValueT> initRecordWriter(
             taskAttemptContext.getJobID().getJtIdentifier());
         return outputFormatObj.getRecordWriter(taskAttemptContext);
       } catch (InterruptedException | IOException e) {
+        if (e instanceof InterruptedException) {
+          Thread.currentThread().interrupt();
+        }
         throw new IllegalStateException("Unable to create RecordWriter object: ", e);
       }
     }

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java

@@ -610,6 +610,7 @@ private void consumerPollLoop() {
           commitCheckpointMark();
         } catch (InterruptedException e) {
           LOG.warn("{}: consumer thread is interrupted", this, e); // not expected
+          Thread.currentThread().interrupt();
           break;
         } catch (WakeupException e) {
           break;

sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponse/Call.java

@@ -297,7 +297,7 @@ public void teardown() throws UserCodeExecutionException {
         try {
           boolean ignored = executor.awaitTermination(3L, TimeUnit.SECONDS);
         } catch (InterruptedException ignored) {
-          // Ignore the interrupt during teardown.
+          Thread.currentThread().interrupt();
         }
       }
     }
@@ -354,7 +354,7 @@ private void backoffIfNeeded(BackOff backOff, Sleeper sleeper) {
           incIfPresent(sleeperCounter);
           sleeper.sleep(backOff.nextBackOffMillis());
         } catch (InterruptedException ignored) {
-          // Ignore the interrupt and try again.
+          Thread.currentThread().interrupt();
         }
       }
     }
@@ -535,6 +535,9 @@ public ResponseT call(RequestT request) throws UserCodeExecutionException {
       try {
         return future.get(timeout.getMillis(), TimeUnit.MILLISECONDS);
       } catch (TimeoutException | InterruptedException e) {
+        if (e instanceof InterruptedException) {
+          Thread.currentThread().interrupt();
+        }
         future.cancel(true);
         throw new UserCodeTimeoutException(e);
       } catch (ExecutionException e) {
@@ -585,6 +588,9 @@ private void executeAsync(Callable<Void> callable) throws UserCodeExecutionExcep
       try {
         future.get(timeout.getMillis(), TimeUnit.MILLISECONDS);
       } catch (TimeoutException | InterruptedException e) {
+        if (e instanceof InterruptedException) {
+          Thread.currentThread().interrupt();
+        }
         future.cancel(true);
         throw new UserCodeTimeoutException(e);
       } catch (ExecutionException e) {

sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponse/Repeater.java

@@ -117,7 +117,7 @@ OutputT apply(InputT input) throws UserCodeExecutionException {
         }
         latestError = Optional.of(e);
       } catch (InterruptedException ignored) {
-        // Ignore the interrupt and try again.
+        Thread.currentThread().interrupt();
       }
       incIfPresent(getBackoffCounter());
       waitFor = getBackOff().nextBackOffMillis();

sdks/java/io/sparkreceiver/3/src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java

@@ -293,6 +293,7 @@ public ProcessContinuation processElement(
         TimeUnit.SECONDS.sleep(startPollTimeoutSec);
       } catch (InterruptedException e) {
         LOG.error("SparkReceiver was interrupted before polling started", e);
+        Thread.currentThread().interrupt();
         throw new IllegalStateException("Spark Receiver was interrupted before polling started");
       }
       if (!sparkConsumer.hasRecords()) {
@@ -306,6 +307,7 @@ public ProcessContinuation processElement(
             TimeUnit.SECONDS.sleep(pullFrequencySec);
           } catch (InterruptedException e) {
             LOG.error("SparkReceiver was interrupted while waiting to poll new records", e);
+            Thread.currentThread().interrupt();
             throw new IllegalStateException(
                 "Spark Receiver was interrupted while waiting to poll new records");
           }

sdks/java/testing/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java

@@ -555,7 +555,7 @@ private void invokeBuilderForPublishOnlyPipeline(PipelineBuilder<NexmarkOptions>
       try {
         Thread.sleep(PERF_DELAY.getMillis());
       } catch (InterruptedException e) {
-        Thread.interrupted();
+        Thread.currentThread().interrupt();
         NexmarkUtils.console("Interrupted: pipeline is still running");
       }
     }

sdks/java/transform-service/launcher/src/main/java/org/apache/beam/sdk/transformservice/launcher/TransformServiceLauncher.java

@@ -275,6 +275,7 @@ private void runDockerComposeCommand(List<String> command, @Nullable File output
     try {
       this.wait(STATUS_LOGGER_WAIT_TIME);
     } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
       throw new RuntimeException(e);
     }
   }
@@ -301,7 +302,7 @@ public synchronized void waitTillUp(int timeout) throws IOException, TimeoutExce
         // We are just waiting for a local process. No need for exponential backoff.
         this.wait(1000);
       } catch (InterruptedException e) {
-        // Ignore and retry.
+        Thread.currentThread().interrupt();
       }
 
       String output = String.join(" ", java.nio.file.Files.readAllLines(Paths.get(statusFileName)));

sdks/java/transform-service/src/main/java/org/apache/beam/sdk/transformservice/ExpansionService.java

@@ -89,7 +89,7 @@ private void waitForAllServicesToBeReady() throws TimeoutException {
           try {
             Thread.sleep(duration);
           } catch (InterruptedException e) {
-            // Ignore
+            Thread.currentThread().interrupt();
           }
           duration = (long) (duration * 1.2);
         }