You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/12/20 16:29:19 UTC
[GitHub] [kafka] lucasbru opened a new pull request, #13025: KAFKA-14299: Make sure no progress is made on paused topologies
lucasbru opened a new pull request, #13025:
URL: https://github.com/apache/kafka/pull/13025
The state updater performed at least one call to `poll` on the restore
consumer before realizing that a newly added task was already
in paused state when being added. To be fully correct, no progress
at all should be made on already paused topologies.
This is tested by the existing `PauseResumeIntegrationTest`. It was not
enabled and it not pass for the state updater code path.
### Committer Checklist (excluded from commit message)
- [x] Verify design and implementation
- [x] Verify test coverage and CI build status
- [x] 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 pull request #13025: KAFKA-14299: Fix pause and resume with state updater
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #13025:
URL: https://github.com/apache/kafka/pull/13025#issuecomment-1436959926
> @lucasbru the pause/ resume integration test fails again for J11/S13, could you take a look into it?
Ah yes, that was coming from the rebase. should be fixed now
--
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] guozhangwang commented on a diff in pull request #13025: KAFKA-14299: Fix pause and resume with state updater
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on code in PR #13025:
URL: https://github.com/apache/kafka/pull/13025#discussion_r1092598270
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java:
##########
@@ -190,7 +190,7 @@ public void clearTaskTimeout() {
@Override
public boolean commitNeeded() {
- throw new UnsupportedOperationException("This task is read-only");
+ return task.commitNeeded();
Review Comment:
Why do we need to change these two functions?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -258,21 +269,27 @@ private List<TaskAndAction> getTasksAndActions() {
}
private void addTask(final Task task) {
+ final TaskId taskId = task.id();
if (isStateless(task)) {
addToRestoredTasks((StreamTask) task);
- log.info("Stateless active task " + task.id() + " was added to the restored tasks of the state updater");
+ log.info("Stateless active task " + taskId + " was added to the restored tasks of the state updater");
+ } else if (topologyMetadata.isPaused(taskId.topologyName())) {
+ pausedTasks.put(taskId, task);
Review Comment:
I'm wondering if this complexity is necessary, since we do not make strict ordering guarantees for paused topologies -- i.e. it's okay to still processing those tasks for a while after the `pause()` call is triggered. Is it really a correctness or concurrency issue?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1534,7 +1540,13 @@ Set<TaskId> standbyTaskIds() {
Map<TaskId, Task> allTasks() {
// not bothering with an unmodifiable map, since the tasks themselves are mutable, but
// if any outside code modifies the map or the tasks, it would be a severe transgression.
- return tasks.allTasksPerId();
+ if (stateUpdater != null) {
+ final Map<TaskId, Task> ret = stateUpdater.getTasks().stream().collect(Collectors.toMap(Task::id, x -> x));
+ ret.putAll(tasks.allTasksPerId());
Review Comment:
I've changed the func name slightly in another PR, so if that PR is merged we need to do a slight rebase/conflict resolution, just FYI.
--
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] guozhangwang commented on pull request #13025: KAFKA-14299: Fix pause and resume with state updater
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on PR #13025:
URL: https://github.com/apache/kafka/pull/13025#issuecomment-1438910896
LGTM! Merged to trunk.
--
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 pull request #13025: KAFKA-14299: Make sure no progress is made on paused topologies
Posted by GitBox <gi...@apache.org>.
lucasbru commented on PR #13025:
URL: https://github.com/apache/kafka/pull/13025#issuecomment-1359682705
@cadonna Could you have a look?
--
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] guozhangwang merged pull request #13025: KAFKA-14299: Fix pause and resume with state updater
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang merged PR #13025:
URL: https://github.com/apache/kafka/pull/13025
--
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 #13025: KAFKA-14299: Fix pause and resume with state updater
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #13025:
URL: https://github.com/apache/kafka/pull/13025#discussion_r1107498812
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java:
##########
@@ -190,7 +190,7 @@ public void clearTaskTimeout() {
@Override
public boolean commitNeeded() {
- throw new UnsupportedOperationException("This task is read-only");
+ return task.commitNeeded();
Review Comment:
```
1) introduce a TaskManager.allProcessingTasks, 2) depending on the updater enabled flag, let maybeCommit call either this one or allTasks
```
--
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] guozhangwang commented on pull request #13025: KAFKA-14299: Fix pause and resume with state updater
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on PR #13025:
URL: https://github.com/apache/kafka/pull/13025#issuecomment-1410815034
ack, will take a look asap.
--
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] guozhangwang commented on pull request #13025: KAFKA-14299: Fix pause and resume with state updater
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on PR #13025:
URL: https://github.com/apache/kafka/pull/13025#issuecomment-1434891336
@lucasbru the pause/ resume integration test fails again for J11/S13, could you take a look into it?
--
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 pull request #13025: KAFKA-14299: Fix pause and resume with state updater
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #13025:
URL: https://github.com/apache/kafka/pull/13025#issuecomment-1410798829
@guozhangwang also wanted to have a look (note that I didn't address Brunos comments yet)
--
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 #13025: KAFKA-14299: Fix pause and resume with state updater
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #13025:
URL: https://github.com/apache/kafka/pull/13025#discussion_r1097298086
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1534,7 +1540,13 @@ Set<TaskId> standbyTaskIds() {
Map<TaskId, Task> allTasks() {
// not bothering with an unmodifiable map, since the tasks themselves are mutable, but
// if any outside code modifies the map or the tasks, it would be a severe transgression.
- return tasks.allTasksPerId();
+ if (stateUpdater != null) {
+ final Map<TaskId, Task> ret = stateUpdater.getTasks().stream().collect(Collectors.toMap(Task::id, x -> x));
+ ret.putAll(tasks.allTasksPerId());
+ return ret;
+ } else {
+ return tasks.allTasksPerId();
+ }
Review Comment:
Done
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1534,7 +1540,13 @@ Set<TaskId> standbyTaskIds() {
Map<TaskId, Task> allTasks() {
// not bothering with an unmodifiable map, since the tasks themselves are mutable, but
// if any outside code modifies the map or the tasks, it would be a severe transgression.
- return tasks.allTasksPerId();
+ if (stateUpdater != null) {
+ final Map<TaskId, Task> ret = stateUpdater.getTasks().stream().collect(Collectors.toMap(Task::id, x -> x));
+ ret.putAll(tasks.allTasksPerId());
Review Comment:
Yes, once you merge the other one we can rebase this one.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java:
##########
@@ -190,7 +190,7 @@ public void clearTaskTimeout() {
@Override
public boolean commitNeeded() {
- throw new UnsupportedOperationException("This task is read-only");
+ return task.commitNeeded();
Review Comment:
`StreamThread.maybeCommit` uses `TaskManager.allTasks` that excluded tasks in the state updater before, but includes tasks in the state updater now. However, `commitNeeded` should be `false` for all those tasks right? I was scratching my head here a bit, because `StreamThread.maybeCommit` was processing `RESTORING` tasks in the old code path, but not in the state updater code path.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -258,21 +269,27 @@ private List<TaskAndAction> getTasksAndActions() {
}
private void addTask(final Task task) {
+ final TaskId taskId = task.id();
if (isStateless(task)) {
addToRestoredTasks((StreamTask) task);
- log.info("Stateless active task " + task.id() + " was added to the restored tasks of the state updater");
+ log.info("Stateless active task " + taskId + " was added to the restored tasks of the state updater");
+ } else if (topologyMetadata.isPaused(taskId.topologyName())) {
+ pausedTasks.put(taskId, task);
Review Comment:
w.r.t. Bruno's comment: done
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -258,21 +269,27 @@ private List<TaskAndAction> getTasksAndActions() {
}
private void addTask(final Task task) {
+ final TaskId taskId = task.id();
if (isStateless(task)) {
addToRestoredTasks((StreamTask) task);
- log.info("Stateless active task " + task.id() + " was added to the restored tasks of the state updater");
+ log.info("Stateless active task " + taskId + " was added to the restored tasks of the state updater");
+ } else if (topologyMetadata.isPaused(taskId.topologyName())) {
+ pausedTasks.put(taskId, task);
Review Comment:
The `PauseResumeIntegrationTest` is testing precisely this (that _no_ progress is made when a task is added in paused state). I also wasn't sure how important this property is for KSQL, but I suppose it is important enough that Jim wrote a unit test for it. I also think it makes sense to have the property, making a little progress on a paused task is unintuitive IMHO.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -411,6 +422,7 @@ private void checkAllUpdatingTaskStates(final long now) {
private final Condition restoredActiveTasksCondition = restoredActiveTasksLock.newCondition();
private final BlockingQueue<ExceptionAndTasks> exceptionsAndFailedTasks = new LinkedBlockingQueue<>();
private final BlockingQueue<Task> removedTasks = new LinkedBlockingQueue<>();
+ private final AtomicBoolean needsResumeCheck = new AtomicBoolean(false);
Review Comment:
Done
--
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 #13025: KAFKA-14299: Fix pause and resume with state updater
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #13025:
URL: https://github.com/apache/kafka/pull/13025#discussion_r1105612105
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java:
##########
@@ -190,7 +190,7 @@ public void clearTaskTimeout() {
@Override
public boolean commitNeeded() {
- throw new UnsupportedOperationException("This task is read-only");
+ return task.commitNeeded();
Review Comment:
It gives me a bit of a headache to forward the call to `commitNeeded()` since it changes the state of a stream task by updating the consumed offsets.
I am also wondering why Streams also commits restoring tasks.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java:
##########
@@ -200,7 +200,7 @@ public StateStore getStore(final String name) {
@Override
public Map<TopicPartition, Long> changelogOffsets() {
- throw new UnsupportedOperationException("This task is read-only");
+ return task.changelogOffsets();
Review Comment:
If I understand the code correctly, this forwarding is fine.
--
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 #13025: KAFKA-14299: Fix pause and resume with state updater
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #13025:
URL: https://github.com/apache/kafka/pull/13025#discussion_r1111909621
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java:
##########
@@ -3159,7 +3159,7 @@ private void addRecord(final MockConsumer<byte[], byte[]> mockConsumer,
}
StreamTask activeTask(final TaskManager taskManager, final TopicPartition partition) {
- final Stream<Task> standbys = taskManager.allTasks().values().stream().filter(Task::isActive);
+ final Stream<Task> standbys = taskManager.allOwnedTasks().values().stream().filter(Task::isActive);
Review Comment:
Good point, we don't
--
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 #13025: KAFKA-14299: Fix pause and resume with state updater
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #13025:
URL: https://github.com/apache/kafka/pull/13025#discussion_r1111950411
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -258,21 +269,27 @@ private List<TaskAndAction> getTasksAndActions() {
}
private void addTask(final Task task) {
+ final TaskId taskId = task.id();
if (isStateless(task)) {
addToRestoredTasks((StreamTask) task);
- log.info("Stateless active task " + task.id() + " was added to the restored tasks of the state updater");
+ log.info("Stateless active task " + taskId + " was added to the restored tasks of the state updater");
+ } else if (topologyMetadata.isPaused(taskId.topologyName())) {
+ pausedTasks.put(taskId, task);
Review Comment:
Sorry, almost missed this comment. I think this would work, but it would mean cycling through the list of all tasks in every single iteration of main state updater loop. I thought the pause code was only run after `commitMs` to reduce this overhead, so I thought it makes sense. It's different for resume, because we get a resume signal from 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] guozhangwang commented on a diff in pull request #13025: KAFKA-14299: Fix pause and resume with state updater
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on code in PR #13025:
URL: https://github.com/apache/kafka/pull/13025#discussion_r1102201605
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -258,21 +269,27 @@ private List<TaskAndAction> getTasksAndActions() {
}
private void addTask(final Task task) {
+ final TaskId taskId = task.id();
if (isStateless(task)) {
addToRestoredTasks((StreamTask) task);
- log.info("Stateless active task " + task.id() + " was added to the restored tasks of the state updater");
+ log.info("Stateless active task " + taskId + " was added to the restored tasks of the state updater");
+ } else if (topologyMetadata.isPaused(taskId.topologyName())) {
+ pausedTasks.put(taskId, task);
Review Comment:
While working on another PR I realized in `checkAllUpdatingTaskStates` we only try to pause tasks when the commit interval has elapsed, i.e. when we pause a named topology, its corresponding tasks may only be paused after a while, while when we resume, tasks are resumed immediately.
So I think we should move the `pausing` logic out of the `checkAllUpdatingTaskStates` as well like `resuming`, which would leave `checkAllUpdatingTaskStates` to just become `maybeCheckpointUpdatingTasks`. WDYT?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java:
##########
@@ -190,7 +190,7 @@ public void clearTaskTimeout() {
@Override
public boolean commitNeeded() {
- throw new UnsupportedOperationException("This task is read-only");
+ return task.commitNeeded();
Review Comment:
I think the actual culprit is like @lucasbru said, `StreamThread.maybeCommit uses TaskManager.allTasks`, while the callee `TaskManager.allTasks` is used in many different places.. For example for IQ, it was used and in that case it should return all tasks, whereas in this caller, `TaskManager.allTasks` should just return all processing tasks when updater is enabled, i.e. only the ones in `TaskRegistry`.
> I am also wondering why Streams also commits restoring tasks.
For that, the rationale is to allow the restoration progress to be recorded as well in case it was paused due to another rebalance (though it's `prepareCommit` would always return empty map, so the `committing` process for those restoring tasks would be reduced to writing checkpoints). But somehow in the middle of history we lost the logic to ever change `commitNeeded` flag for restoring tasks, so they would always not be triggered. Hence a new JIRA is proposed to fix it, and we fixed it in state updater.
I thought about fixing it in a probably better way, which involves more changes: 1) introduce a `TaskManager.allProcessingTasks`, 2) depending on the updater enabled flag, let `maybeCommit` call either this one or `allTasks`. When I thought about that, I admit I was thinking this change is a bit too much, and since tasks in updater the task should not make any side-effects and return `commitNeeded` false this maybe just okay.
But thinking about this a bit more, I think it's indeed worrisome to let `ReadyOnlyTask.commitNeeded` to have any potential side effects if we ever have bugs..
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java:
##########
@@ -190,7 +190,7 @@ public void clearTaskTimeout() {
@Override
public boolean commitNeeded() {
- throw new UnsupportedOperationException("This task is read-only");
+ return task.commitNeeded();
Review Comment:
Ack. That makes sense.
--
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 #13025: KAFKA-14299: Make sure no progress is made on paused topologies
Posted by GitBox <gi...@apache.org>.
lucasbru commented on code in PR #13025:
URL: https://github.com/apache/kafka/pull/13025#discussion_r1067891929
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -276,6 +276,11 @@ private void addTask(final Task task) {
changelogReader.transitToUpdateStandby();
}
}
+
+ // Move task to paused tasks immediately to not make any progress
+ if (topologyMetadata.isPaused(task.id().topologyName())) {
+ pauseTask(task);
Review Comment:
Done. But I also had to make more changes to make resuming reliable. I have to wake up the state updater thread when a topology is resumed, to make sure that the state updater does in fact resume the task
--
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 #13025: KAFKA-14299: Make sure no progress is made on paused topologies
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #13025:
URL: https://github.com/apache/kafka/pull/13025#discussion_r1053549663
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -276,6 +276,11 @@ private void addTask(final Task task) {
changelogReader.transitToUpdateStandby();
}
}
+
+ // Move task to paused tasks immediately to not make any progress
+ if (topologyMetadata.isPaused(task.id().topologyName())) {
+ pauseTask(task);
Review Comment:
Why not have this `if` on line 264 and adding it to the `pausedTasks` right away instead of calling `pauseTask()` which does a couple of steps like writing a checkpoint which is not needed at when the task is just added to the state updater?
--
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] guozhangwang commented on pull request #13025: KAFKA-14299: Fix pause and resume with state updater
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on PR #13025:
URL: https://github.com/apache/kafka/pull/13025#issuecomment-1421340480
I do not have further comments. @cadonna do you want to take a final look? If not I can go ahead and merge it.
--
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] guozhangwang commented on pull request #13025: KAFKA-14299: Fix pause and resume with state updater
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on PR #13025:
URL: https://github.com/apache/kafka/pull/13025#issuecomment-1430280435
@lucasbru seems I forgot to send those pending comments from last week, sorry... anyways there's only one comment that I had here: https://github.com/apache/kafka/pull/13025#discussion_r1102201605 and I think it's okay to no address in this PR actually, as we can do it in a different PR (I still need to update the logic a bit while I add more metrics recordings anyways). just FYI.
--
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 #13025: KAFKA-14299: Fix pause and resume with state updater
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #13025:
URL: https://github.com/apache/kafka/pull/13025#discussion_r1107498812
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java:
##########
@@ -190,7 +190,7 @@ public void clearTaskTimeout() {
@Override
public boolean commitNeeded() {
- throw new UnsupportedOperationException("This task is read-only");
+ return task.commitNeeded();
Review Comment:
> 1) introduce a TaskManager.allProcessingTasks, 2) depending on the updater enabled flag, let maybeCommit call either this one or allTasks
let's do :this:
I will update the PR tomorrow first thing
--
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] guozhangwang commented on a diff in pull request #13025: KAFKA-14299: Fix pause and resume with state updater
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on code in PR #13025:
URL: https://github.com/apache/kafka/pull/13025#discussion_r1109175366
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java:
##########
@@ -3159,7 +3159,7 @@ private void addRecord(final MockConsumer<byte[], byte[]> mockConsumer,
}
StreamTask activeTask(final TaskManager taskManager, final TopicPartition partition) {
- final Stream<Task> standbys = taskManager.allTasks().values().stream().filter(Task::isActive);
+ final Stream<Task> standbys = taskManager.allOwnedTasks().values().stream().filter(Task::isActive);
Review Comment:
For my understanding: why we need to change the callee here as well?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/ReadOnlyTaskTest.java:
##########
@@ -41,7 +41,6 @@ class ReadOnlyTaskTest {
add("changelogPartitions");
add("commitRequested");
add("isActive");
- add("commitNeeded");
Review Comment:
Good to know :)
--
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 #13025: KAFKA-14299: Make sure no progress is made on paused topologies
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #13025:
URL: https://github.com/apache/kafka/pull/13025#discussion_r1068096858
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -411,6 +422,7 @@ private void checkAllUpdatingTaskStates(final long now) {
private final Condition restoredActiveTasksCondition = restoredActiveTasksLock.newCondition();
private final BlockingQueue<ExceptionAndTasks> exceptionsAndFailedTasks = new LinkedBlockingQueue<>();
private final BlockingQueue<Task> removedTasks = new LinkedBlockingQueue<>();
+ private final AtomicBoolean needsResumeCheck = new AtomicBoolean(false);
Review Comment:
nit: I find the name a bit confusing. What do you think about `isTopologyResumed` or 'topologyResumed'?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -258,21 +269,27 @@ private List<TaskAndAction> getTasksAndActions() {
}
private void addTask(final Task task) {
+ final TaskId taskId = task.id();
if (isStateless(task)) {
addToRestoredTasks((StreamTask) task);
- log.info("Stateless active task " + task.id() + " was added to the restored tasks of the state updater");
+ log.info("Stateless active task " + taskId + " was added to the restored tasks of the state updater");
+ } else if (topologyMetadata.isPaused(taskId.topologyName())) {
+ pausedTasks.put(taskId, task);
Review Comment:
To be consistent, we should also verify here if the task already exists similar to the `else`-branch.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1534,7 +1540,13 @@ Set<TaskId> standbyTaskIds() {
Map<TaskId, Task> allTasks() {
// not bothering with an unmodifiable map, since the tasks themselves are mutable, but
// if any outside code modifies the map or the tasks, it would be a severe transgression.
- return tasks.allTasksPerId();
+ if (stateUpdater != null) {
+ final Map<TaskId, Task> ret = stateUpdater.getTasks().stream().collect(Collectors.toMap(Task::id, x -> x));
+ ret.putAll(tasks.allTasksPerId());
+ return ret;
+ } else {
+ return tasks.allTasksPerId();
+ }
Review Comment:
I could not find a test for this change. Could you add one?
--
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