hachikuji commented on a change in pull request #11452:
URL: https://github.com/apache/kafka/pull/11452#discussion_r751490837
##########
File path:
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
##########
@@ -1184,15 +1186,19 @@ private TxnOffsetCommitHandler
txnOffsetCommitHandler(TransactionalRequestResult
}
private TransactionalRequestResult handleCachedTransactionRequestResult(
- Supplier<TransactionalRequestResult>
transactionalRequestResultSupplier,
- State targetState) {
+ Supplier<TransactionalRequestResult>
transactionalRequestResultSupplier,
+ State transientState
+ ) {
ensureTransactional();
- if (pendingResult != null && currentState == targetState) {
+ if (pendingResult != null && !pendingResult.isAcked()) {
TransactionalRequestResult result = pendingResult;
- if (result.isCompleted())
+ if (result.isCompleted() && priorState == transientState) {
pendingResult = null;
- return result;
+ return result;
+ } else if (currentState == transientState) {
+ return result;
Review comment:
The logic was pretty confusing, so I attempted to simplify it in the
latest commit. The need to ensure that the result gets communicated to the user
definitely makes this code more awkward.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
