junrao commented on code in PR #14111:
URL: https://github.com/apache/kafka/pull/14111#discussion_r1290474356
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -140,17 +167,30 @@ public long metadataExpireMs() {
}
/**
- * Request an update of the current cluster metadata info, return the
current updateVersion before the update
+ * Request an update of the current cluster metadata info, permitting
backoff based on the number of
+ * equivalent responses, which indicate that metadata responses did not
make progress and may be stale.
Review Comment:
Could we add a comment on when the caller should set
permitBackoffOnEquivalentResponses to true? Also, when should
backoffOnEquivalentResponses be reset to false?
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -114,18 +127,32 @@ public synchronized Cluster fetch() {
/**
* Return the next time when the current cluster info can be updated
(i.e., backoff time has elapsed).
+ * There are two calculations for backing off based on how many attempts
to retrieve metadata have been made
+ * since the last successful response, and how many equivalent metadata
responses have been received.
+ * The second of these allows backing off when there are errors to do with
stale metadata, even though the
+ * metadata responses are clean.
+ * <p>
+ * This can be used to check whether it's worth requesting an update in
the knowledge that it will
+ * not be delayed if this method returns 0.
*
* @param nowMs current time in ms
* @return remaining time in ms till the cluster info can be updated again
*/
public synchronized long timeToAllowUpdate(long nowMs) {
- return Math.max(this.lastRefreshMs + this.refreshBackoffMs - nowMs, 0);
+ // Calculate the backoff for attempts which acts when metadata
responses fail
+ long backoffForAttempts = Math.max(this.lastRefreshMs +
this.refreshBackoff.backoff(this.attempts) - nowMs, 0);
+
+ // Calculate the backoff for equivalent responses which acts when
metadata responses as not making progress
Review Comment:
as not making => are not making ?
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:
##########
@@ -598,22 +608,22 @@ private void insertInSequenceOrder(Deque<ProducerBatch>
deque, ProducerBatch bat
/**
* Add the leader to the ready nodes if the batch is ready
*
- * @param nowMs The current time
* @param exhausted 'true' is the buffer pool is exhausted
* @param part The partition
* @param leader The leader for the partition
* @param waitedTimeMs How long batch waited
* @param backingOff Is backing off
+ * @param backoffAttempts Number of attempts for calculating backoff delay
* @param full Is batch full
* @param nextReadyCheckDelayMs The delay for next check
* @param readyNodes The set of ready nodes (to be filled in)
* @return The delay for next check
*/
- private long batchReady(long nowMs, boolean exhausted, TopicPartition
part, Node leader,
- long waitedTimeMs, boolean backingOff, boolean
full,
- long nextReadyCheckDelayMs, Set<Node> readyNodes) {
+ private long batchReady(boolean exhausted, TopicPartition part, Node
leader,
+ long waitedTimeMs, boolean backingOff, int
backoffAttempts,
+ boolean full, long nextReadyCheckDelayMs,
Set<Node> readyNodes) {
if (!readyNodes.contains(leader) && !isMuted(part)) {
- long timeToWaitMs = backingOff ? retryBackoffMs : lingerMs;
+ long timeToWaitMs = backingOff ?
retryBackoff.backoff(backoffAttempts) : lingerMs;
Review Comment:
Here, backoffAttempts is already subtracted by one when calling
retryBackoff.backoff. In other places, we pass in the true attempts and
subtract by one when calling retryBackoff.backoff. It would be useful to make
that consistent.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java:
##########
@@ -550,7 +550,7 @@ public boolean poll(Timer timer, boolean waitForJoinGroup) {
// refresh metadata before re-joining the group as long as
the refresh backoff time has
// passed.
if (this.metadata.timeToAllowUpdate(timer.currentTimeMs())
== 0) {
- this.metadata.requestUpdate();
+ this.metadata.requestUpdate(true);
Review Comment:
Why is permitBackoffOnEquivalentResponses set to true? We refresh the
metadata here not because we have discovered stale metadata.
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -140,17 +167,30 @@ public long metadataExpireMs() {
}
/**
- * Request an update of the current cluster metadata info, return the
current updateVersion before the update
+ * Request an update of the current cluster metadata info, permitting
backoff based on the number of
+ * equivalent responses, which indicate that metadata responses did not
make progress and may be stale.
+ * @param permitBackoffOnEquivalentResponses Whether to permit backoff
when consecutive responses are equivalent
+ * @return The current updateVersion before the update
*/
- public synchronized int requestUpdate() {
+ public synchronized int requestUpdate(final boolean
permitBackoffOnEquivalentResponses) {
this.needFullUpdate = true;
+ if (this.backoffOnEquivalentResponses) {
Review Comment:
Hmm, we need to check backoffOnEquivalentResponses when calculating
`timeToAllowUpdate`, right?
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -267,11 +307,14 @@ public synchronized void update(int requestVersion,
MetadataResponse response, b
this.needPartialUpdate = requestVersion < this.requestVersion;
this.lastRefreshMs = nowMs;
+ this.attempts = 0;
this.updateVersion += 1;
if (!isPartialUpdate) {
this.needFullUpdate = false;
this.lastSuccessfulRefreshMs = nowMs;
}
+ this.backoffOnEquivalentResponses = true;
+ this.equivalentResponseCount++;
Review Comment:
The metadata request can also be driven by periodic metadata refresh. In
that case, we don't want to trigger the exponential backoff if the metadata
response doesn't change. However, in the PR, it seems that once
backoffOnEquivalentResponses is set to true. It's not reset to false before the
periodic metadata refresh?
--
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
