You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org> on 2026/02/18 00:35:02 UTC
[PR] MINOR: Implement TODOs to use StreamTask and StandbyTask whenever it's possible. [kafka]
Nikita-Shupletsov opened a new pull request, #21502:
URL: https://github.com/apache/kafka/pull/21502
(no comment)
--
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
Re: [PR] MINOR: Implement TODOs to use StreamTask and StandbyTask whenever it's possible. [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21502:
URL: https://github.com/apache/kafka/pull/21502#discussion_r2824151874
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreator.java:
##########
@@ -131,10 +131,9 @@ boolean isClosed() {
return isClosed;
}
- // TODO: convert to StreamTask when we remove TaskManager#StateMachineTask with mocks
- public Collection<Task> createTasks(final Consumer<byte[], byte[]> consumer,
+ public Collection<StreamTask> createTasks(final Consumer<byte[], byte[]> consumer,
final Map<TaskId, Set<TopicPartition>> tasksToBeCreated) {
Review Comment:
nit: fix indention
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1529,20 +1527,20 @@ void closeAndCleanUpTasks(final Collection<Task> activeTasks, final Collection<T
}
// Returns the set of active tasks that must be closed dirty
- private Collection<Task> tryCloseCleanActiveTasks(final Collection<Task> activeTasksToClose,
+ private Collection<StreamTask> tryCloseCleanActiveTasks(final Collection<StreamTask> activeTasksToClose,
final boolean clean,
final AtomicReference<RuntimeException> firstException) {
Review Comment:
nit: fix indention
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1622,16 +1620,16 @@ private Collection<Task> tryCloseCleanActiveTasks(final Collection<Task> activeT
}
// Returns the set of standby tasks that must be closed dirty
- private Collection<Task> tryCloseCleanStandbyTasks(final Collection<Task> standbyTasksToClose,
+ private Collection<StandbyTask> tryCloseCleanStandbyTasks(final Collection<StandbyTask> standbyTasksToClose,
final boolean clean,
final AtomicReference<RuntimeException> firstException) {
Review Comment:
nit: fix intention
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java:
##########
@@ -170,42 +168,55 @@ public boolean hasPendingTasksToClose() {
}
@Override
- public void addActiveTasks(final Collection<Task> newTasks) {
+ public void addActiveTasks(final Collection<StreamTask> newTasks) {
if (!newTasks.isEmpty()) {
- for (final Task activeTask : newTasks) {
- addTask(activeTask);
+ for (final StreamTask activeTask : newTasks) {
+ addActiveTask(activeTask);
}
}
}
@Override
- public void addStandbyTasks(final Collection<Task> newTasks) {
+ public void addStandbyTasks(final Collection<StandbyTask> newTasks) {
if (!newTasks.isEmpty()) {
- for (final Task standbyTask : newTasks) {
- addTask(standbyTask);
+ for (final StandbyTask standbyTask : newTasks) {
+ addStandbyTask(standbyTask);
}
}
}
@Override
- public synchronized void addTask(final Task task) {
- final TaskId taskId = task.id();
- if (activeTasksPerId.containsKey(taskId)) {
- throw new IllegalStateException("Attempted to create an active task that we already own: " + taskId);
+ public void addTask(final Task task) {
+ if (task.isActive()) {
+ addActiveTask((StreamTask) task);
+ } else {
+ addStandbyTask((StandbyTask) task);
}
+ }
- if (standbyTasksPerId.containsKey(taskId)) {
- throw new IllegalStateException("Attempted to create an active task while we already own its standby: " + taskId);
Review Comment:
Both original error messages say `Attempted to create an active task` -- so wondering why the original method is called `addTask`? Or was the original message not fully correct to begin with? In the end, we cannot have active and standby (with task task-id) at the same time, but it goes both ways. So it would also be a valid error message to say `Attempted to create an standby task...`?
Also, by extracting `checkTaskDoesNotExist()`, we actually get a single more general error message -- given that `checkTaskDoesNotExist()` is very short, might be better to keep the code inlined, and use two different error messages.
--
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
Re: [PR] MINOR: Implement TODOs to use StreamTask and StandbyTask whenever it's possible. [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #21502:
URL: https://github.com/apache/kafka/pull/21502#discussion_r2824356089
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java:
##########
@@ -170,42 +168,55 @@ public boolean hasPendingTasksToClose() {
}
@Override
- public void addActiveTasks(final Collection<Task> newTasks) {
+ public void addActiveTasks(final Collection<StreamTask> newTasks) {
if (!newTasks.isEmpty()) {
- for (final Task activeTask : newTasks) {
- addTask(activeTask);
+ for (final StreamTask activeTask : newTasks) {
+ addActiveTask(activeTask);
}
}
}
@Override
- public void addStandbyTasks(final Collection<Task> newTasks) {
+ public void addStandbyTasks(final Collection<StandbyTask> newTasks) {
if (!newTasks.isEmpty()) {
- for (final Task standbyTask : newTasks) {
- addTask(standbyTask);
+ for (final StandbyTask standbyTask : newTasks) {
+ addStandbyTask(standbyTask);
}
}
}
@Override
- public synchronized void addTask(final Task task) {
- final TaskId taskId = task.id();
- if (activeTasksPerId.containsKey(taskId)) {
- throw new IllegalStateException("Attempted to create an active task that we already own: " + taskId);
+ public void addTask(final Task task) {
+ if (task.isActive()) {
+ addActiveTask((StreamTask) task);
+ } else {
+ addStandbyTask((StandbyTask) task);
}
+ }
- if (standbyTasksPerId.containsKey(taskId)) {
- throw new IllegalStateException("Attempted to create an active task while we already own its standby: " + taskId);
Review Comment:
yeah, there was addActiveTask with this errors, then it became a more generic addTask method, but the errors stayed the same
sounds good. will do
--
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
Re: [PR] MINOR: Replace generic Task type with StreamTask and StandbyTask [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #21502:
URL: https://github.com/apache/kafka/pull/21502
--
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
Re: [PR] MINOR: Implement TODOs to use StreamTask and StandbyTask whenever it's possible. [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #21502:
URL: https://github.com/apache/kafka/pull/21502#discussion_r2824356089
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java:
##########
@@ -170,42 +168,55 @@ public boolean hasPendingTasksToClose() {
}
@Override
- public void addActiveTasks(final Collection<Task> newTasks) {
+ public void addActiveTasks(final Collection<StreamTask> newTasks) {
if (!newTasks.isEmpty()) {
- for (final Task activeTask : newTasks) {
- addTask(activeTask);
+ for (final StreamTask activeTask : newTasks) {
+ addActiveTask(activeTask);
}
}
}
@Override
- public void addStandbyTasks(final Collection<Task> newTasks) {
+ public void addStandbyTasks(final Collection<StandbyTask> newTasks) {
if (!newTasks.isEmpty()) {
- for (final Task standbyTask : newTasks) {
- addTask(standbyTask);
+ for (final StandbyTask standbyTask : newTasks) {
+ addStandbyTask(standbyTask);
}
}
}
@Override
- public synchronized void addTask(final Task task) {
- final TaskId taskId = task.id();
- if (activeTasksPerId.containsKey(taskId)) {
- throw new IllegalStateException("Attempted to create an active task that we already own: " + taskId);
+ public void addTask(final Task task) {
+ if (task.isActive()) {
+ addActiveTask((StreamTask) task);
+ } else {
+ addStandbyTask((StandbyTask) task);
}
+ }
- if (standbyTasksPerId.containsKey(taskId)) {
- throw new IllegalStateException("Attempted to create an active task while we already own its standby: " + taskId);
Review Comment:
yeah, there was addActiveTask with these errors, then it became a more generic addTask method, but the errors stayed the same
sounds good. will do
--
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
Re: [PR] MINOR: Implement TODOs to use StreamTask and StandbyTask whenever it's possible. [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #21502:
URL: https://github.com/apache/kafka/pull/21502#discussion_r2825102055
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
Review Comment:
not 100% if stateUpdater.tasks() needs to return ReadOnlyTask. or maybe we should have a separate method to get all active tasks to call it here, as we do for standby tasks
--
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