ableegoldman commented on a change in pull request #10342:
URL: https://github.com/apache/kafka/pull/10342#discussion_r604366210
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java
##########
@@ -444,22 +408,30 @@ public synchronized void clean() {
* Remove the directories for any {@link TaskId}s that are no-longer
* owned by this {@link StreamThread} and aren't locked by either
* another process or another {@link StreamThread}
- * @param cleanupDelayMs only remove directories if they haven't been
modified for at least
- * this amount of time (milliseconds)
+ * @param cleanupDelayMs only remove directories if they haven't
been modified for at least
+ * this amount of time (milliseconds)
+ * @param currentThreadNames the names of all non-DEAD stream threads
so we can clean up any
+ * orphaned task directories
*/
- public synchronized void cleanRemovedTasks(final long cleanupDelayMs) {
+ public synchronized void cleanRemovedTasks(final long cleanupDelayMs,
final Set<String> currentThreadNames) {
try {
- cleanRemovedTasksCalledByCleanerThread(cleanupDelayMs);
+ cleanRemovedTasksCalledByCleanerThread(cleanupDelayMs,
currentThreadNames);
} catch (final Exception cannotHappen) {
throw new IllegalStateException("Should have swallowed
exception.", cannotHappen);
}
}
- private void cleanRemovedTasksCalledByCleanerThread(final long
cleanupDelayMs) {
+ private void cleanRemovedTasksCalledByCleanerThread(final long
cleanupDelayMs, final Set<String> currentThreadNames) {
for (final File taskDir : listNonEmptyTaskDirectories()) {
final String dirName = taskDir.getName();
final TaskId id = TaskId.parse(dirName);
- if (!locks.containsKey(id)) {
+
+ final String owningThread = lockedTasksToStreamThreadOwner.get(id);
+ if (owningThread != null &&
!currentThreadNames.contains(owningThread)) {
+ log.warn("Deleting lock for task directory {} since the thread
owning the lock is gone: {}", id, owningThread);
Review comment:
Ok actually I decided to do this in a separate PR, to keep this from
creeping scope and to unblock this while I work on adding tests for
KAFKA-10563. Reverted these changes from this PR for now
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
