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/09/01 14:49:16 UTC
[GitHub] [kafka] cadonna opened a new pull request, #12583: KAFKA-10199: Separate state updater from old restore
cadonna opened a new pull request, #12583:
URL: https://github.com/apache/kafka/pull/12583
Separates the code path for the new state updater from
the code path of the old restoration.
Ensures that with the state updater tasks are processed
before all tasks are running.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] 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] cadonna commented on a diff in pull request #12583: KAFKA-10199: Separate state updater from old restore
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12583:
URL: https://github.com/apache/kafka/pull/12583#discussion_r962717234
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -867,37 +868,47 @@ void runOnce() {
}
private void initializeAndRestorePhase() {
- // only try to initialize the assigned tasks
- // if the state is still in PARTITION_ASSIGNED after the poll call
+ final java.util.function.Consumer<Set<TopicPartition>> offsetResetter = partitions -> resetOffsets(partitions, null);
final State stateSnapshot = state;
- if (stateSnapshot == State.PARTITIONS_ASSIGNED
- || stateSnapshot == State.RUNNING && taskManager.needsInitializationOrRestoration()) {
+ if (stateUpdaterEnabled) {
+ checkStateUpdater();
+ } else {
+ // only try to initialize the assigned tasks
+ // if the state is still in PARTITION_ASSIGNED after the poll call
+ if (stateSnapshot == State.PARTITIONS_ASSIGNED
+ || stateSnapshot == State.RUNNING && taskManager.needsInitializationOrRestoration()) {
- log.debug("State is {}; initializing tasks if necessary", stateSnapshot);
+ log.debug("State is {}; initializing tasks if necessary", stateSnapshot);
- // transit to restore active is idempotent so we can call it multiple times
- changelogReader.enforceRestoreActive();
+ if (taskManager.tryToCompleteRestoration(now, offsetResetter)) {
+ log.info("Restoration took {} ms for all tasks {}", time.milliseconds() - lastPartitionAssignedMs,
+ taskManager.allTasks().keySet());
+ setState(State.RUNNING);
+ }
- if (taskManager.tryToCompleteRestoration(now, partitions -> resetOffsets(partitions, null))) {
- changelogReader.transitToUpdateStandby();
- log.info("Restoration took {} ms for all tasks {}", time.milliseconds() - lastPartitionAssignedMs,
- taskManager.allTasks().keySet());
- setState(State.RUNNING);
+ if (log.isDebugEnabled()) {
+ log.debug("Initialization call done. State is {}", state);
+ }
}
if (log.isDebugEnabled()) {
- log.debug("Initialization call done. State is {}", state);
+ log.debug("Idempotently invoking restoration logic in state {}", state);
Review Comment:
I somehow agree with you, but I did actually not change anything here. I only added the `if`-branch with the call to `checkStateUpdater()` and did not touch the content of the `else`-branch. The diff is really odd.
--
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 #12583: KAFKA-10199: Separate state updater from old restore
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12583:
URL: https://github.com/apache/kafka/pull/12583#discussion_r964144146
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -685,6 +691,7 @@ private void recycleTask(final Task task,
final Map<TaskId, RuntimeException> taskExceptions) {
Task newTask = null;
try {
+ task.suspend();
Review Comment:
Yes, I agree. That was actually the starting point of our discussion the other day. I will guard it by the presence of 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] cadonna commented on pull request #12583: KAFKA-10199: Separate state updater from old restore
Posted by GitBox <gi...@apache.org>.
cadonna commented on PR #12583:
URL: https://github.com/apache/kafka/pull/12583#issuecomment-1239316670
Build failures are unrelated:
```
Build / JDK 17 and Scala 2.13 / testLegacyAlterConfigs() – kafka.server.KRaftClusterTest
Build / JDK 11 and Scala 2.13 / shouldNotMakeStoreAvailableUntilAllStoresAvailable – org.apache.kafka.streams.integration.QueryableStateIntegrationTest
```
--
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] wcarlson5 commented on a diff in pull request #12583: KAFKA-10199: Separate state updater from old restore
Posted by GitBox <gi...@apache.org>.
wcarlson5 commented on code in PR #12583:
URL: https://github.com/apache/kafka/pull/12583#discussion_r960982217
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -396,8 +397,7 @@ public static StreamThread create(final TopologyMetadata topologyMetadata,
new Tasks(new LogContext(logPrefix)),
topologyMetadata,
adminClient,
- stateDirectory,
- maybeCreateAndStartStateUpdater(config, changelogReader, time)
+ stateDirectory, maybeCreateAndStartStateUpdater(config, changelogReader, time)
Review Comment:
nit: new line
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -867,37 +868,47 @@ void runOnce() {
}
private void initializeAndRestorePhase() {
- // only try to initialize the assigned tasks
- // if the state is still in PARTITION_ASSIGNED after the poll call
+ final java.util.function.Consumer<Set<TopicPartition>> offsetResetter = partitions -> resetOffsets(partitions, null);
final State stateSnapshot = state;
- if (stateSnapshot == State.PARTITIONS_ASSIGNED
- || stateSnapshot == State.RUNNING && taskManager.needsInitializationOrRestoration()) {
+ if (stateUpdaterEnabled) {
+ checkStateUpdater();
+ } else {
+ // only try to initialize the assigned tasks
+ // if the state is still in PARTITION_ASSIGNED after the poll call
+ if (stateSnapshot == State.PARTITIONS_ASSIGNED
+ || stateSnapshot == State.RUNNING && taskManager.needsInitializationOrRestoration()) {
- log.debug("State is {}; initializing tasks if necessary", stateSnapshot);
+ log.debug("State is {}; initializing tasks if necessary", stateSnapshot);
- // transit to restore active is idempotent so we can call it multiple times
- changelogReader.enforceRestoreActive();
+ if (taskManager.tryToCompleteRestoration(now, offsetResetter)) {
+ log.info("Restoration took {} ms for all tasks {}", time.milliseconds() - lastPartitionAssignedMs,
+ taskManager.allTasks().keySet());
+ setState(State.RUNNING);
+ }
- if (taskManager.tryToCompleteRestoration(now, partitions -> resetOffsets(partitions, null))) {
- changelogReader.transitToUpdateStandby();
- log.info("Restoration took {} ms for all tasks {}", time.milliseconds() - lastPartitionAssignedMs,
- taskManager.allTasks().keySet());
- setState(State.RUNNING);
+ if (log.isDebugEnabled()) {
+ log.debug("Initialization call done. State is {}", state);
+ }
}
if (log.isDebugEnabled()) {
- log.debug("Initialization call done. State is {}", state);
+ log.debug("Idempotently invoking restoration logic in state {}", state);
Review Comment:
`Idempotently` only for the state. I think that this log could be confusing if you read it as restore logic is Idempotently which I don't think if correct.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -770,7 +771,7 @@ void runOnce() {
long totalCommitLatency = 0L;
long totalProcessLatency = 0L;
long totalPunctuateLatency = 0L;
- if (state == State.RUNNING) {
+ if (state == State.RUNNING || stateUpdaterEnabled) {
Review Comment:
This seems risky. Can you explain what states this flag could be true?
I am worries about CREATED and ERROR states mostly?
--
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 #12583: KAFKA-10199: Separate state updater from old restore
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12583:
URL: https://github.com/apache/kafka/pull/12583#discussion_r960754245
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -280,25 +280,17 @@ public void shouldClassifyExistingTasksWithStateUpdater() {
@Test
public void shouldAddTasksToStateUpdater() {
- final StreamTask task00 = statefulTask(taskId00, taskId00Partitions)
+ final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
.withInputPartitions(taskId00Partitions)
- .inState(State.RESTORING)
- .build();
- final StandbyTask task01 = standbyTask(taskId01, taskId01Partitions)
+ .inState(State.RESTORING).build();
+ final StandbyTask task01 = standbyTask(taskId01, taskId01ChangelogPartitions)
.withInputPartitions(taskId01Partitions)
- .inState(State.RUNNING)
- .build();
- expect(changeLogReader.completedChangelogs()).andReturn(emptySet()).anyTimes();
- expect(consumer.assignment()).andReturn(emptySet()).anyTimes();
- consumer.resume(anyObject());
- expectLastCall().anyTimes();
- expect(activeTaskCreator.createTasks(anyObject(), eq(taskId00Assignment))).andStubReturn(singletonList(task00));
- expect(standbyTaskCreator.createTasks(eq(taskId01Assignment))).andStubReturn(singletonList(task01));
- replay(activeTaskCreator, standbyTaskCreator, consumer, changeLogReader);
+ .inState(State.RUNNING).build();
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ when(tasks.drainPendingTaskToInit()).thenReturn(mkSet(task00, task01));
+ taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
- taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, true);
- taskManager.handleAssignment(taskId00Assignment, taskId01Assignment);
- taskManager.tryToCompleteRestoration(time.milliseconds(), noOpResetter -> { });
+ taskManager.checkStateUpdater(time.milliseconds(), noOpResetter -> { });
Review Comment:
In this file, I did a lot of refactorings to use the utils to create task mocks and to mock the tasks registry. I had to touch those tests anyways so I thought I will bring them up-to-date.
--
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 pull request #12583: KAFKA-10199: Separate state updater from old restore
Posted by GitBox <gi...@apache.org>.
cadonna commented on PR #12583:
URL: https://github.com/apache/kafka/pull/12583#issuecomment-1234388852
Call for review: @wcarlson5
--
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 #12583: KAFKA-10199: Separate state updater from old restore
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12583:
URL: https://github.com/apache/kafka/pull/12583#discussion_r964143315
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -610,73 +610,79 @@ private StreamTask convertStandbyToActive(final StandbyTask standbyTask, final S
* @throws StreamsException if the store's change log does not contain the partition
* @return {@code true} if all tasks are fully restored
*/
- boolean tryToCompleteRestoration(final long now, final java.util.function.Consumer<Set<TopicPartition>> offsetResetter) {
- if (stateUpdater == null) {
- boolean allRunning = true;
+ boolean tryToCompleteRestoration(final long now,
Review Comment:
I pulled `changelogReader.enforceRestoreActive()` and `changelogReader.transitToUpdateStandby()` from the stream thread into `tryToCompleteRestoration()` and removed the `if`-statement that separated the code path without the state updater from the code path with the state updater. I extracted the code path with the state updater into `checkStateUpdater()`.
--
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 #12583: KAFKA-10199: Separate state updater from old restore
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on PR #12583:
URL: https://github.com/apache/kafka/pull/12583#issuecomment-1238855791
LGTM! Thanks @cadonna . No more major comments.
--
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 #12583: KAFKA-10199: Separate state updater from old restore
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12583:
URL: https://github.com/apache/kafka/pull/12583#discussion_r964458919
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -685,6 +691,7 @@ private void recycleTask(final Task task,
final Map<TaskId, RuntimeException> taskExceptions) {
Task newTask = null;
try {
+ task.suspend();
Review Comment:
I will rename the method in my next PR.
--
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 #12583: KAFKA-10199: Separate state updater from old restore
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12583:
URL: https://github.com/apache/kafka/pull/12583#discussion_r962711819
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -396,8 +397,7 @@ public static StreamThread create(final TopologyMetadata topologyMetadata,
new Tasks(new LogContext(logPrefix)),
topologyMetadata,
adminClient,
- stateDirectory,
- maybeCreateAndStartStateUpdater(config, changelogReader, time)
+ stateDirectory, maybeCreateAndStartStateUpdater(config, changelogReader, time)
Review Comment:
Thanks! I have no idea why the new line was removed here.
--
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 #12583: KAFKA-10199: Separate state updater from old restore
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12583:
URL: https://github.com/apache/kafka/pull/12583#discussion_r964144146
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -685,6 +691,7 @@ private void recycleTask(final Task task,
final Map<TaskId, RuntimeException> taskExceptions) {
Task newTask = null;
try {
+ task.suspend();
Review Comment:
Yes, I agree. That was actually the starting point of our discussion the other day. I will remove 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] cadonna commented on a diff in pull request #12583: KAFKA-10199: Separate state updater from old restore
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12583:
URL: https://github.com/apache/kafka/pull/12583#discussion_r964174732
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -685,6 +691,7 @@ private void recycleTask(final Task task,
final Map<TaskId, RuntimeException> taskExceptions) {
Task newTask = null;
try {
+ task.suspend();
Review Comment:
`recycleTask()` is only called from the code path that uses the state updater and in that code path we do not suspend active tasks on revocation. The only place we suspend tasks (active and standbys) coming from the state updater is here. In the code path without state updater, standbys are suspended before recycling but that code path is not executed with state updater enabled. During shutdown and error handling tasks are suspended in the code path that is executed with and without state updater.
So I think the answer to your questions is "no" and "no".
--
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 #12583: KAFKA-10199: Separate state updater from old restore
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12583:
URL: https://github.com/apache/kafka/pull/12583#discussion_r964139165
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -867,37 +870,47 @@ void runOnce() {
}
private void initializeAndRestorePhase() {
- // only try to initialize the assigned tasks
- // if the state is still in PARTITION_ASSIGNED after the poll call
+ final java.util.function.Consumer<Set<TopicPartition>> offsetResetter = partitions -> resetOffsets(partitions, null);
final State stateSnapshot = state;
- if (stateSnapshot == State.PARTITIONS_ASSIGNED
- || stateSnapshot == State.RUNNING && taskManager.needsInitializationOrRestoration()) {
+ if (stateUpdaterEnabled) {
+ checkStateUpdater();
+ } else {
+ // only try to initialize the assigned tasks
+ // if the state is still in PARTITION_ASSIGNED after the poll call
+ if (stateSnapshot == State.PARTITIONS_ASSIGNED
+ || stateSnapshot == State.RUNNING && taskManager.needsInitializationOrRestoration()) {
- log.debug("State is {}; initializing tasks if necessary", stateSnapshot);
+ log.debug("State is {}; initializing tasks if necessary", stateSnapshot);
- // transit to restore active is idempotent so we can call it multiple times
- changelogReader.enforceRestoreActive();
+ if (taskManager.tryToCompleteRestoration(now, offsetResetter)) {
+ log.info("Restoration took {} ms for all tasks {}", time.milliseconds() - lastPartitionAssignedMs,
+ taskManager.allTasks().keySet());
+ setState(State.RUNNING);
+ }
- if (taskManager.tryToCompleteRestoration(now, partitions -> resetOffsets(partitions, null))) {
- changelogReader.transitToUpdateStandby();
- log.info("Restoration took {} ms for all tasks {}", time.milliseconds() - lastPartitionAssignedMs,
- taskManager.allTasks().keySet());
- setState(State.RUNNING);
+ if (log.isDebugEnabled()) {
+ log.debug("Initialization call done. State is {}", state);
+ }
}
if (log.isDebugEnabled()) {
- log.debug("Initialization call done. State is {}", state);
+ log.debug("Idempotently invoking restoration logic in state {}", state);
}
+ // we can always let changelog reader try restoring in order to initialize the changelogs;
+ // if there's no active restoring or standby updating it would not try to fetch any data
+ // After KAFKA-13873, we only restore the not paused tasks.
+ changelogReader.restore(taskManager.notPausedTasks());
+ log.debug("Idempotent restore call done. Thread state has not changed.");
}
+ }
- if (log.isDebugEnabled()) {
- log.debug("Idempotently invoking restoration logic in state {}", state);
+ private void checkStateUpdater() {
+ final java.util.function.Consumer<Set<TopicPartition>> offsetResetter = partitions -> resetOffsets(partitions, null);
+ final State stateSnapshot = state;
Review Comment:
I have to admit that I copied this from `initializeAndRestorePhase()` and also there I did not 100% understand why it was done since I also thought that there is no concurrency. I then decided to be conservative and trust the original author.
--
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 #12583: KAFKA-10199: Separate state updater from old restore
Posted by GitBox <gi...@apache.org>.
cadonna merged PR #12583:
URL: https://github.com/apache/kafka/pull/12583
--
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 #12583: KAFKA-10199: Separate state updater from old restore
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on code in PR #12583:
URL: https://github.com/apache/kafka/pull/12583#discussion_r963256124
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -867,37 +870,47 @@ void runOnce() {
}
private void initializeAndRestorePhase() {
- // only try to initialize the assigned tasks
- // if the state is still in PARTITION_ASSIGNED after the poll call
+ final java.util.function.Consumer<Set<TopicPartition>> offsetResetter = partitions -> resetOffsets(partitions, null);
final State stateSnapshot = state;
- if (stateSnapshot == State.PARTITIONS_ASSIGNED
- || stateSnapshot == State.RUNNING && taskManager.needsInitializationOrRestoration()) {
+ if (stateUpdaterEnabled) {
+ checkStateUpdater();
+ } else {
+ // only try to initialize the assigned tasks
+ // if the state is still in PARTITION_ASSIGNED after the poll call
+ if (stateSnapshot == State.PARTITIONS_ASSIGNED
+ || stateSnapshot == State.RUNNING && taskManager.needsInitializationOrRestoration()) {
- log.debug("State is {}; initializing tasks if necessary", stateSnapshot);
+ log.debug("State is {}; initializing tasks if necessary", stateSnapshot);
- // transit to restore active is idempotent so we can call it multiple times
- changelogReader.enforceRestoreActive();
+ if (taskManager.tryToCompleteRestoration(now, offsetResetter)) {
+ log.info("Restoration took {} ms for all tasks {}", time.milliseconds() - lastPartitionAssignedMs,
+ taskManager.allTasks().keySet());
+ setState(State.RUNNING);
+ }
- if (taskManager.tryToCompleteRestoration(now, partitions -> resetOffsets(partitions, null))) {
- changelogReader.transitToUpdateStandby();
- log.info("Restoration took {} ms for all tasks {}", time.milliseconds() - lastPartitionAssignedMs,
- taskManager.allTasks().keySet());
- setState(State.RUNNING);
+ if (log.isDebugEnabled()) {
+ log.debug("Initialization call done. State is {}", state);
+ }
}
if (log.isDebugEnabled()) {
- log.debug("Initialization call done. State is {}", state);
+ log.debug("Idempotently invoking restoration logic in state {}", state);
}
+ // we can always let changelog reader try restoring in order to initialize the changelogs;
+ // if there's no active restoring or standby updating it would not try to fetch any data
+ // After KAFKA-13873, we only restore the not paused tasks.
+ changelogReader.restore(taskManager.notPausedTasks());
+ log.debug("Idempotent restore call done. Thread state has not changed.");
}
+ }
- if (log.isDebugEnabled()) {
- log.debug("Idempotently invoking restoration logic in state {}", state);
+ private void checkStateUpdater() {
+ final java.util.function.Consumer<Set<TopicPartition>> offsetResetter = partitions -> resetOffsets(partitions, null);
+ final State stateSnapshot = state;
Review Comment:
Why bookkeep a snapshot here? We would only change state within this thread, including the rebalance listener, so there's no concurrency to worry I think?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -610,73 +610,79 @@ private StreamTask convertStandbyToActive(final StandbyTask standbyTask, final S
* @throws StreamsException if the store's change log does not contain the partition
* @return {@code true} if all tasks are fully restored
*/
- boolean tryToCompleteRestoration(final long now, final java.util.function.Consumer<Set<TopicPartition>> offsetResetter) {
- if (stateUpdater == null) {
- boolean allRunning = true;
+ boolean tryToCompleteRestoration(final long now,
Review Comment:
Just to confirm there's no changes in `tryToCompleteRestoration` except the param right? The diffs are quite confusing... If it's the case I will skip comparing each line.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -685,6 +691,7 @@ private void recycleTask(final Task task,
final Map<TaskId, RuntimeException> taskExceptions) {
Task newTask = null;
try {
+ task.suspend();
Review Comment:
Is it assuming that suspend is idempotent here? Without state updater would we call it twice, in `handleRevocation` and `handleAssignment`?
--
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 #12583: KAFKA-10199: Separate state updater from old restore
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12583:
URL: https://github.com/apache/kafka/pull/12583#discussion_r962743402
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -770,7 +771,7 @@ void runOnce() {
long totalCommitLatency = 0L;
long totalProcessLatency = 0L;
long totalPunctuateLatency = 0L;
- if (state == State.RUNNING) {
+ if (state == State.RUNNING || stateUpdaterEnabled) {
Review Comment:
I agree with the error states and I will check for those.
A stream thread in `CREATED` would never reach this code because in `run()` the state is either set to `STARTING` or the stream thread is stopped. I think the state `STARTING` is OK, since with `STARTING` the loop should be a no-op since there would not be any tasks to process.
--
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 #12583: KAFKA-10199: Separate state updater from old restore
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on code in PR #12583:
URL: https://github.com/apache/kafka/pull/12583#discussion_r964339811
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -685,6 +691,7 @@ private void recycleTask(final Task task,
final Map<TaskId, RuntimeException> taskExceptions) {
Task newTask = null;
try {
+ task.suspend();
Review Comment:
Ah right! Thanks.
In that case could we change the func name to `recycleTaskFromStateUpdater` to be clearer?
--
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 #12583: KAFKA-10199: Separate state updater from old restore
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12583:
URL: https://github.com/apache/kafka/pull/12583#discussion_r964139165
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -867,37 +870,47 @@ void runOnce() {
}
private void initializeAndRestorePhase() {
- // only try to initialize the assigned tasks
- // if the state is still in PARTITION_ASSIGNED after the poll call
+ final java.util.function.Consumer<Set<TopicPartition>> offsetResetter = partitions -> resetOffsets(partitions, null);
final State stateSnapshot = state;
- if (stateSnapshot == State.PARTITIONS_ASSIGNED
- || stateSnapshot == State.RUNNING && taskManager.needsInitializationOrRestoration()) {
+ if (stateUpdaterEnabled) {
+ checkStateUpdater();
+ } else {
+ // only try to initialize the assigned tasks
+ // if the state is still in PARTITION_ASSIGNED after the poll call
+ if (stateSnapshot == State.PARTITIONS_ASSIGNED
+ || stateSnapshot == State.RUNNING && taskManager.needsInitializationOrRestoration()) {
- log.debug("State is {}; initializing tasks if necessary", stateSnapshot);
+ log.debug("State is {}; initializing tasks if necessary", stateSnapshot);
- // transit to restore active is idempotent so we can call it multiple times
- changelogReader.enforceRestoreActive();
+ if (taskManager.tryToCompleteRestoration(now, offsetResetter)) {
+ log.info("Restoration took {} ms for all tasks {}", time.milliseconds() - lastPartitionAssignedMs,
+ taskManager.allTasks().keySet());
+ setState(State.RUNNING);
+ }
- if (taskManager.tryToCompleteRestoration(now, partitions -> resetOffsets(partitions, null))) {
- changelogReader.transitToUpdateStandby();
- log.info("Restoration took {} ms for all tasks {}", time.milliseconds() - lastPartitionAssignedMs,
- taskManager.allTasks().keySet());
- setState(State.RUNNING);
+ if (log.isDebugEnabled()) {
+ log.debug("Initialization call done. State is {}", state);
+ }
}
if (log.isDebugEnabled()) {
- log.debug("Initialization call done. State is {}", state);
+ log.debug("Idempotently invoking restoration logic in state {}", state);
}
+ // we can always let changelog reader try restoring in order to initialize the changelogs;
+ // if there's no active restoring or standby updating it would not try to fetch any data
+ // After KAFKA-13873, we only restore the not paused tasks.
+ changelogReader.restore(taskManager.notPausedTasks());
+ log.debug("Idempotent restore call done. Thread state has not changed.");
}
+ }
- if (log.isDebugEnabled()) {
- log.debug("Idempotently invoking restoration logic in state {}", state);
+ private void checkStateUpdater() {
+ final java.util.function.Consumer<Set<TopicPartition>> offsetResetter = partitions -> resetOffsets(partitions, null);
+ final State stateSnapshot = state;
Review Comment:
I have to admit that I copied this from `initializeAndRestorePhase()` and also there I did not 100% understand why it was done there since I also thought that there is no concurrency. I then decided to be conservative and trust the original author.
--
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