guwensheng created KAFKA-21070:
----------------------------------
Summary: KafkaStreams StateUpdater Task Handle Leak on Unclean
Shutdown
Key: KAFKA-21070
URL: https://issues.apache.org/jira/browse/KAFKA-21070
Project: Kafka
Issue Type: Bug
Reporter: guwensheng
# KafkaStreams StateUpdater Task Handle Leak on Unclean Shutdown
## Summary
KafkaStreams `TaskManager.shutdownStateUpdater()` fails to close tasks under
certain conditions during unclean shutdown, leading to permanent RocksDB file
descriptor and POSIX advisory lock leaks. This occurs through two distinct code
paths in the `DefaultStateUpdater` and `TaskManager` classes.
**Affected Version**: Apache Kafka 3.9.2 (streams module)
**Severity**: High — leaked RocksDB file handles accumulate across pipeline
restarts, eventually causing `lock hold by current process` errors and
requiring a full process restart.
---
## Bug 1: `shutdownStateUpdater()` has no exception guard — interrupt skips all
task closing logic
### Location
`streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java`,
lines 1520-1543
### Description
`shutdownStateUpdater()` calls `addToTasksToClose()` which internally calls
`waitForFuture()` → `future.get()` to wait for the StateUpdater thread to
process REMOVE actions. If the calling thread's interrupt flag is set (which
can happen via `ClosedByInterruptException` from NIO operations — this
exception does **not** clear the interrupt flag, unlike
`InterruptedException`), `future.get()` immediately throws
`InterruptedException`.
`waitForFuture()` catches this and throws `IllegalStateException`, which
propagates uncaught through `shutdownStateUpdater()`. As a result:
- `stateUpdater.shutdown()` is **never called** — the StateUpdater thread keeps
running
- `closeTaskDirty()` for all tasks is **never called** — tasks' RocksDB handles
are never released
- `drainExceptionsAndFailedTasks()` is **never called**
### Problematic Code
```java
// TaskManager.java:1520-1543
private void shutdownStateUpdater() {
if (stateUpdater != null) {
final Map<TaskId, CompletableFuture<StateUpdater.RemovedTaskResult>>
futures = new LinkedHashMap<>();
for (final Task task : stateUpdater.getTasks()) {
final CompletableFuture<StateUpdater.RemovedTaskResult> future =
stateUpdater.remove(task.id());
futures.put(task.id(), future);
}
final Set<Task> tasksToCloseClean = new HashSet<>();
final Set<Task> tasksToCloseDirty = new HashSet<>();
addToTasksToClose(futures, tasksToCloseClean, tasksToCloseDirty); // ←
throws if interrupted
stateUpdater.shutdown(Duration.ofMillis(Long.MAX_VALUE)); //
← NEVER REACHED
for (final Task task : tasksToCloseClean) { tasks.addTask(task); }
for (final Task task : tasksToCloseDirty) {
closeTaskDirty(task, false); //
← NEVER REACHED
}
for (final StateUpdater.ExceptionAndTask exceptionAndTask :
stateUpdater.drainExceptionsAndFailedTasks()) {
closeTaskDirty(exceptionAndTask.task(), false); // ←
NEVER REACHED
}
}
}
```
```java
// TaskManager.java:700-719
private StateUpdater.RemovedTaskResult waitForFuture(final TaskId taskId,
final
CompletableFuture<StateUpdater.RemovedTaskResult> future) {
try {
removedTaskResult = future.get(); // ← throws InterruptedException if
flag set
// ...
} catch (final InterruptedException shouldNotHappen) {
Thread.currentThread().interrupt();
log.error(INTERRUPTED_ERROR_MESSAGE, shouldNotHappen);
throw new IllegalStateException(INTERRUPTED_ERROR_MESSAGE,
shouldNotHappen); // ← propagates, no guard in caller
}
}
```
### Why the Interrupt Flag Persists
When a thread is interrupted during NIO channel operations (e.g.,
`EPoll.wait()` → `Selector.select()`), the JVM throws
`ClosedByInterruptException` (a subclass of
`AsynchronousCloseException`/`IOException`). **Unlike `InterruptedException`,
`ClosedByInterruptException` does NOT clear the thread's interrupt flag.** This
means any subsequent blocking call that checks the interrupt flag (such as
`Future.get()`, `LockSupport.park()`, `Object.wait()`) will immediately throw.
The `StreamThread.completeShutdown()` method is called from the `finally` block
of `StreamThread.run()`:
```java
// StreamThread.java:726-730
} catch (final Throwable e) {
streamsUncaughtExceptionHandler.accept(e, false);
} finally {
completeShutdown(cleanRun); // ← called while interrupt flag may still be
set
}
```
`completeShutdown()` calls `taskManager.shutdown(false)` →
`shutdownStateUpdater()` → `waitForFuture()` → `future.get()`, which
immediately fails due to the persistent interrupt flag.
### Suggested Fix
Wrap the `addToTasksToClose()` call in a try-catch, and ensure
`stateUpdater.shutdown()` and `closeTaskDirty()` are always called:
```java
private void shutdownStateUpdater() {
if (stateUpdater != null) {
final Map<TaskId, CompletableFuture<StateUpdater.RemovedTaskResult>>
futures = new LinkedHashMap<>();
for (final Task task : stateUpdater.getTasks()) {
final CompletableFuture<StateUpdater.RemovedTaskResult> future =
stateUpdater.remove(task.id());
futures.put(task.id(), future);
}
final Set<Task> tasksToCloseClean = new HashSet<>();
final Set<Task> tasksToCloseDirty = new HashSet<>();
try {
addToTasksToClose(futures, tasksToCloseClean, tasksToCloseDirty);
} catch (final RuntimeException e) {
log.warn("Failed to wait for state updater futures, will close all
tasks dirty", e);
// Add all tasks from stateUpdater to dirty close set
for (final Task task : stateUpdater.getTasks()) {
tasksToCloseDirty.add(task);
}
}
stateUpdater.shutdown(Duration.ofMillis(Long.MAX_VALUE));
for (final Task task : tasksToCloseClean) { tasks.addTask(task); }
for (final Task task : tasksToCloseDirty) { closeTaskDirty(task,
false); }
for (final StateUpdater.ExceptionAndTask exceptionAndTask :
stateUpdater.drainExceptionsAndFailedTasks()) {
closeTaskDirty(exceptionAndTask.task(), false);
}
}
}
```
---
## Bug 2: `StateUpdaterThread.run()` finally block discards tasks without
closing them
### Location
`streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java`,
lines 144-160
### Description
When the StateUpdaterThread stops (either normally or due to an exception), its
`finally` block calls `clearInputQueue()` and `clearUpdatingAndPausedTasks()`.
Both methods simply clear their respective collections **without closing any
tasks**. Additionally, the `restoredActiveTasks` and `removedTasks` queues are
**not cleared at all** in the finally block.
This means any task that was in any of these queues when the thread stops will
have its RocksDB store left open, leaking file descriptors and POSIX locks.
### Problematic Code
```java
// DefaultStateUpdater.java:144-160
@Override
public void run() {
log.info("State updater thread started");
try {
while (isRunning.get()) {
runOnce();
}
} catch (final RuntimeException anyOtherException) {
handleRuntimeException(anyOtherException);
} finally {
clearInputQueue(); // discards pending ADD/REMOVE actions
clearUpdatingAndPausedTasks(); // clears maps WITHOUT closing tasks
// restoredActiveTasks: NOT CLEARED AT ALL
// removedTasks: NOT CLEARED AT ALL
updaterMetrics.clear();
shutdownGate.countDown();
log.info("State updater thread stopped");
}
}
private void clearUpdatingAndPausedTasks() {
updatingTasks.clear(); // ← Task objects are GC'd, but RocksDB native
handles/POSIX locks are NOT released
pausedTasks.clear();
changelogReader.clear();
}
private void clearInputQueue() {
tasksAndActionsLock.lock();
try {
tasksAndActions.clear(); // ← Pending REMOVE actions are lost;
futures never complete
} finally {
tasksAndActionsLock.unlock();
}
}
```
### Task Queues Not Cleaned in finally
The `getStreamOfTasks()` method (line 1027-1048) shows all the queues that hold
Task references:
```java
private Stream<Task> getStreamOfTasks() {
return Stream.concat(
getStreamOfNonPausedTasks(),
getPausedTasks().stream()
);
}
private Stream<Task> getStreamOfNonPausedTasks() {
return Stream.concat(
tasksAndActions.stream()
.filter(taskAndAction -> taskAndAction.action() == Action.ADD)
.map(TaskAndAction::task),
Stream.concat(
getUpdatingTasks().stream(),
Stream.concat(
restoredActiveTasks.stream(), // ← NOT cleared in finally
Stream.concat(
exceptionsAndFailedTasks.stream().map(ExceptionAndTask::task), // ← NOT cleared
removedTasks.stream())))); // ← NOT cleared
}
```
### Suggested Fix
The finally block should close all tasks in all queues before clearing them:
```java
@Override
public void run() {
log.info("State updater thread started");
try {
while (isRunning.get()) {
runOnce();
}
} catch (final RuntimeException anyOtherException) {
handleRuntimeException(anyOtherException);
} finally {
clearInputQueue();
closeAndClearAllTasks(); // ← new method: close all tasks before
clearing
updaterMetrics.clear();
shutdownGate.countDown();
log.info("State updater thread stopped");
}
}
private void closeAndClearAllTasks() {
// Close all tasks in all queues
final List<Task> allTasks = new ArrayList<>();
allTasks.addAll(getUpdatingTasks());
allTasks.addAll(getPausedTasks());
restoredActiveTasksLock.lock();
try {
allTasks.addAll(restoredActiveTasks);
} finally {
restoredActiveTasksLock.unlock();
}
exceptionsAndFailedTasksLock.lock();
try {
exceptionsAndFailedTasks.stream()
.map(ExceptionAndTask::task)
.forEach(allTasks::add);
} finally {
exceptionsAndFailedTasksLock.unlock();
}
allTasks.addAll(removedTasks);
for (final Task task : allTasks) {
try {
task.closeDirty(); // or appropriate close method
} catch (final RuntimeException e) {
log.warn("Failed to close task " + task.id() + " during state
updater shutdown", e);
}
}
// Now safe to clear
updatingTasks.clear();
pausedTasks.clear();
restoredActiveTasks.clear();
removedTasks.clear();
exceptionsAndFailedTasks.clear();
changelogReader.clear();
}
```
---
## Reproduction Scenario
### Environment
- Apache Kafka 3.9.2 (streams module)
- KafkaStreams with StateUpdater enabled (processing threads)
- Stateful tasks using RocksDB state stores
- 3 KafkaStreams instances in a consumer group
### Trigger Sequence
1. A StreamThread encounters an exception (e.g., `TimeoutException` on
repartition topic position query) and is replaced by a new StreamThread
2. The new StreamThread creates tasks and adds them to the StateUpdater
(RocksDB stores are opened, file descriptors acquired)
3. A missed rebalance causes some tasks to be closed, but tasks in the
StateUpdater (especially standby tasks) may not be closed
4. `KafkaStreams.close()` is called with a short timeout, which starts a
shutdown thread that calls `StreamThread.shutdown()` and waits for the thread
to join
5. If the close times out, the calling application may interrupt the
StreamThread (this is a legitimate use case — the `KafkaStreams` API doesn't
prevent this)
6. The interrupt hits the StreamThread during an NIO operation (e.g.,
`KafkaConsumer.poll()` → `EPoll.wait()`), causing `ClosedByInterruptException`
with the interrupt flag preserved
7. `StreamThread.run()`'s `finally` block calls `completeShutdown(false)`
(unclean)
8. `completeShutdown()` → `taskManager.shutdown(false)` →
`shutdownStateUpdater()` → `waitForFuture()` → `future.get()` immediately
throws `InterruptedException` due to the persistent interrupt flag
9. `stateUpdater.shutdown()` and `closeTaskDirty()` are never called — tasks in
the StateUpdater are never closed
10. RocksDB file descriptors and POSIX advisory locks are permanently leaked
### Observed Evidence
**Log output (abbreviated):**
```
Shutting down unclean
Thread got interrupted. This indicates a bug.
java.lang.IllegalStateException: Thread got interrupted. This indicates a bug.
at TaskManager.waitForFuture(TaskManager.java:718)
at TaskManager.addToTasksToClose(TaskManager.java:678)
at TaskManager.shutdownStateUpdater(TaskManager.java:1529)
at TaskManager.shutdown(TaskManager.java:1478)
at StreamThread.completeShutdown(StreamThread.java:1524)
at StreamThread.run(StreamThread.java:729)
Failed to close task manager due to the following error:
(same stack trace)
Some task directories still locked while closing state, this indicates unclean
shutdown: {1_0=..., 4_0=..., 1_2=...}
State transition from PENDING_SHUTDOWN to DEAD
```
**Process file descriptors:**
```
$ ls -la /proc/<PID>/fd | grep deleted | grep LOCK | wc -l
41 # (deleted) LOCK files still held with POSIX advisory write locks
```
The leaked file descriptors show as `(deleted)` because the state directories
are later deleted, but the POSIX locks on the file descriptors remain active,
preventing new KafkaStreams instances from opening the same RocksDB stores
(`lock hold by current process`).
---
## Impact
1. **RocksDB file descriptor leak**: Each unclean shutdown leaks all tasks
currently in the StateUpdater (updating, paused, restored, removed, and failed
queues). File descriptors are never reclaimed.
2. **POSIX advisory lock leak**: The leaked file descriptors hold POSIX
advisory write locks on RocksDB LOCK files. These locks persist until the
process exits.
3. **Pipeline restart failures**: New KafkaStreams instances cannot open the
same state stores because the POSIX locks are still held, resulting in
`RocksDBException: lock hold by current process`.
4. **Accumulation over time**: In production environments with frequent
pipeline restarts (e.g., due to rebalances or configuration changes), the
leaked handles accumulate and eventually cause the process to run out of file
descriptors.
---
## Comparison: Both Bugs Must Be Fixed
| Bug | Path | Root Cause |
|-----|------|------------|
| Bug 1 | `shutdownStateUpdater()` → `waitForFuture()` → `future.get()` | No
try-catch guard; interrupt flag from NIO `ClosedByInterruptException` causes
immediate failure, skipping all task closing logic |
| Bug 2 | `StateUpdaterThread.run()` finally block |
`clearUpdatingAndPausedTasks()` discards tasks without closing;
`restoredActiveTasks` and `removedTasks` not cleared at all |
**Bug 1** is the primary trigger in the interrupt scenario — even if Bug 2 were
fixed, the tasks would still leak because `stateUpdater.shutdown()` and
`closeTaskDirty()` are never called.
**Bug 2** is a defense-in-depth issue — even in a normal (non-interrupt)
shutdown where `stateUpdater.shutdown()` is called, if the StateUpdater
thread's `runOnce()` throws an unexpected `RuntimeException`, the finally block
would still discard tasks without closing them.
Both bugs need to be fixed to ensure RocksDB handles are always properly
released.
## Environment Details
- **Kafka Version**: 3.9.2
- **JDK**: OpenJDK 21
- **OS**: Linux (SUSE/Red Hat based)
- **State Store**: RocksDB (rocksdbjni 7.9.2)
- **Stream Configuration**: Single StreamThread per KafkaStreams instance,
standby tasks enabled, processing threads enabled
---
## Files Involved
| File | Key Lines | Role |
|------|-----------|------|
| `TaskManager.java` | 700-719 (`waitForFuture`), 1520-1543
(`shutdownStateUpdater`), 1477-1518 (`shutdown`) | No interrupt guard in
shutdown path |
| `DefaultStateUpdater.java` | 144-160 (`run` finally), 458-462
(`clearUpdatingAndPausedTasks`), 162-169 (`clearInputQueue`), 1027-1048
(`getStreamOfTasks`) | Tasks discarded without closing in finally block |
| `StreamThread.java` | 1513-1561 (`completeShutdown`), 720-731 (`run` finally)
| Calls `taskManager.shutdown()` while interrupt flag may be set |
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
