TimurRakhmatullin86 opened a new pull request, #23587: URL: https://github.com/apache/kafka/pull/23587
## Summary When catching `InterruptedException`, Java clears the thread's interrupt flag. Per Java concurrency best practices ([Java Concurrency in Practice](https://jcip.net/) §5.4, §7.1), the interrupt status must be restored by calling `Thread.currentThread().interrupt()` unless the exception is re-thrown as `InterruptedException`. This PR fixes **24 files** across `clients`, `connect`, `streams`, `raft` and `tools` modules where the interrupt flag was silently swallowed. The bugs fall into three categories: ### 1. Swallowed interrupt flag (most severe) Catch blocks that log the exception and return without restoring the interrupt flag. This causes threads to miss shutdown signals and continue running indefinitely. **Notable case:** `TaskManager.maybeLockTasks()` catches `InterruptedException` inside a `while (!locked)` loop without restoring the flag, creating an **infinite retry loop** when the thread is interrupted during shutdown. Other examples include `Connect.awaitStop()`, `WorkerTask.awaitStop()`, `WorkerConnector.awaitShutdown()`, `MirrorMaker.awaitStop()`, `DistributedHerder.startAndStop()`, `Scheduler.execute()`/`close()`/`run()`, and `ConsumerNetworkThread.closeInternal()`. ### 2. Wrong API call: `Thread.interrupted()` vs `Thread.currentThread().interrupt()` `Thread.interrupted()` is a static method that **reads AND clears** the interrupt flag — the opposite of the intended behavior. Found in: - `TopicAdmin` (4 occurrences) - `MirrorUtils.createCompactedTopic()` - `ClientUtils.fetchEndOffsets()` ### 3. Multi-catch blocks without interrupt restoration `catch (InterruptedException | ExecutionException e)` blocks that re-throw as `RuntimeException` without restoring the interrupt flag. Split into separate catch blocks to add the `Thread.currentThread().interrupt()` call specifically for `InterruptedException`. Found in `EndToEndLatency`, `GroupOffsetsResetter` (3 occurrences), `ShareConsumerPerformance`. ### Additional fix `Shell.java`: `new IOException(ie.toString())` now passes the original `InterruptedException` as the cause: `new IOException(ie.toString(), ie)`. ## Testing These are mechanical, single-line additions of `Thread.currentThread().interrupt()` before existing handling code. The behavioral change is that the thread's interrupt flag is preserved for callers to observe, which is the correct Java threading contract. No new tests are needed as the fix restores standard JDK interrupt semantics. ## Files changed (24) | Module | File | Bug type | |--------|------|----------| | clients | `ConsumerNetworkThread` | swallowed | | clients | `RefreshingHttpsJwks` | swallowed | | clients | `Shell` | swallowed + missing cause | | connect/mirror | `MirrorCheckpointTask` | swallowed | | connect/mirror | `MirrorMaker` | swallowed | | connect/mirror | `MirrorSourceTask` | swallowed | | connect/mirror | `MirrorUtils` | wrong API | | connect/mirror | `Scheduler` (×3) | swallowed | | connect/runtime | `Connect` | swallowed | | connect/runtime | `SubmittedRecords` | swallowed | | connect/runtime | `WorkerConnector` | swallowed | | connect/runtime | `WorkerSourceTask` (×2) | swallowed | | connect/runtime | `WorkerTask` | swallowed | | connect/runtime | `DistributedHerder` | swallowed | | connect/runtime | `RetryWithToleranceOperator` | swallowed | | connect/runtime | `TopicAdmin` (×4) | wrong API | | raft | `TimingWheelExpirationService` | swallowed | | streams | `KafkaStreams` | swallowed | | streams | `ClientUtils` | wrong API | | streams | `TaskManager` | swallowed (infinite loop) | | tools | `EndToEndLatency` | multi-catch | | tools | `GroupOffsetsResetter` (×3) | multi-catch | | tools | `LeaderElectionCommand` | swallowed | | tools | `ShareConsumerPerformance` | multi-catch | 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
