msn-tldr commented on code in PR #14384:
URL: https://github.com/apache/kafka/pull/14384#discussion_r1345590803
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java:
##########
@@ -94,9 +100,40 @@ public ProducerBatch(TopicPartition tp,
MemoryRecordsBuilder recordsBuilder, lon
this.isSplitBatch = isSplitBatch;
float compressionRatioEstimation =
CompressionRatioEstimator.estimation(topicPartition.topic(),
recordsBuilder.compressionType());
+ this.currentLeaderEpoch = PartitionInfo.UNKNOWN_LEADER_EPOCH;
+ this.leaderChangedAttempts = -1;
recordsBuilder.setEstimatedCompressionRatio(compressionRatioEstimation);
}
+ /*
+ * Returns whether the leader epoch has changed since the last attempt.
+ * @param latestLeaderEpoch The latest leader epoch.
+ * @return true if the leader has changed, otherwise false.
+ */
+ boolean hasLeaderChanged(int latestLeaderEpoch) {
+ boolean leaderChanged = false;
+ // Checking for leader change makes sense only from 1st retry
onwards(attempt >=1).
+ log.trace("For {}, attempting to change leader, currentLeaderEpoch:{},
leaderChangedAttempts:{}, latestLeaderEpoch: {}, current Attempt: {}",
+ this, currentLeaderEpoch, leaderChangedAttempts,
latestLeaderEpoch, attempts());
+ if (attempts() >= 1) {
+ // If the leader's epoch has changed, this counts as a leader
change
+ if (currentLeaderEpoch != latestLeaderEpoch) {
+ leaderChangedAttempts = attempts();
+ leaderChanged = true;
+ } else {
+ // Otherwise, it's only a leader change until the first
attempt is made with this leader
Review Comment:
> I think I can see how the flow of the code would end up calling that
multiple times.
I gets called multiple times. Its sendProducerData -> ready ->
partitionReady, and then again in sendProducerData -> drain ->
drainBatchesForOneNode.
--
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
