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); }