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/19 13:21:40 UTC
[GitHub] [kafka] lucasbru opened a new pull request, #13017: KAFKA-14530: Check state updater more often
lucasbru opened a new pull request, #13017:
URL: https://github.com/apache/kafka/pull/13017
In the new state restoration code, the state updater needs to be checked regularly by the main thread to transfer ownership of tasks back to the main thread once the state of the task is restored. The more often we check this, the faster we can start processing the tasks.
Currently, we only check the state updater once in every loop iteration of the state updater. And while we couldn't observe this to be strictly not often enough, we can increase the number of checks easily by moving the check inside the inner processing loop. This means that once we have iterated over `numIterations` records, we can already start processing tasks that have finished restoration in the meantime.
### 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 #13017: KAFKA-14530: Check state updater more often
Posted by GitBox <gi...@apache.org>.
lucasbru commented on PR #13017:
URL: https://github.com/apache/kafka/pull/13017#issuecomment-1368785082
Test failures are unrelated.
```
Build / JDK 17 and Scala 2.13 / testSeparateOffsetsTopic – org.apache.kafka.connect.integration.ExactlyOnceSourceIntegrationTest
1m 40s
Build / JDK 8 and Scala 2.12 / testOutdatedCoordinatorAssignment() – org.apache.kafka.clients.consumer.internals.EagerConsumerCoordinatorTest
```
@cadonna could you please merge this?
--
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 #13017: KAFKA-14530: Check state updater more often
Posted by GitBox <gi...@apache.org>.
lucasbru commented on PR #13017:
URL: https://github.com/apache/kafka/pull/13017#issuecomment-1357663698
@cadonna could you please 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] lucasbru commented on a diff in pull request #13017: KAFKA-14530: Check state updater more often
Posted by GitBox <gi...@apache.org>.
lucasbru commented on code in PR #13017:
URL: https://github.com/apache/kafka/pull/13017#discussion_r1054215201
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -880,9 +885,7 @@ void runOnce() {
private void initializeAndRestorePhase() {
final java.util.function.Consumer<Set<TopicPartition>> offsetResetter = partitions -> resetOffsets(partitions, null);
final State stateSnapshot = state;
- if (stateUpdaterEnabled) {
- checkStateUpdater();
- } else {
+ if (!stateUpdaterEnabled) {
Review Comment:
Good idea. 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 merged pull request #13017: KAFKA-14530: Check state updater more often
Posted by GitBox <gi...@apache.org>.
cadonna merged PR #13017:
URL: https://github.com/apache/kafka/pull/13017
--
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 #13017: KAFKA-14530: Check state updater more often
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #13017:
URL: https://github.com/apache/kafka/pull/13017#discussion_r1053559527
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -880,9 +885,7 @@ void runOnce() {
private void initializeAndRestorePhase() {
final java.util.function.Consumer<Set<TopicPartition>> offsetResetter = partitions -> resetOffsets(partitions, null);
final State stateSnapshot = state;
- if (stateUpdaterEnabled) {
- checkStateUpdater();
- } else {
+ if (!stateUpdaterEnabled) {
Review Comment:
Shall we move this `if` to where `initializeAndRestorePhase()` is called? Then we would have both checks about whether the state updater is enabled in the same method not far from each other.
--
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