From 2b69009447c1c991c1b2d17b379697b23e0335dc Mon Sep 17 00:00:00 2001 From: Bruno Cadonna Date: Mon, 15 Nov 2021 16:17:52 +0100 Subject: [PATCH 1/3] [WIP] KAFKA-10199: Add interface for state updater --- .../processor/internals/StateUpdater.java | 50 +++++++++++++++++++ .../streams/processor/internals/Task.java | 7 +++ 2 files changed, 57 insertions(+) create mode 100644 streams/src/main/java/org/apache/kafka/streams/processor/internals/StateUpdater.java diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateUpdater.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateUpdater.java new file mode 100644 index 0000000000000..c55cb0c66864e --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateUpdater.java @@ -0,0 +1,50 @@ +package org.apache.kafka.streams.processor.internals; + +import java.time.Duration; +import java.util.List; + +public interface StateUpdater { + + /** + * Adds a task (active or standby) to the state updater. + * + * The state of the task will be updated. + * + * @param task task + */ + void add(final Task task); + + /** + * Removes a task (active and standby) from the state updater. + * + * A task is removed from the state updater irrespective of whether its state is up-to-date or not. + * + * @param task tasks to remove + */ + void remove(final Task task); + + /** + * Gets restored active tasks from state restoration/update + * + * @param timeout duration how long the calling thread should wait for restored active tasks + * + * @return list of active tasks with up-to-date states + */ + List getRestoredActiveTasks(final Duration timeout); + + /** + * Gets a list of tasks that failed during restoration. + * + * The exception that caused the failure can be retrieved by {@link Task#getException()} + * + * @return failed tasks + */ + List getFailedTasks(); + + /** + * Shuts down the state updater. + * + * @param timeout duration how long to wait until the state updater is shut down + */ + void shutdown(final Duration timeout); +} diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java index 3549ba2b18cf3..250cc739c540b 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java @@ -246,4 +246,11 @@ default boolean commitRequested() { * @return This returns the time the task started idling. If it is not idling it returns empty. */ Optional timeCurrentIdlingStarted(); + + /** + * Gets the exception that caused the failure of the task. + * + * @return exception that caused the failure of the task + */ + Optional getException(); } From 0388e5f39bf4a80ad4dbcce26ccc6f583d61d169 Mon Sep 17 00:00:00 2001 From: Bruno Cadonna Date: Wed, 23 Feb 2022 11:09:53 +0100 Subject: [PATCH 2/3] Update state updater interface --- .../processor/internals/StateUpdater.java | 43 +++++++++++++------ 1 file changed, 31 insertions(+), 12 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateUpdater.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateUpdater.java index c55cb0c66864e..8965abfbe9a41 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateUpdater.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateUpdater.java @@ -1,25 +1,38 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ package org.apache.kafka.streams.processor.internals; import java.time.Duration; import java.util.List; +import java.util.Set; public interface StateUpdater { /** * Adds a task (active or standby) to the state updater. * - * The state of the task will be updated. - * - * @param task task + * @param task task to add */ void add(final Task task); /** - * Removes a task (active and standby) from the state updater. - * - * A task is removed from the state updater irrespective of whether its state is up-to-date or not. + * Removes a task (active or standby) from the state updater. * - * @param task tasks to remove + * @param task task ro remove */ void remove(final Task task); @@ -30,16 +43,22 @@ public interface StateUpdater { * * @return list of active tasks with up-to-date states */ - List getRestoredActiveTasks(final Duration timeout); + Set getRestoredActiveTasks(final Duration timeout); /** - * Gets a list of tasks that failed during restoration. + * Gets a list of exceptions thrown during restoration. * - * The exception that caused the failure can be retrieved by {@link Task#getException()} + * @return exceptions + */ + List getExceptions(); + + + /** + * Get all tasks (active and standby) that are managed by the state updater. * - * @return failed tasks + * @return list of tasks managed by the state updater */ - List getFailedTasks(); + Set getAllTasks(); /** * Shuts down the state updater. From ff28ab121708a09d18822886fec442a823757b41 Mon Sep 17 00:00:00 2001 From: Bruno Cadonna Date: Wed, 23 Feb 2022 11:12:13 +0100 Subject: [PATCH 3/3] Remove getter for exceptions in the Task class --- .../org/apache/kafka/streams/processor/internals/Task.java | 7 ------- 1 file changed, 7 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java index 250cc739c540b..3549ba2b18cf3 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java @@ -246,11 +246,4 @@ default boolean commitRequested() { * @return This returns the time the task started idling. If it is not idling it returns empty. */ Optional timeCurrentIdlingStarted(); - - /** - * Gets the exception that caused the failure of the task. - * - * @return exception that caused the failure of the task - */ - Optional getException(); }