From ef3d75e201705598029b7d2aa83184c900764ed1 Mon Sep 17 00:00:00 2001 From: TimurRakhmatullin Date: Thu, 24 Sep 2026 20:54:26 -0700 Subject: [PATCH] Restore thread interrupt status in InterruptedException catch blocks When a thread catches InterruptedException, the interrupt flag is cleared by the JVM. Per Java concurrency best practices, the catch block must either re-throw the InterruptedException or restore the flag via Thread.currentThread().interrupt(). Failing to do so silently swallows the interrupt signal, which can cause the thread (and its callers) to miss a shutdown or cancellation request. This change fixes 33 occurrences across 27 files in the Beam codebase where InterruptedException was caught without restoring the interrupt status. The affected modules span core SDK, I/O connectors (Kafka, Kinesis, BigQuery, Spanner, FHIR, Hadoop, Spark Receiver, Solr), runner integrations (Flink, Jet, Prism, fn-execution), extensions (Python, GCP), harness, and the transform service. Fixes fall into several categories: - Catch blocks that silently swallowed the interrupt ("// ignore") - Multi-catch blocks (e.g. IOException | InterruptedException) that handled InterruptedException the same as IOException - Catch blocks that wrapped and re-threw as RuntimeException without first restoring the interrupt flag - One case (NexmarkLauncher) that called Thread.interrupted() (which clears the flag) instead of Thread.currentThread().interrupt() Co-Authored-By: Claude Opus 4.6 --- .../wrappers/streaming/io/StreamingImpulseSource.java | 1 + .../wrappers/streaming/SplittableDoFnOperator.java | 1 + .../wrappers/streaming/io/StreamingImpulseSource.java | 1 + .../beam/runners/fnexecution/data/GrpcDataService.java | 3 +++ .../runners/fnexecution/environment/DockerCommand.java | 3 +++ .../status/BeamWorkerStatusGrpcService.java | 3 +++ .../org/apache/beam/runners/jet/JetPipelineResult.java | 3 +++ .../org/apache/beam/runners/prism/PrismExecutor.java | 4 ++-- .../beam/sdk/fn/data/BeamFnDataGrpcMultiplexer.java | 3 +++ .../beam/sdk/util/construction/TransformUpgrader.java | 2 +- .../gcp/util/RetryHttpRequestInitializer.java | 3 +++ .../sdk/extensions/python/PythonExternalTransform.java | 8 ++++++-- .../beam/sdk/extensions/python/PythonService.java | 6 ++++-- .../beam/fn/harness/control/FinalizeBundleHandler.java | 2 +- .../beam/sdk/io/aws2/kinesis/ShardReadersPool.java | 1 + .../io/gcp/bigquery/BigQueryStorageStreamSource.java | 3 +++ .../io/gcp/bigquery/DynamicDestinationsHelpers.java | 4 ++++ .../org/apache/beam/sdk/io/gcp/healthcare/FhirIO.java | 3 +++ .../org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java | 2 +- .../beam/sdk/io/hadoop/format/HadoopFormatIO.java | 6 ++++++ .../apache/beam/sdk/io/kafka/KafkaUnboundedReader.java | 1 + .../java/org/apache/beam/io/requestresponse/Call.java | 10 ++++++++-- .../org/apache/beam/io/requestresponse/Repeater.java | 2 +- .../ReadFromSparkReceiverWithOffsetDoFn.java | 2 ++ .../org/apache/beam/sdk/nexmark/NexmarkLauncher.java | 2 +- .../launcher/TransformServiceLauncher.java | 3 ++- .../beam/sdk/transformservice/ExpansionService.java | 2 +- 27 files changed, 69 insertions(+), 15 deletions(-) diff --git a/runners/flink/2.0/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/StreamingImpulseSource.java b/runners/flink/2.0/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/StreamingImpulseSource.java index 63c4cfb6b034..69ccdd3fddf7 100644 --- a/runners/flink/2.0/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/StreamingImpulseSource.java +++ b/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> ctx) { Thread.sleep(intervalMillis); } } catch (InterruptedException e) { + Thread.currentThread().interrupt(); LOG.warn("Interrupted while sleeping", e); } } diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/SplittableDoFnOperator.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/SplittableDoFnOperator.java index 3400e79b225e..edb2a4daa659 100644 --- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/SplittableDoFnOperator.java +++ b/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; } diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/StreamingImpulseSource.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/StreamingImpulseSource.java index e8b6ee786152..0a6ec9dd8ef7 100644 --- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/StreamingImpulseSource.java +++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/StreamingImpulseSource.java @@ -71,6 +71,7 @@ public void run(SourceContext> ctx) { Thread.sleep(intervalMillis); } } catch (InterruptedException e) { + Thread.currentThread().interrupt(); LOG.warn("Interrupted while sleeping", e); } } diff --git a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/data/GrpcDataService.java b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/data/GrpcDataService.java index d31f7298fbf9..4b1b306ea9c8 100644 --- a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/data/GrpcDataService.java +++ b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/data/GrpcDataService.java @@ -113,6 +113,9 @@ public StreamObserver 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); } } diff --git a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/environment/DockerCommand.java b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/environment/DockerCommand.java index ff69ee3c4171..0b9ac30f4577 100644 --- a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/environment/DockerCommand.java +++ b/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 dockerOpts, List ar ImmutableList.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 { diff --git a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/status/BeamWorkerStatusGrpcService.java b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/status/BeamWorkerStatusGrpcService.java index cb28404a475f..89959a359749 100644 --- a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/status/BeamWorkerStatusGrpcService.java +++ b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/status/BeamWorkerStatusGrpcService.java @@ -117,6 +117,9 @@ public StreamObserver workerStatus( return existingClientFuture; } } catch (IOException | InterruptedException | ExecutionException e) { + if (e instanceof InterruptedException) { + Thread.currentThread().interrupt(); + } LOG.warn("Error closing worker status client", e); } } diff --git a/runners/jet/src/main/java/org/apache/beam/runners/jet/JetPipelineResult.java b/runners/jet/src/main/java/org/apache/beam/runners/jet/JetPipelineResult.java index dccd52303987..e929a2fab710 100644 --- a/runners/jet/src/main/java/org/apache/beam/runners/jet/JetPipelineResult.java +++ b/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()); diff --git a/runners/prism/java/src/main/java/org/apache/beam/runners/prism/PrismExecutor.java b/runners/prism/java/src/main/java/org/apache/beam/runners/prism/PrismExecutor.java index 0e13b526b0fc..024e9953fbaf 100644 --- a/runners/prism/java/src/main/java/org/apache/beam/runners/prism/PrismExecutor.java +++ b/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(); } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer.java index e8afd1b7e6cd..0997f1c9dd00 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer.java +++ b/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; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/TransformUpgrader.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/TransformUpgrader.java index d3f9d9f91948..43e28c58c706 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/TransformUpgrader.java +++ b/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(); } } } diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/RetryHttpRequestInitializer.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/RetryHttpRequestInitializer.java index 13462db2e3a7..abec6cb0a9f7 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/RetryHttpRequestInitializer.java +++ b/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; } } diff --git a/sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/PythonExternalTransform.java b/sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/PythonExternalTransform.java index 3d368cb0087f..ab7a300f1465 100644 --- a/sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/PythonExternalTransform.java +++ b/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; } diff --git a/sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/PythonService.java b/sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/PythonService.java index 8658cd42cda6..678225812660 100644 --- a/sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/PythonService.java +++ b/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(); } } } diff --git a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/FinalizeBundleHandler.java b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/FinalizeBundleHandler.java index 186b0927f160..2854cf923c7d 100644 --- a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/FinalizeBundleHandler.java +++ b/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(); } diff --git a/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/ShardReadersPool.java b/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/ShardReadersPool.java index 318b7b31c4a8..a3fb681cf2db 100644 --- a/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/ShardReadersPool.java +++ b/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/ShardReadersPool.java @@ -210,6 +210,7 @@ CustomOptional nextRecord() { return CustomOptional.of(record); } catch (InterruptedException e) { LOG.warn("Interrupted while waiting for KinesisRecord from the buffer"); + Thread.currentThread().interrupt(); return CustomOptional.absent(); } } diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryStorageStreamSource.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryStorageStreamSource.java index 1bc0b2895d9a..0f39bbb75dbd 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryStorageStreamSource.java +++ b/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 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 {}.", diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/DynamicDestinationsHelpers.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/DynamicDestinationsHelpers.java index 4cd5d25202a3..e795bb3356ee 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/DynamicDestinationsHelpers.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/DynamicDestinationsHelpers.java @@ -468,6 +468,9 @@ 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) { @@ -475,6 +478,7 @@ private MatchTableDynamicDestinations( } } while (nextBackOff(Sleeper.DEFAULT, backoff)); } catch (InterruptedException e) { + Thread.currentThread().interrupt(); throw new RuntimeException(e); } return null; diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIO.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIO.java index 7b7e8952ad09..c0940c6d76f7 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIO.java +++ b/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( diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java index 1d892fb11e4a..1de72835a64b 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java +++ b/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 mutationIterable) try { sleeper.sleep(sleepTimeMsecs); } catch (InterruptedException e) { - // ignore. + Thread.currentThread().interrupt(); } } else { // Some other failure: pass up the stack. diff --git a/sdks/java/io/hadoop-format/src/main/java/org/apache/beam/sdk/io/hadoop/format/HadoopFormatIO.java b/sdks/java/io/hadoop-format/src/main/java/org/apache/beam/sdk/io/hadoop/format/HadoopFormatIO.java index 9b8bf1edb36f..92f2cc8daa82 100644 --- a/sdks/java/io/hadoop-format/src/main/java/org/apache/beam/sdk/io/hadoop/format/HadoopFormatIO.java +++ b/sdks/java/io/hadoop-format/src/main/java/org/apache/beam/sdk/io/hadoop/format/HadoopFormatIO.java @@ -979,6 +979,9 @@ public KV 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 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); } } diff --git a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java index c3aa6f5418bc..f7a50a67d42e 100644 --- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java +++ b/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; diff --git a/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponse/Call.java b/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponse/Call.java index 5b9fa4968429..8c95f89944d4 100644 --- a/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponse/Call.java +++ b/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 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) { diff --git a/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponse/Repeater.java b/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponse/Repeater.java index 0627d77fed29..bb2798d67a88 100644 --- a/sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponse/Repeater.java +++ b/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(); diff --git a/sdks/java/io/sparkreceiver/3/src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java b/sdks/java/io/sparkreceiver/3/src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java index e641e01cb3a4..6ab1a0d65a9c 100644 --- a/sdks/java/io/sparkreceiver/3/src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java +++ b/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"); } diff --git a/sdks/java/testing/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java b/sdks/java/testing/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java index 04645d859b98..93967c3eecb9 100644 --- a/sdks/java/testing/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java +++ b/sdks/java/testing/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java @@ -555,7 +555,7 @@ private void invokeBuilderForPublishOnlyPipeline(PipelineBuilder try { Thread.sleep(PERF_DELAY.getMillis()); } catch (InterruptedException e) { - Thread.interrupted(); + Thread.currentThread().interrupt(); NexmarkUtils.console("Interrupted: pipeline is still running"); } } diff --git a/sdks/java/transform-service/launcher/src/main/java/org/apache/beam/sdk/transformservice/launcher/TransformServiceLauncher.java b/sdks/java/transform-service/launcher/src/main/java/org/apache/beam/sdk/transformservice/launcher/TransformServiceLauncher.java index c26729a66a4e..522c034551ad 100644 --- a/sdks/java/transform-service/launcher/src/main/java/org/apache/beam/sdk/transformservice/launcher/TransformServiceLauncher.java +++ b/sdks/java/transform-service/launcher/src/main/java/org/apache/beam/sdk/transformservice/launcher/TransformServiceLauncher.java @@ -275,6 +275,7 @@ private void runDockerComposeCommand(List 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))); diff --git a/sdks/java/transform-service/src/main/java/org/apache/beam/sdk/transformservice/ExpansionService.java b/sdks/java/transform-service/src/main/java/org/apache/beam/sdk/transformservice/ExpansionService.java index 32a131700fef..97c73c004ddc 100644 --- a/sdks/java/transform-service/src/main/java/org/apache/beam/sdk/transformservice/ExpansionService.java +++ b/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); }