Repository navigation
Restore thread interrupt status in InterruptedException catch blocks - #40275
Open
TimurRakhmatullin86 wants to merge 1 commit into
Open
TimurRakhmatullin86 wants to merge 1 commit into
TimurRakhmatullin86 wants to merge 1 commit into
Conversation
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 <noreply@anthropic.com>
github-actions
Bot
added
java
io
extensions
runners
aws
gcp
hadoop
kafka
core
flink
fn-execution
jet
healthcare
spanner
prism
labels
Contributor
|
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment |
sjvanrossum
reviewed
Oct 2, 2026
Comment on lines
611
to
614
| } catch (InterruptedException e) { | ||
| LOG.warn("{}: consumer thread is interrupted", this, e); // not expected | ||
| Thread.currentThread().interrupt(); | ||
| break; |
Contributor
There was a problem hiding this comment.
Note: This block only catches interrupts that did not originate from a Kafka client (preexisting bug).
Kafka clients do not throw InterruptedException, but they may throw their own wrapper exception InterruptException instead.
No action required on this PR.
Just making sure I won't forget to address it later.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
When a thread catches
InterruptedException, the JVM clears the thread's interrupt flag. Per Java concurrency best practices (Java Concurrency in Practice, Ch. 7), the catch block must either re-throw theInterruptedExceptionor restore the flag viaThread.currentThread().interrupt(). Failing to do so silently swallows the interrupt signal, which can cause the thread (and its callers) to miss shutdown or cancellation requests — a particularly dangerous bug in a distributed data processing framework like Beam where pipeline cancellation relies on interrupt propagation.This PR fixes 33 occurrences across 27 files where
InterruptedExceptionwas caught without restoring the interrupt status.Affected modules
TransformUpgrader,BeamFnDataGrpcMultiplexerKafkaUnboundedReader), Kinesis (ShardReadersPool), BigQuery (BigQueryStorageStreamSource,DynamicDestinationsHelpers), Spanner (SpannerIO), FHIR (FhirIO), Hadoop (HadoopFormatIO), Spark Receiver (ReadFromSparkReceiverWithOffsetDoFn), Request-Response (Call,Repeater)SplittableDoFnOperator,StreamingImpulseSourcex2), Jet (JetPipelineResult), Prism (PrismExecutor), fn-execution (GrpcDataService,DockerCommand,BeamWorkerStatusGrpcService)PythonService,PythonExternalTransform), GCP (RetryHttpRequestInitializer)FinalizeBundleHandlerExpansionService,TransformServiceLauncherNexmarkLauncher— usedThread.interrupted()which clears the flag instead ofThread.currentThread().interrupt()which sets it)Categories of fixes
// ignorecomment that discarded the interrupt entirelycatch (IOException | InterruptedException e)blocks that handledInterruptedExceptionidentically toIOException, losing the interrupt semantics. Fixed by addingif (e instanceof InterruptedException) Thread.currentThread().interrupt();RuntimeException/IllegalStateExceptionand threw, but didn't restore the interrupt flag firstNexmarkLaunchercalledThread.interrupted()(a static method that reads and clears the flag) instead ofThread.currentThread().interrupt()(which sets the flag)Test plan
Thread.currentThread().interrupt()calls, which is a no-op if the thread is not being interrupted)🤖 Generated with Claude Code