hachikuji commented on code in PR #12915:
URL: https://github.com/apache/kafka/pull/12915#discussion_r1036359658
##########
core/src/main/scala/kafka/coordinator/transaction/ProducerIdManager.scala:
##########
@@ -167,7 +167,7 @@ class RPCProducerIdManager(brokerId: Int,
if (nextProducerId > currentProducerIdBlock.lastProducerId) {
val block = nextProducerIdBlock.poll(maxWaitMs, TimeUnit.MILLISECONDS)
if (block == null) {
- throw Errors.REQUEST_TIMED_OUT.exception("Timed out waiting for next
producer ID block")
+ throw Errors.CONCURRENT_TRANSACTIONS.exception("Timed out waiting
for next producer ID block")
Review Comment:
Should we do this fix separately? I thought we had considered whether a
short KIP would be needed. We probably also need to consider the behavior of
other clients (such as rdkafka).
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -1515,7 +1515,7 @@ public void handleResponse(AbstractResponse response) {
} else if (error == Errors.COORDINATOR_NOT_AVAILABLE || error ==
Errors.NOT_COORDINATOR) {
lookupCoordinator(FindCoordinatorRequest.CoordinatorType.TRANSACTION,
transactionalId);
reenqueue();
- } else if (error == Errors.COORDINATOR_LOAD_IN_PROGRESS || error
== Errors.CONCURRENT_TRANSACTIONS) {
+ } else if (error == Errors.CONCURRENT_TRANSACTIONS ||
error.exception() instanceof RetriableException) {
Review Comment:
I think `ConcurrentTransactionsException` should be retriable.
--
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
