Shekharrajak commented on code in PR #22357:
URL: https://github.com/apache/kafka/pull/22357#discussion_r3294077952
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1051,6 +1052,116 @@ public CompletableFuture<Void> acknowledge(
return future;
}
+ public CompletableFuture<Void> stageTxnAcknowledge(
+ String memberId,
+ long producerId,
+ short producerEpoch,
+ List<ShareAcknowledgementBatch> acknowledgementBatches
+ ) {
+ log.trace("Txn stage request for share partition: {}-{}
producerId={}", groupId, topicIdPartition, producerId);
+
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ Throwable throwable = null;
+ lock.writeLock().lock();
+ try {
+ for (ShareAcknowledgementBatch batch : acknowledgementBatches) {
+ Map<Long, Byte> ackTypeMap;
+ try {
+ ackTypeMap = fetchAckTypeMapForBatch(batch);
+ } catch (IllegalArgumentException e) {
+ throwable = new InvalidRequestException("Invalid
acknowledge type: " + batch.acknowledgeTypes());
+ break;
+ }
+
+ for (Byte ackType : ackTypeMap.values()) {
+ if (ackType != AcknowledgeType.ACCEPT.id && ackType !=
AcknowledgeType.REJECT.id) {
+ throwable = new InvalidRequestException("Only ACCEPT
or REJECT permitted in transactional ack");
+ break;
+ }
+ }
+ if (throwable != null) break;
+
+ if (batch.lastOffset() < startOffset) continue;
+
+ NavigableMap<Long, InFlightBatch> subMap;
+ try {
+ subMap = fetchSubMapForAcknowledgementBatch(batch);
+ } catch (InvalidRecordStateException | InvalidRequestException
e) {
+ throwable = e;
+ break;
+ }
+
+ throwable = stageBatchTxnRecords(memberId, producerId,
producerEpoch, batch, ackTypeMap, subMap);
+ if (throwable != null) break;
Review Comment:
break if there is one stage txn failed with any exception - if previous ones
are success, then those updated with txn_pending state but current txn batch
not.
transaction will abort and applyTxnMarker(ABORT) will revert them to
AVAILABLE both batch 1 (which was succeed and batch 2 which got failed in some
step)
```
# user code :
try {
producer.sendShareAcknowledgementsToTransaction(acks, shareGroupMeta);
// ← throws
producer.commitTransaction();
} catch (KafkaException e) {
producer.abortTransaction(); // ← this is what reverts the records
state back to
}
```
--
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]
