hachikuji commented on a change in pull request #11452:
URL: https://github.com/apache/kafka/pull/11452#discussion_r755352383
##########
File path:
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
##########
@@ -423,34 +426,31 @@ public synchronized TransactionalRequestResult
sendOffsetsToTransaction(final Ma
return handler.result;
}
- public synchronized void maybeAddPartitionToTransaction(TopicPartition
topicPartition) {
- if (isPartitionAdded(topicPartition) ||
isPartitionPendingAdd(topicPartition))
- return;
+ public synchronized void maybeAddPartition(TopicPartition topicPartition) {
+ maybeFailWithError();
+ throwIfPendingState("send");
- log.debug("Begin adding new partition {} to transaction",
topicPartition);
- topicPartitionBookkeeper.addPartition(topicPartition);
- newPartitionsInTransaction.add(topicPartition);
+ if (isTransactional()) {
+ if (!hasProducerId()) {
+ throw new KafkaException("Cannot add partition " +
topicPartition +
+ "to transaction before completing a call to
initTransactions");
+ } else if (currentState != State.IN_TRANSACTION) {
+ throw new KafkaException("Cannot add partition " +
topicPartition +
Review comment:
Consistency was the only reason. It looked weird in the test case.
--
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
