KAFKA-10199: Implement adding active tasks to the state updater by cadonna · Pull Request #12128 · apache/kafka

cadonna · 2022-05-05T18:59:54Z

This PR adds the default implementation of the state updater.
The implementation only implements adding active tasks to the
state updater.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

This PR adds the default implementation of the state updater. The implementation only implements adding active tasks to the state updater.

guozhangwang

Thanks @cadonna ! I made a pass on that, and I will merge the PR as-is once we get green builds.

Please feel free to reply those comments later on and if some makes sense, we can have incorporated in the next PR.

guozhangwang · 2022-05-05T20:31:02Z

     */
    Set<TopicPartition> completedChangelogs();

+    boolean allChangelogsCompleted();


nit: also add a javadoc explanation, especially emphasize that standby task's changelogs would never be completed, and hence this function would only return true if there's no standby tasks.

guozhangwang · 2022-05-05T20:32:31Z

+
+public class DefaultStateUpdater implements StateUpdater {
+
+    private final static String BUG_ERROR_MESSAGE = "This indicates a bug. " +


Maybe we can move this static field to the StateUpdater interface?

I would not do that since it is actually an implementation detail. I would rather move it to a global location where it is accessible from multiple classes since it is a quite general message. I am not aware of a class for global constants in Streams.

Makes sense.

guozhangwang · 2022-05-05T20:43:21Z

+
+        private final ChangelogReader changelogReader;
+        private final AtomicBoolean isRunning = new AtomicBoolean(true);
+        private final java.util.function.Consumer<Set<TopicPartition>> offsetResetter;


nit: why not also import java.util.function.Consumer?

I saw that we did it in other places like #10000, but I cannot remember why..

Here it is not needed and I removed it. In other places, I think there might be a name clash between java.util.function.Consumer and org.apache.kafka.clients.consumer.Consumer.

guozhangwang · 2022-05-05T20:45:56Z

+
+    @Override
+    public void add(final Task task) {
+        if (stateUpdaterThread == null) {


I think in the end state, we may want to have more than the state-updater as the prefix since we would have multiple such threads. But this can be added later.

guozhangwang · 2022-05-05T20:47:00Z

+            }
+        }
+
+        private void performActionsOnTasks() throws InterruptedException {


I'm assuming throws InterruptedException is for other actions for the future PRs, while ADD does not throw any yet.

Same assumption elsewhere.

Those were indeed not needed.

guozhangwang · 2022-05-05T21:10:39Z

+            }
+        }
+
+        private boolean isStateless(final Task task) {


This is a meta thought: I think in the end state, we should let the stream thread to make the call and skip sending them to the tasksAndActions queue, than letting such stateless task ping-pong between the two sides, wdyt?

Yeah, that is an option. Let's see how we feel when we integrate the state updater with the rest of the code.

guozhangwang · 2022-05-05T21:13:20Z

+            return task.changelogPartitions().isEmpty() && task.isActive();
+        }
+
+        private void endRestorationIfChangelogsCompletelyRead(final Task task,


nit: rename to CompleteTaskRestorationIfPossible?

I renamed it to maybeCompleteRestoration(). It is a bit shorter 🙂.

guozhangwang · 2022-05-05T21:15:01Z

+                restoredActiveTasksLock.lock();
+                try {
+                    while (restoredActiveTasks.isEmpty() && now <= deadline) {
+                        final boolean elapsed = restoredActiveTasksCondition.await(deadline - now, TimeUnit.MILLISECONDS);


Seems elapsed not used.

I know, but IIRC I got a spotbug error if I did not use the return value.

Hmm. I think we have lots of places here return values are not used, don't know why spotbug is sensitive about this one (maybe we just disabled this rule in other classes :P ) thanks for letting me know.

guozhangwang · 2022-05-05T22:40:29Z

+                        final boolean elapsed = restoredActiveTasksCondition.await(deadline - now, TimeUnit.MILLISECONDS);
+                        now = time.milliseconds();
+                    }
+                    while (!restoredActiveTasks.isEmpty()) {


Can we just result.addAll and then restoredActiveTasks.clear() with just a if condition?

Good point! We even do not need the if condition.

guozhangwang · 2022-05-05T22:58:53Z

+            stateUpdaterThread.isRunning.set(false);
+            stateUpdaterThread.interrupt();
+            try {
+                stateUpdaterThread.join(timeout.toMillis());


Generally we use a latch to check if the thread terminates successfully in time since the join function returns void and we cannot check if it did die within the timeout. If it did not terminate within the timeout, we could consider throw an exception for it.

KAFKA-10199: Implement adding active tasks to the state updater

9d3485f

This PR adds the default implementation of the state updater. The implementation only implements adding active tasks to the state updater.

cadonna requested a review from guozhangwang May 5, 2022 19:00

guozhangwang reviewed May 5, 2022

View reviewed changes

guozhangwang merged commit ced5989 into apache:trunk May 5, 2022


		public class DefaultStateUpdater implements StateUpdater {

		private final static String BUG_ERROR_MESSAGE = "This indicates a bug. " +

Conversation

cadonna commented May 5, 2022

Committer Checklist (excluded from commit message)

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants