nicktelford commented on code in PR #20954:
URL: https://github.com/apache/kafka/pull/20954#discussion_r2845385236
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1242,35 +1239,20 @@ public void signalResume() {
* Does not include stateless or non-logged tasks.
*/
public Map<TaskId, Long> taskOffsetSums() {
- final Map<TaskId, Long> taskOffsetSums = new HashMap<>();
// Not all tasks will create directories, and there may be directories
for tasks we don't currently own,
// so we consider all tasks that are either owned or on disk. This
includes stateless tasks, which should
// just have an empty changelogOffsets map.
final Map<TaskId, Task> tasks = allTasks();
final Set<TaskId>
lockedTaskDirectoriesOfNonOwnedTasksAndClosedAndCreatedTasks =
union(HashSet::new, lockedTaskDirectories, tasks.keySet());
- for (final Task task : tasks.values()) {
- if (task.state() != State.CREATED && task.state() != State.CLOSED)
{
- final Map<TopicPartition, Long> changelogOffsets =
task.changelogOffsets();
- if (changelogOffsets.isEmpty()) {
- log.debug("Skipping to encode apparently stateless (or
non-logged) offset sum for task {}",
- task.id());
- } else {
- taskOffsetSums.put(task.id(),
sumOfChangelogOffsets(task.id(), changelogOffsets));
- }
-
lockedTaskDirectoriesOfNonOwnedTasksAndClosedAndCreatedTasks.remove(task.id());
- }
- }
- for (final TaskId id :
lockedTaskDirectoriesOfNonOwnedTasksAndClosedAndCreatedTasks) {
- final File checkpointFile = stateDirectory.checkpointFileFor(id);
- try {
- if (checkpointFile.exists()) {
- taskOffsetSums.put(id, sumOfChangelogOffsets(id, new
OffsetCheckpoint(checkpointFile).read()));
- }
- } catch (final IOException e) {
- log.warn(String.format("Exception caught while trying to read
checkpoint for task %s:", id), e);
+ final Map<TaskId, Long> taskOffsetSums =
stateDirectory.taskOffsetSums(lockedTaskDirectoriesOfNonOwnedTasksAndClosedAndCreatedTasks);
+
+ // overlay latest offsets from assigned tasks
+ for (final Task task : tasks.values()) {
+ if (task.isActive() && task.state() == State.RUNNING &&
taskOffsetSums.put(task.id(), Task.LATEST_OFFSET) == null) {
+ log.error("Could not find cached offset for assigned ACTIVE
Task {}", task.id());
Review Comment:
Hmm, I'm trying to think of situations where this might occur. This should
only happen if we've been assigned an ACTIVE task with no previous local state,
so in that case it's actually a valid corner case, the first time a task is
assigned to an instance (e.g during fresh application startup). Since there are
actually valid cases, perhaps we shouldn't log anything at all?
And looking at this in more detail, I think there's a bug here: this method
is intended to exclude stateless and non-logged tasks, but this new code would
include all tasks. I'll fix that 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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
