lucasbru commented on code in PR #23791:
URL: https://github.com/apache/kafka/pull/23791#discussion_r4269047773
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java:
##########
@@ -601,13 +601,15 @@ public void closeClean() {
@Override
public void closeDirty() {
log.info("Closing record collector dirty");
-
- if (streamsProducer.eosEnabled()) {
- // We may be closing dirty because the commit failed, so we must
abort the transaction to be safe
- streamsProducer.abortTransaction();
+ try {
+ if (streamsProducer.eosEnabled()) {
+ // We may be closing dirty because the commit failed, so we
must abort the transaction to be safe
+ streamsProducer.abortTransaction();
+ }
+
Review Comment:
Nit: whitespace-only line.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -2204,7 +2204,7 @@ public static void executeAndMaybeSwallow(final boolean
clean,
e -> {
throw e;
},
- e -> log.debug("Ignoring error in unclean {}", name));
+ e -> log.error("Error closing unclean {}", name));
Review Comment:
The lambda drops `e`, so this logs only the name and neither the message nor
the stack trace. Should it be `log.error("Error closing unclean {}", name, e)`?
Also, this covers every dirty close, not just the abort failure, so I wonder if
ERROR gets noisy for expected errors.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java:
##########
@@ -601,13 +601,15 @@ public void closeClean() {
@Override
public void closeDirty() {
log.info("Closing record collector dirty");
-
- if (streamsProducer.eosEnabled()) {
- // We may be closing dirty because the commit failed, so we must
abort the transaction to be safe
- streamsProducer.abortTransaction();
+ try {
+ if (streamsProducer.eosEnabled()) {
+ // We may be closing dirty because the commit failed, so we
must abort the transaction to be safe
+ streamsProducer.abortTransaction();
+ }
+
+ } finally {
+ close();
Review Comment:
Coming back to my earlier question: `close()` still calls
`checkForException()`, so a pending send error thrown here replaces the abort
exception. Is that fine, or should the offsets be cleared in the `finally` and
the send error thrown only on the normal path?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
