Shekharrajak commented on code in PR #22357:
URL: https://github.com/apache/kafka/pull/22357#discussion_r3430476846
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -828,13 +828,7 @@ private CompletableFuture<SharePartition>
initializeSharePartitionForTransaction
SharePartitionKey sharePartitionKey
) {
SharePartition sharePartition =
getOrCreateSharePartition(sharePartitionKey);
- return sharePartition.maybeInitialize().thenCompose(ignored -> {
- if (sharePartition.hasPendingTransactionalRecords()) {
- SharePartition refreshedSharePartition =
reinitializePendingTransactionalSharePartition(sharePartitionKey);
Review Comment:
SharePartitionManager.initializeSharePartitionForTransactionalAcknowledge
now keeps the current cached SharePartition for transactional ack staging
instead of reinitializing when
pending transactional records exist.
Otherwise this throws invalid state
https://github.com/apache/kafka/pull/22357/changes/4564874a1623c41798e846348691a0c0e7770cec#r3430460314
--
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]
