aliehsaeedii commented on code in PR #22832:
URL: https://github.com/apache/kafka/pull/22832#discussion_r3585447135
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java:
##########
@@ -128,6 +128,11 @@ public StateDirectoryProcessFile() {
private final StreamsConfig config;
private final Set<TaskId> tasksInLocalState = new
ConcurrentSkipListSet<>();
private final Map<TaskId, Long> taskOffsetSums = new ConcurrentHashMap<>();
+ // Wall-clock time of the last committed changelog offsets per task. Used
by the cleanup thread
+ // to decide directory obsolescence: with KIP-1035 (StateStore-managed
changelog offsets) the
+ // per-commit .checkpoint file is no longer written, so the task
directory's filesystem mtime is
+ // not refreshed during normal processing and cannot be relied upon to
reflect recent activity.
+ private final Map<TaskId, Long> taskLastCommitMs = new
ConcurrentHashMap<>();
Review Comment:
This map needs to stay in sync with `taskOffsetSums` everywhere, but
`maybeCleanEmptyNamedTopologyDirs` and `clearLocalStateForNamedTopology` strip
ids from `taskOffsetSums.keySet()` without touching this map, so entries leak
(unbounded across named-topology add/remove cycles). Mirror the `removeIf` on
both maps. BTW, NM if you follow Matthias' suggested approach.
--
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]
