Shekharrajak commented on code in PR #22357:
URL: https://github.com/apache/kafka/pull/22357#discussion_r3994720515
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -1860,6 +1864,34 @@ class KafkaApis(val requestChannel: RequestChannel,
}
addResultAndMaybeComplete(partition, error)
}
+ } else if (partition.topic == SHARE_GROUP_STATE_TOPIC_NAME) {
+ shareCoordinator.completeTransaction(
+ partition,
+ marker.producerId,
+ marker.producerEpoch,
+ marker.coordinatorEpoch,
+ marker.transactionResult,
+ markerTransactionVersion
+ ).whenComplete { (affectedSharePartitions, exception) =>
+ val markerError = if (exception == null) {
+ if (affectedSharePartitions != null &&
!affectedSharePartitions.isEmpty) {
+ try {
+
sharePartitionManager.invalidateSharePartitions(affectedSharePartitions)
Review Comment:
How Share Coordinator will know that records are committed during the txn
commit marker process :
```
Broker A: leader serving source topic orders-2 and its ShareFetch cache
Broker B: leader for __share_group_state-7
Broker C: transaction coordinator for transactional.id = flink-job-1
1. Worker stages an acknowledgement for orders-2.
Broker A handles TxnShareAcknowledge:
- stages records in its local SharePartition cache as TX_PENDING
- writes that state through the persister to Broker B's
__share_group_state-7
- registers __share_group_state-7 with Broker C's transaction
2. Producer commits.
Broker C records the transaction decision and sends WriteTxnMarkers
to Broker B, because B owns __share_group_state-7.
3. Broker B finalizes durable share state.
ShareCoordinatorShard changes matching state:
TX_PENDING -> ACKNOWLEDGED / ARCHIVED on commit
TX_PENDING -> AVAILABLE on abort
4. Broker B runs the current invalidation code.
KafkaApis on B calls B's local:
sharePartitionManager.invalidateSharePartitions(keys)
Also if we need to invalidate in Broker A as well then Broker A will not
redeliver the pending range, even though its cache has not yet heard about the
commit.
On Broker A’s next ShareFetch, SharePartitionManager detects TX_PENDING in
its local cache:
if (sharePartition.hasPendingTransactionalRecords()) {
sharePartition = reinitializePendingTransactionalSharePartition(...);
}
It removes the old cache entry and reloads the authoritative state from
__share_group_state.
After reload:
commit -> terminal state, no redelivery
abort -> AVAILABLE, redelivery becomes possible
```
--
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]
