jolshan commented on code in PR #15524:
URL: https://github.com/apache/kafka/pull/15524#discussion_r1523931941
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerRequestCompletionHandler.scala:
##########
@@ -39,22 +39,23 @@ class TransactionMarkerRequestCompletionHandler(brokerId:
Int,
if (response.wasDisconnected) {
trace(s"Cancelled request with header $requestHeader due to node
${response.destination} being disconnected")
- for (txnIdAndMarker <- txnIdAndMarkerEntries.asScala) {
- val transactionalId = txnIdAndMarker.txnId
- val txnMarker = txnIdAndMarker.txnMarkerEntry
+ for (pendingCompleteTxnAndMarker <-
pendingCompleteTxnAndMarkerEntries.asScala) {
+ val pendingCompleteTxn = pendingCompleteTxnAndMarker.pendingCompleteTxn
+ val transactionalId = pendingCompleteTxn.transactionalId
+ val txnMarker = pendingCompleteTxnAndMarker.txnMarkerEntry
txnStateManager.getTransactionState(transactionalId) match {
case Left(Errors.NOT_COORDINATOR) =>
info(s"I am no longer the coordinator for $transactionalId; cancel
sending transaction markers $txnMarker to the brokers")
- txnMarkerChannelManager.removeMarkersForTxnId(transactionalId)
Review Comment:
unrelated to the changes here, but the first person ("I am") logging here is
somewhat strange 😅
--
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]
