cadonna commented on code in PR #12795:
URL: https://github.com/apache/kafka/pull/12795#discussion_r1018928862
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -121,21 +121,32 @@ private void runOnce() throws InterruptedException {
}
private void performActionsOnTasks() {
- tasksAndActionsLock.lock();
- try {
- for (final TaskAndAction taskAndAction : getTasksAndActions())
{
+ while (!tasksAndActions.isEmpty()) {
+ Task toInitialize = null;
+ tasksAndActionsLock.lock();
+ try {
+ final TaskAndAction taskAndAction =
tasksAndActions.remove();
final Action action = taskAndAction.getAction();
switch (action) {
case ADD:
- addTask(taskAndAction.getTask());
+ final Task task = taskAndAction.getTask();
+ final Task existingTask =
updatingTasks.putIfAbsent(task.id(), task);
+ if (existingTask != null) {
+ throw new
IllegalStateException((existingTask.isActive() ? "Active" : "Standby") + " task
" + task.id() + " already exist, " +
+ "should not try to add another " +
(task.isActive() ? "active" : "standby") + " task with the same id. " +
BUG_ERROR_MESSAGE);
+ }
+ toInitialize = task;
Review Comment:
Yeah, sorry! My bad! I meant more or less what you did, actually!
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -250,30 +261,31 @@ private void removeUpdatingAndPausedTasks() {
pausedTasks.clear();
}
- private List<TaskAndAction> getTasksAndActions() {
- final List<TaskAndAction> tasksAndActionsToProcess = new
ArrayList<>(tasksAndActions);
- tasksAndActions.clear();
- return tasksAndActionsToProcess;
- }
-
- private void addTask(final Task task) {
- if (isStateless(task)) {
- addToRestoredTasks((StreamTask) task);
- log.info("Stateless active task " + task.id() + " was added to
the restored tasks of the state updater");
- } else {
- final Task existingTask = updatingTasks.putIfAbsent(task.id(),
task);
- if (existingTask != null) {
- throw new IllegalStateException((existingTask.isActive() ?
"Active" : "Standby") + " task " + task.id() + " already exist, " +
- "should not try to add another " + (task.isActive() ?
"active" : "standby") + " task with the same id. " + BUG_ERROR_MESSAGE);
+ private void initializeTask(final Task task) {
+ boolean initializationFailed = false;
+ if (task.state() == Task.State.CREATED) {
+ try {
+ task.initializeIfNeeded();
+ } catch (final StreamsException streamsException) {
+
addToExceptionsAndFailedTasksThenRemoveFromUpdatingTasks(new
ExceptionAndTasks(Collections.singleton(task), streamsException));
+ initializationFailed = true;
}
- changelogReader.register(task.changelogPartitions(),
task.stateManager());
- if (task.isActive()) {
- log.info("Stateful active task " + task.id() + " was added
to the state updater");
- changelogReader.enforceRestoreActive();
+ }
+ if (!initializationFailed) {
Review Comment:
I see what you mean. You could just return in the `catch`-block after the
exception is added to the exception queue.
--
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]
