You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "cadonna (via GitHub)" <gi...@apache.org> on 2023/06/28 14:19:22 UTC
[GitHub] [kafka] cadonna opened a new pull request, #13925: KAFKA-10199: Consider tasks in state updater when computing offset sums
cadonna opened a new pull request, #13925:
URL: https://github.com/apache/kafka/pull/13925
With the state updater, the task manager needs also to look into the tasks owned by the state updater when computing the sum of offsets of the state. This sum of offsets is used by the high availability assignor to assign warm-up replicas.
If the task manager does not take into account tasks in the state updater, a warm-up replica will never report back that the state for the corresponding task has caught up. Consequently, the warm-up replica will never be dismissed and probing rebalances will never end..
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lucasbru commented on a diff in pull request #13925: KAFKA-10199: Consider tasks in state updater when computing offset sums
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #13925:
URL: https://github.com/apache/kafka/pull/13925#discussion_r1245664165
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1138,28 +1138,33 @@ public void signalResume() {
public Map<TaskId, Long> getTaskOffsetSums() {
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.
- for (final TaskId id : union(HashSet::new, lockedTaskDirectories, tasks.allTaskIds())) {
- final Task task = tasks.contains(id) ? tasks.task(id) : null;
- // Closed and uninitialized tasks don't have any offsets so we should read directly from the checkpoint
- if (task != null && task.state() != State.CREATED && task.state() != State.CLOSED) {
+ final Map<TaskId, Task> tasks = allTasks();
+ final Set<TaskId> lockedTaskDirectoriesOfNonOwnedTasksAndClosedAndCreatedTasks =
Review Comment:
Well, if there is no task directory, there is no checkpoint to process. So it's safe to not do anything in this case.
All you'd do by adding more tasks is to later skip on the check `checkPointFile.exists()`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lucasbru commented on a diff in pull request #13925: KAFKA-10199: Consider tasks in state updater when computing offset sums
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #13925:
URL: https://github.com/apache/kafka/pull/13925#discussion_r1246298268
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1138,28 +1138,33 @@ public void signalResume() {
public Map<TaskId, Long> getTaskOffsetSums() {
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.
- for (final TaskId id : union(HashSet::new, lockedTaskDirectories, tasks.allTaskIds())) {
- final Task task = tasks.contains(id) ? tasks.task(id) : null;
- // Closed and uninitialized tasks don't have any offsets so we should read directly from the checkpoint
- if (task != null && task.state() != State.CREATED && task.state() != State.CLOSED) {
+ final Map<TaskId, Task> tasks = allTasks();
+ final Set<TaskId> lockedTaskDirectoriesOfNonOwnedTasksAndClosedAndCreatedTasks =
Review Comment:
What could make the simplified code break, is that we decide to not release the lock before transitioning to the `CLOSED` state.
So yeah, being defensive here and going through all `CREATED` and `CLOSED` tasks as well to make sure that they do not have state directories that are locked but not inside `lockedTaskDirectories` sound good to me as well.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lucasbru commented on a diff in pull request #13925: KAFKA-10199: Consider tasks in state updater when computing offset sums
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #13925:
URL: https://github.com/apache/kafka/pull/13925#discussion_r1245664165
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1138,28 +1138,33 @@ public void signalResume() {
public Map<TaskId, Long> getTaskOffsetSums() {
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.
- for (final TaskId id : union(HashSet::new, lockedTaskDirectories, tasks.allTaskIds())) {
- final Task task = tasks.contains(id) ? tasks.task(id) : null;
- // Closed and uninitialized tasks don't have any offsets so we should read directly from the checkpoint
- if (task != null && task.state() != State.CREATED && task.state() != State.CLOSED) {
+ final Map<TaskId, Task> tasks = allTasks();
+ final Set<TaskId> lockedTaskDirectoriesOfNonOwnedTasksAndClosedAndCreatedTasks =
Review Comment:
Well, if there is no task directory, there is no checkpoint to process. So it's safe to not do anything in this case.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #13925: KAFKA-10199: Consider tasks in state updater when computing offset sums
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #13925:
URL: https://github.com/apache/kafka/pull/13925#discussion_r1245477150
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1138,28 +1138,33 @@ public void signalResume() {
public Map<TaskId, Long> getTaskOffsetSums() {
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.
- for (final TaskId id : union(HashSet::new, lockedTaskDirectories, tasks.allTaskIds())) {
- final Task task = tasks.contains(id) ? tasks.task(id) : null;
- // Closed and uninitialized tasks don't have any offsets so we should read directly from the checkpoint
- if (task != null && task.state() != State.CREATED && task.state() != State.CLOSED) {
+ final Map<TaskId, Task> tasks = allTasks();
+ final Set<TaskId> lockedTaskDirectoriesOfNonOwnedTasksAndClosedAndCreatedTasks =
Review Comment:
I do not think there is guarantee that `lockedTaskDirectories` contains any tasks the client owns. `lockedTaskDirectories` are just the non-empty task directories in the state directory when a rebalance starts. However, a task directory is created when a task is created, i.e., it is in state `CREATE`. A task directory is not deleted when a task is closed, i.e., in state `CLOSED`. This might be a correlation and not a thought-out invariant. At least, the original code did not rely on this since it used `union(HashSet::new, lockedTaskDirectories, tasks.allTaskIds())`.
I am also somehow reluctant to rely on such -- IMO -- brittle invariant.
As an example, in future we could decide to move the creation of the task directory to other parts of the code -- like when the task is initialized -- which would mean that there is a interval in which the task is in state `CREATED` but does not have a task directory.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on a diff in pull request #13925: KAFKA-10199: Consider tasks in state updater when computing offset sums
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #13925:
URL: https://github.com/apache/kafka/pull/13925#discussion_r1247273467
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1141,25 +1141,30 @@ public Map<TaskId, Long> getTaskOffsetSums() {
// 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.
- for (final TaskId id : union(HashSet::new, lockedTaskDirectories, tasks.allTaskIds())) {
Review Comment:
It seems with the state updated enabled, `tasks` is actually only containing "running tasks". It seems appropriate the rename this variable to `runningTasks` (can also happen in a follow up PR).
I am actually also wondering if we still need this `Tasks` container any longer to begin with? The purpose of the `Tasks` container was to simplify `TaskManager` that manages both active and standby tasks. With the state updated (from my understanding) the `TaskManager` only manages active tasks, while standby tasks will be owned by the state-updated-thread (would it still be useful for the state-updated-thread to use `Tasks` container, given that is also own active tasks as long as they are restoring?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #13925: KAFKA-10199: Consider tasks in state updater when computing offset sums
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #13925:
URL: https://github.com/apache/kafka/pull/13925#discussion_r1246336689
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1177,14 +1181,15 @@ private void tryToLockAllNonEmptyTaskDirectories() {
// current set of actually-locked tasks.
lockedTaskDirectories.clear();
+ final Map<TaskId, Task> allTasks = allTasks();
for (final TaskDirectory taskDir : stateDirectory.listNonEmptyTaskDirectories()) {
final File dir = taskDir.file();
final String namedTopology = taskDir.namedTopology();
try {
final TaskId id = parseTaskDirectoryName(dir.getName(), namedTopology);
if (stateDirectory.lock(id)) {
lockedTaskDirectories.add(id);
- if (!tasks.contains(id)) {
+ if (!allTasks.containsKey(id)) {
Review Comment:
For this debug log, we did only consider tasks owned by the stream thread.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #13925: KAFKA-10199: Consider tasks in state updater when computing offset sums
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #13925:
URL: https://github.com/apache/kafka/pull/13925#discussion_r1246365489
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1138,28 +1138,33 @@ public void signalResume() {
public Map<TaskId, Long> getTaskOffsetSums() {
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.
- for (final TaskId id : union(HashSet::new, lockedTaskDirectories, tasks.allTaskIds())) {
- final Task task = tasks.contains(id) ? tasks.task(id) : null;
- // Closed and uninitialized tasks don't have any offsets so we should read directly from the checkpoint
- if (task != null && task.state() != State.CREATED && task.state() != State.CLOSED) {
+ final Map<TaskId, Task> tasks = allTasks();
+ final Set<TaskId> lockedTaskDirectoriesOfNonOwnedTasksAndClosedAndCreatedTasks =
Review Comment:
Let's be defensive then!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #13925: KAFKA-10199: Consider tasks in state updater when computing offset sums
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #13925:
URL: https://github.com/apache/kafka/pull/13925#discussion_r1250497598
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1141,25 +1141,30 @@ public Map<TaskId, Long> getTaskOffsetSums() {
// 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.
- for (final TaskId id : union(HashSet::new, lockedTaskDirectories, tasks.allTaskIds())) {
Review Comment:
> It seems with the state updated enabled, tasks is actually only containing "running tasks". It seems appropriate the rename this variable to runningTasks (can also happen in a follow up PR).
The old code path with disabled state updater does still exist and we can disable the state updater if we encounter a major bug after releasing. So, I would postpone such renamings to the removal of the old code path.
> I am actually also wondering if we still need this Tasks container any longer to begin with?
I would keep it, because it allows to cleanly set a specific state of the task manager in unit tests. Anyways, I would wait for the upcoming thread refactoring to make such changes.
> would it still be useful for the state-updated-thread to use Tasks container, given that is also own active tasks as long as they are restoring?
I do not think so, since access by the state updater would imply that the tasks registry (aka tasks container) needs to be concurrently accessed. For this reason, we defined a invariant, that a task can only be owned either by the stream thread or by the state updater, but not both. Sharing the tasks registry between stream thread and state updater would break that invariant. If you meant to use an separate instance of the tasks registry for the state updater, that would be not useful IMO.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on pull request #13925: KAFKA-10199: Consider tasks in state updater when computing offset sums
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on PR #13925:
URL: https://github.com/apache/kafka/pull/13925#issuecomment-1618456578
Build failures are unrelated:
```
Build / JDK 17 and Scala 2.13 / org.apache.kafka.controller.QuorumControllerTest.testBalancePartitionLeaders()
Build / JDK 8 and Scala 2.12 / org.apache.kafka.controller.QuorumControllerTest.testBalancePartitionLeaders()
Build / JDK 11 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest.testOffsetTranslationBehindReplicationFlow()
Build / JDK 20 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.IdentityReplicationIntegrationTest.testReplicateTargetDefault()
Build / JDK 20 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationBaseTest.testReplicateSourceDefault()
Build / JDK 20 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationBaseTest.testReplicateSourceDefault()
Build / JDK 20 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest.testOffsetTranslationBehindReplicationFlow()
Build / JDK 20 and Scala 2.13 / org.apache.kafka.connect.integration.BlockingConnectorTest.testBlockInConnectorValidate
Build / JDK 20 and Scala 2.13 / org.apache.kafka.connect.integration.ConnectWorkerIntegrationTest.testAddAndRemoveWorker
Build / JDK 20 and Scala 2.13 / integration.kafka.server.FetchFromFollowerIntegrationTest.testRackAwareRangeAssignor()
Build / JDK 20 and Scala 2.13 / kafka.admin.TopicCommandIntegrationTest.testDescribeUnderMinIsrPartitionsMixed(String).quorum=kraft
Build / JDK 20 and Scala 2.13 / kafka.api.ConsumerBounceTest.testCloseDuringRebalance()
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna merged pull request #13925: KAFKA-10199: Consider tasks in state updater when computing offset sums
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna merged PR #13925:
URL: https://github.com/apache/kafka/pull/13925
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #13925: KAFKA-10199: Consider tasks in state updater when computing offset sums
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #13925:
URL: https://github.com/apache/kafka/pull/13925#discussion_r1246331625
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1138,28 +1138,33 @@ public void signalResume() {
public Map<TaskId, Long> getTaskOffsetSums() {
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.
- for (final TaskId id : union(HashSet::new, lockedTaskDirectories, tasks.allTaskIds())) {
- final Task task = tasks.contains(id) ? tasks.task(id) : null;
- // Closed and uninitialized tasks don't have any offsets so we should read directly from the checkpoint
- if (task != null && task.state() != State.CREATED && task.state() != State.CLOSED) {
+ final Map<TaskId, Task> tasks = allTasks();
+ final Set<TaskId> lockedTaskDirectoriesOfNonOwnedTasksAndClosedAndCreatedTasks =
Review Comment:
OK, I agree with you!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lucasbru commented on a diff in pull request #13925: KAFKA-10199: Consider tasks in state updater when computing offset sums
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #13925:
URL: https://github.com/apache/kafka/pull/13925#discussion_r1245383102
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1138,28 +1138,35 @@ public void signalResume() {
public Map<TaskId, Long> getTaskOffsetSums() {
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.
- for (final TaskId id : union(HashSet::new, lockedTaskDirectories, tasks.allTaskIds())) {
- final Task task = tasks.contains(id) ? tasks.task(id) : null;
- // Closed and uninitialized tasks don't have any offsets so we should read directly from the checkpoint
- if (task != null && task.state() != State.CREATED && task.state() != State.CLOSED) {
+ final Map<TaskId, Task> tasks = allTasks();
+ final Set<TaskId> createdAndClosedTasks = new HashSet<>();
+ 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 {}", id);
+ log.debug("Skipping to encode apparently stateless (or non-logged) offset sum for task {}",
+ task.id());
} else {
- taskOffsetSums.put(id, sumOfChangelogOffsets(id, changelogOffsets));
+ taskOffsetSums.put(task.id(), sumOfChangelogOffsets(task.id(), changelogOffsets));
}
} else {
- 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);
+ createdAndClosedTasks.add(task.id());
Review Comment:
nit: if you want to do it with fewer collections, you could inititialize `lockedTaskDirectoriesOfNonOwnedTasks` earlier, and just remove directly from that set in the `if` branch, instead of adding to `createdAndClosedTasks` in the `else` branch.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lucasbru commented on a diff in pull request #13925: KAFKA-10199: Consider tasks in state updater when computing offset sums
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #13925:
URL: https://github.com/apache/kafka/pull/13925#discussion_r1245427580
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1138,28 +1138,33 @@ public void signalResume() {
public Map<TaskId, Long> getTaskOffsetSums() {
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.
- for (final TaskId id : union(HashSet::new, lockedTaskDirectories, tasks.allTaskIds())) {
- final Task task = tasks.contains(id) ? tasks.task(id) : null;
- // Closed and uninitialized tasks don't have any offsets so we should read directly from the checkpoint
- if (task != null && task.state() != State.CREATED && task.state() != State.CLOSED) {
+ final Map<TaskId, Task> tasks = allTasks();
+ final Set<TaskId> lockedTaskDirectoriesOfNonOwnedTasksAndClosedAndCreatedTasks =
Review Comment:
Ah, I recommended this change thinking that `lockedTaskDirectories` always includes all `ClosedAndCreatedTasks` -- I think it does right? So it should be enough to assign this to `lockedTaskDirectories`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #13925: KAFKA-10199: Consider tasks in state updater when computing offset sums
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #13925:
URL: https://github.com/apache/kafka/pull/13925#discussion_r1245477150
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1138,28 +1138,33 @@ public void signalResume() {
public Map<TaskId, Long> getTaskOffsetSums() {
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.
- for (final TaskId id : union(HashSet::new, lockedTaskDirectories, tasks.allTaskIds())) {
- final Task task = tasks.contains(id) ? tasks.task(id) : null;
- // Closed and uninitialized tasks don't have any offsets so we should read directly from the checkpoint
- if (task != null && task.state() != State.CREATED && task.state() != State.CLOSED) {
+ final Map<TaskId, Task> tasks = allTasks();
+ final Set<TaskId> lockedTaskDirectoriesOfNonOwnedTasksAndClosedAndCreatedTasks =
Review Comment:
I do not think there is guarantee that `lockedTaskDirectories` contains any tasks the client owns. `lockedTaskDirectories` are just the non-empty task directories in the state directory when a rebalance starts. However, a task directory is created when a task is created, i.e., it is in state `CREATE`. A task directory is not deleted when a task is closed, i.e., in state `CLOSED`. This might be a correlation and not a thought-out invariant. At least, the original code did not rely on this since it used `union(HashSet::new, lockedTaskDirectories, tasks.allTaskIds())`.
I am also somehow reluctant to rely on such -- IMO -- brittle invariant.
The creation of the task directory can probably be moved to other parts of the code like when the task is initialized which would mean that there is a interval in which the task is in state `CREATED` but does not have a task directory.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org