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/06 00:27:25 UTC
[GitHub] [kafka] abbccdda opened a new pull request #8820: KAFKA-10097: Internalize checkpoint data
abbccdda opened a new pull request #8820:
URL: https://github.com/apache/kafka/pull/8820
- make checkpoint an internal data structure to avoid awkward passing around.
- remove the possibility of getting a null checkpoint map
### 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on pull request #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#issuecomment-639960633
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] abbccdda commented on a change in pull request #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
abbccdda commented on a change in pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#discussion_r436220769
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -546,14 +547,14 @@ public void closeAndRecycleState() {
* otherwise, just close open resources
* @throws TaskMigratedException if the task producer got fenced (EOS)
*/
- private Map<TopicPartition, Long> prepareClose(final boolean clean) {
- final Map<TopicPartition, Long> checkpoint;
+ private void prepareClose(final boolean clean) {
+ // Reset any previously scheduled checkpoint.
Review comment:
I think we could just reset any checkpoint at the beginning, so that we only do checkpointing if this call thinks so. @mjsax
----------------------------------------------------------------
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] mjsax commented on a change in pull request #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#discussion_r436223027
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -598,9 +606,9 @@ public void closeAndRecycleState() {
* 3. finally release the state manager lock
* </pre>
*/
- private void close(final boolean clean, final Map<TopicPartition, Long> checkpoint) {
- if (clean && checkpoint != null) {
- executeAndMaybeSwallow(clean, () -> stateMgr.checkpoint(checkpoint), "state manager checkpoint", log);
+ private void close(final boolean clean) {
+ if (clean && checkpointNeeded) {
Review comment:
A `if (clean)` should be sufficient 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on pull request #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#issuecomment-640081002
okay 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] abbccdda commented on a change in pull request #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
abbccdda commented on a change in pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#discussion_r436219178
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -562,32 +561,41 @@ public void closeAndRecycleState() {
if (clean) {
stateMgr.flush();
recordCollector.flush();
- checkpoint = checkpointableOffsets();
+
+ scheduleCheckpoint(checkpointableOffsets());
} else {
- checkpoint = null; // `null` indicates to not write a checkpoint
executeAndMaybeSwallow(false, stateMgr::flush, "state manager flush", log);
}
break;
case RESTORING:
executeAndMaybeSwallow(clean, stateMgr::flush, "state manager flush", log);
- checkpoint = Collections.emptyMap();
+ scheduleCheckpoint(emptyMap());
break;
case SUSPENDED:
case CLOSED:
// not need to checkpoint, since when suspending we've already committed the state
- checkpoint = null; // `null` indicates to not write a checkpoint
-
break;
default:
throw new IllegalStateException("Unknown state " + state() + " while prepare closing active task " + id);
}
+ }
- return checkpoint;
+ private void scheduleCheckpoint(final Map<TopicPartition, Long> checkpoint) {
+ this.checkpoint = checkpoint;
+ this.checkpointNeeded = true;
+ }
+
+ private void writeCheckpoint() {
Review comment:
Sounds good
----------------------------------------------------------------
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 #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#discussion_r436229822
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -107,6 +107,9 @@
private boolean commitNeeded = false;
private boolean commitRequested = false;
+ private boolean checkpointNeeded = false;
Review comment:
Do we need two parameters here? Could we still just use one `Map<TopicPartition, Long> checkpoint`, and then in `writeCheckpointIfNeed` if it is not null write it and then set it to null.
----------------------------------------------------------------
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 #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#issuecomment-639961430
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] mjsax commented on a change in pull request #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#discussion_r436215899
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -562,32 +561,41 @@ public void closeAndRecycleState() {
if (clean) {
stateMgr.flush();
recordCollector.flush();
- checkpoint = checkpointableOffsets();
+
+ scheduleCheckpoint(checkpointableOffsets());
} else {
- checkpoint = null; // `null` indicates to not write a checkpoint
executeAndMaybeSwallow(false, stateMgr::flush, "state manager flush", log);
}
break;
case RESTORING:
executeAndMaybeSwallow(clean, stateMgr::flush, "state manager flush", log);
- checkpoint = Collections.emptyMap();
+ scheduleCheckpoint(emptyMap());
break;
case SUSPENDED:
case CLOSED:
// not need to checkpoint, since when suspending we've already committed the state
- checkpoint = null; // `null` indicates to not write a checkpoint
-
break;
default:
throw new IllegalStateException("Unknown state " + state() + " while prepare closing active task " + id);
}
+ }
- return checkpoint;
+ private void scheduleCheckpoint(final Map<TopicPartition, Long> checkpoint) {
+ this.checkpoint = checkpoint;
+ this.checkpointNeeded = true;
+ }
+
+ private void writeCheckpoint() {
Review comment:
Maybe easier to make this `writeCheckpointIfNeeded`:
```
private void writeCheckpointIfNeeded() {
if (checkpointNeeded) {
stateMgr.checkpoint(checkpoint);
checkpointNeeded = false;
}
}
```
For this case, the caller does not need to check the `checkpointNeeded` flag?
----------------------------------------------------------------
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 a change in pull request #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
abbccdda commented on a change in pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#discussion_r436237426
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -107,6 +107,9 @@
private boolean commitNeeded = false;
private boolean commitRequested = false;
+ private boolean checkpointNeeded = false;
Review comment:
A third option is Optional<Map<TopicPartition, Long>> which explicitly states the checkpoint is null. However, I think the benefit here is minimal, as `checkpointNeeded` sounds like a better state variable, than relying on the struct itself.
----------------------------------------------------------------
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] mjsax merged pull request #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
mjsax merged pull request #8820:
URL: https://github.com/apache/kafka/pull/8820
----------------------------------------------------------------
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] mjsax commented on pull request #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#issuecomment-640112555
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] guozhangwang commented on pull request #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#issuecomment-640080953
okay 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] mjsax commented on pull request #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#issuecomment-640112530
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] mjsax commented on a change in pull request #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#discussion_r436222956
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -546,14 +547,14 @@ public void closeAndRecycleState() {
* otherwise, just close open resources
* @throws TaskMigratedException if the task producer got fenced (EOS)
*/
- private Map<TopicPartition, Long> prepareClose(final boolean clean) {
- final Map<TopicPartition, Long> checkpoint;
+ private void prepareClose(final boolean clean) {
+ // Reset any previously scheduled checkpoint.
Review comment:
That should work.
----------------------------------------------------------------
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] mjsax commented on a change in pull request #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#discussion_r436233382
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -107,6 +107,9 @@
private boolean commitNeeded = false;
private boolean commitRequested = false;
+ private boolean checkpointNeeded = false;
Review comment:
I actually had a similar thought, but I am torn though. Using two variables required to keep them "in sync" was is not great. However, using `null` is less explicit... Thus overall I am fine either way as both seems to provide the overall same good/bad ratio.
----------------------------------------------------------------
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] mjsax commented on pull request #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#issuecomment-640138817
Merged to `trunk` and cherry-picked to `2.6` branch.
----------------------------------------------------------------
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 #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#issuecomment-639961343
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 a change in pull request #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
abbccdda commented on a change in pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#discussion_r436229349
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -598,9 +606,9 @@ public void closeAndRecycleState() {
* 3. finally release the state manager lock
* </pre>
*/
- private void close(final boolean clean, final Map<TopicPartition, Long> checkpoint) {
- if (clean && checkpoint != null) {
- executeAndMaybeSwallow(clean, () -> stateMgr.checkpoint(checkpoint), "state manager checkpoint", log);
+ private void close(final boolean clean) {
+ if (clean && checkpointNeeded) {
Review comment:
Make 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] mjsax commented on pull request #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#issuecomment-639949059
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 a change in pull request #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
abbccdda commented on a change in pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#discussion_r436278473
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -107,6 +107,9 @@
private boolean commitNeeded = false;
private boolean commitRequested = false;
+ private boolean checkpointNeeded = false;
Review comment:
Cool
----------------------------------------------------------------
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 #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#issuecomment-640080856
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] guozhangwang commented on a change in pull request #8820: KAFKA-10097: Internalize checkpoint data
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #8820:
URL: https://github.com/apache/kafka/pull/8820#discussion_r436238804
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -107,6 +107,9 @@
private boolean commitNeeded = false;
private boolean commitRequested = false;
+ private boolean checkpointNeeded = false;
Review comment:
Actually I'm not too concerned of relying on `null` to indicate no need to checkpoint, what I originally pointed out is that we are unnecessarily accumulating and then distributing checkpoints between task manager and task, which is already resolved in your PR. So I'd suggest we still just use a single nullable `Map`.
----------------------------------------------------------------
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