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 2020/06/25 01:37:29 UTC
[GitHub] [kafka] ableegoldman opened a new pull request #8926: KAFKA-10166: always invoke `postCommit` before closing a task
ableegoldman opened a new pull request #8926:
URL: https://github.com/apache/kafka/pull/8926
This should address at least some of the excessive TaskCorruptedExceptions we've been seeing lately. Basically, at the moment we only commit tasks if `commitNeeded` is true -- this seems true by definition. But the problem is we do some essential cleanup in `postCommit` that should always be done before a task is closed:
1. clear the PartitionGroup
2. write the checkpoint
2 is actually fine to skip when `commitNeeded = false` with ALOS, as we will have already written a checkpoint during the last commit. But for EOS, we _only_ write the checkpoint before a close -- so even if there is no new pending data since the last commit, we have to write the current offsets. If we don't, the task will be assumed dirty and we will run into our friend the TaskCorruptedException during (re)initialization.
To fix this, we should just always call `prepareCommit` and `postCommit` at the TaskManager level. Within the task, it can decide whether or not to actually do something in those methods based on `commitNeeded`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on pull request #8926: KAFKA-10166: always write checkpoint before closing an (initialized) task
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #8926:
URL: https://github.com/apache/kafka/pull/8926#issuecomment-650426719
Cherry-picked to 2.6 since it is a blocker, cc @rhauch
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on pull request #8926: KAFKA-10166: always write checkpoint before closing an (initialized) task
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #8926:
URL: https://github.com/apache/kafka/pull/8926#issuecomment-649936958
Ok to test
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on pull request #8926: KAFKA-10166: always write checkpoint before closing an (initialized) task
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on pull request #8926:
URL: https://github.com/apache/kafka/pull/8926#issuecomment-650382721
Java 8 and 14 builds passed, Java 11 build failed with...zero failures?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on pull request #8926: KAFKA-10166: always write checkpoint before closing an (initialized) task
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #8926:
URL: https://github.com/apache/kafka/pull/8926#issuecomment-650261830
> Seems like all the Topology testDriver tests failed, but I got a green build running locally. Do they not run with `./gradlew streams:test`?
They are included in streams:test. Maybe try to rebase the branch and see if there's any missing committs?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #8926: KAFKA-10166: always invoke `postCommit` before closing a task
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #8926:
URL: https://github.com/apache/kafka/pull/8926#discussion_r445262369
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
##########
@@ -242,18 +242,16 @@ public void handleAssignment(final Map<TaskId, Set<TopicPartition>> activeTasks,
for (final Task task : tasksToClose) {
try {
- task.suspend(); // Should be a no-op for active tasks since they're suspended in handleRevocation
- if (task.commitNeeded()) {
- if (task.isActive()) {
- log.error("Active task {} was revoked and should have already been committed", task.id());
- throw new IllegalStateException("Revoked active task was not committed during handleRevocation");
Review comment:
This was another "sort-of bug": if we hit an exception in `handleRevocation` we wouldn't finish committing the active tasks, so `commitNeeded` could still be true. But of course, if we hit an exception earlier, we would have thrown it up to ConsumerCoordinator which would only save the first exception, so this didn't really do anything
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on pull request #8926: KAFKA-10166: always write checkpoint before closing an (initialized) task
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #8926:
URL: https://github.com/apache/kafka/pull/8926#issuecomment-649936840
Test this please
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] abbccdda commented on pull request #8926: KAFKA-10166: always write checkpoint before closing an (initialized) task
Posted by GitBox <gi...@apache.org>.
abbccdda commented on pull request #8926:
URL: https://github.com/apache/kafka/pull/8926#issuecomment-649934446
test 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #8926: KAFKA-10166: always invoke `postCommit` before closing a task
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #8926:
URL: https://github.com/apache/kafka/pull/8926#discussion_r445262784
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
##########
@@ -813,27 +821,6 @@ void shutdown(final boolean clean) {
tasksToCloseDirty.add(task);
}
}
-
- for (final Task task : tasksToCommit) {
Review comment:
We can actually simplify the standby task shutdown a LOT
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang merged pull request #8926: KAFKA-10166: always write checkpoint before closing an (initialized) task
Posted by GitBox <gi...@apache.org>.
guozhangwang merged pull request #8926:
URL: https://github.com/apache/kafka/pull/8926
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on pull request #8926: KAFKA-10166: always write checkpoint before closing an (initialized) task
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #8926:
URL: https://github.com/apache/kafka/pull/8926#issuecomment-649937082
Oh, yeah... the magic touch ;)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on pull request #8926: KAFKA-10166: always write checkpoint before closing an (initialized) task
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #8926:
URL: https://github.com/apache/kafka/pull/8926#issuecomment-649937035
Retest this please
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] abbccdda commented on pull request #8926: KAFKA-10166: always write checkpoint before closing an (initialized) task
Posted by GitBox <gi...@apache.org>.
abbccdda commented on pull request #8926:
URL: https://github.com/apache/kafka/pull/8926#issuecomment-650385489
```12:23:50 FAILURE: Build failed with an exception.
12:23:50
12:23:50 * What went wrong:
12:23:50 Execution failed for task ':streams:unitTest'.
12:23:50 > Process 'Gradle Test Executor 9' finished with non-zero exit value 134
12:23:50 This problem might be caused by incorrect test process configuration.
12:23:50 Please refer to the test execution section in the User Manual at https://docs.gradle.org/6.5/userguide/java_testing.html#sec:test_execution
12:23:50 ```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a change in pull request #8926: [WIP] KAFKA-10166: always invoke `postCommit` before closing a task
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #8926:
URL: https://github.com/apache/kafka/pull/8926#discussion_r445907018
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
##########
@@ -469,21 +465,27 @@ void handleRevocation(final Collection<TopicPartition> revokedPartitions) {
throw suspendException;
}
+ final Map<TaskId, Map<TopicPartition, OffsetAndMetadata>> consumedOffsetsAndMetadataPerTask = new HashMap<>();
+ prepareCommitAndAddOffsetsToMap(revokedTasks, consumedOffsetsAndMetadataPerTask);
+
// If using eos-beta, if we must commit any task then we must commit all of them
// TODO: when KAFKA-9450 is done this will be less expensive, and we can simplify by always committing everything
- if (processingMode == EXACTLY_ONCE_BETA && !tasksToCommit.isEmpty()) {
- tasksToCommit.addAll(additionalTasksForCommitting);
- }
-
- final Map<TaskId, Map<TopicPartition, OffsetAndMetadata>> consumedOffsetsAndMetadataPerTask = new HashMap<>();
- for (final Task task : tasksToCommit) {
- final Map<TopicPartition, OffsetAndMetadata> committableOffsets = task.prepareCommit();
- consumedOffsetsAndMetadataPerTask.put(task.id(), committableOffsets);
+ if (processingMode == EXACTLY_ONCE_BETA && !consumedOffsetsAndMetadataPerTask.isEmpty()) {
+ prepareCommitAndAddOffsetsToMap(additionalTasksForCommitting, consumedOffsetsAndMetadataPerTask);
}
commitOffsetsOrTransaction(consumedOffsetsAndMetadataPerTask);
- for (final Task task : tasksToCommit) {
+ for (final Task task : revokedTasks) {
+ try {
+ task.postCommit();
+ } catch (final RuntimeException e) {
+ log.error("Exception caught while post-committing task " + task.id(), e);
+ firstException.compareAndSet(null, e);
+ }
+ }
+
+ for (final Task task : additionalTasksForCommitting) {
Review comment:
If we are not committing these tasks, should we call their `postCommit`?
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
##########
@@ -813,27 +821,6 @@ void shutdown(final boolean clean) {
tasksToCloseDirty.add(task);
}
}
-
- for (final Task task : tasksToCommit) {
Review comment:
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on pull request #8926: KAFKA-10166: always write checkpoint before closing an (initialized) task
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #8926:
URL: https://github.com/apache/kafka/pull/8926#issuecomment-650288465
Test this please
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on pull request #8926: [WIP] KAFKA-10166: always invoke `postCommit` before closing a task
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #8926:
URL: https://github.com/apache/kafka/pull/8926#issuecomment-649882757
@ableegoldman please lmk when you want to trigger jenkins builds on this 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] abbccdda commented on pull request #8926: KAFKA-10166: always write checkpoint before closing an (initialized) task
Posted by GitBox <gi...@apache.org>.
abbccdda commented on pull request #8926:
URL: https://github.com/apache/kafka/pull/8926#issuecomment-649932331
retest 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on pull request #8926: KAFKA-10166: always write checkpoint before closing an (initialized) task
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on pull request #8926:
URL: https://github.com/apache/kafka/pull/8926#issuecomment-650251400
Seems like all the Topology testDriver tests failed, but I got a green build running locally. Do they not run with `./gradlew streams:test`?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on pull request #8926: KAFKA-10166: always write checkpoint before closing an (initialized) task
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #8926:
URL: https://github.com/apache/kafka/pull/8926#issuecomment-650288339
Test this please
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org