mjsax commented on a change in pull request #8856:
URL: https://github.com/apache/kafka/pull/8856#discussion_r441101688
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
##########
@@ -714,13 +717,26 @@ void shutdown(final boolean clean) {
}
}
- if (clean && !consumedOffsetsAndMetadataPerTask.isEmpty()) {
- commitOffsetsOrTransaction(consumedOffsetsAndMetadataPerTask);
+ try {
+ if (clean && !consumedOffsetsAndMetadataPerTask.isEmpty()) {
+ commitOffsetsOrTransaction(consumedOffsetsAndMetadataPerTask);
Review comment:
> Committable offsets here should contain consumed offsets, and
punctuation itself should never update those consumed offsets right
Yes.
> I think we can skip the call if consumedOffsetsAndMetadataPerTask is empty.
For non-eos, yes, because for non-eos `commitOffsetsOrTransaction()` would
only commit offsets via the consumer (this can be skipped if empty). However,
for eos (alpha and beta), we might have a pending transaction that we need to
commit on the producer, too.
----------------------------------------------------------------
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:
us...@infra.apache.org
