You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lianetm (via GitHub)" <gi...@apache.org> on 2024/02/02 17:12:56 UTC
[PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
lianetm opened a new pull request, #15311:
URL: https://github.com/apache/kafka/pull/15311
This fixes a bug that was causing that members wouldn't rejoin the group after receiving an error in the heartbeat response (ex. fenced, not coordinator, as reported in KAFKA-16215 and KAFKA-16178). The issue was that when receiving a response with error, the response receive time was not being updated, so following heartbeat would be skipped, considering that there was already a previous one inflight.
Tests updated to cover this.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #15311:
URL: https://github.com/apache/kafka/pull/15311#discussion_r1476532820
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##########
@@ -106,6 +106,13 @@ public void onSendAttempt(final long currentTimeMs) {
this.lastSentMs = currentTimeMs;
}
+ /**
+ * Update the lastReceivedTime in milliseconds, indicating that a response has been received.
+ */
+ public void updateLastReceivedTime(final long lastReceivedMs) {
+ this.lastReceivedMs = lastReceivedMs;
+ }
+
Review Comment:
Is updating this timestamp something we should do for _all_ `RequestManager`s?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15311:
URL: https://github.com/apache/kafka/pull/15311#discussion_r1477940236
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##########
@@ -106,6 +106,13 @@ public void onSendAttempt(final long currentTimeMs) {
this.lastSentMs = currentTimeMs;
}
+ /**
+ * Update the lastReceivedTime in milliseconds, indicating that a response has been received.
+ */
+ public void updateLastReceivedTime(final long lastReceivedMs) {
+ this.lastReceivedMs = lastReceivedMs;
Review Comment:
I find doing this a little strange because we ends up in a state where the last received time is updated but the backoff and the number of attempts are not. Therefore, we could still have an non-desired backoff applied to the next request. For instance, imagine having a request timeout followed by a coordinator not available error. Is this case, the backoff is not reset. Is my understanding correct?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #15311:
URL: https://github.com/apache/kafka/pull/15311#discussion_r1483183111
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##########
@@ -132,6 +132,16 @@ public void onFailedAttempt(final long currentTimeMs) {
this.numAttempts++;
}
+ /**
+ * Set backoff and number of attempts to 0. This will ensure that the request is sent out
+ * again right away. Expected to be used when receiving errors responses that lead to a new
+ * request that can be sent without backing off (ex. member rejoining after fencing)
+ */
+ public void resetBackoff() {
Review Comment:
yeap, that one was resetting the full state (not only the backoff), so initially I thought better to reset only what we need. But I totally agree that the "retries" after these errors can be seen as a fresh start, it does not harm to reset the full state, and it does simplify to reuse the single reset, done.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #15311:
URL: https://github.com/apache/kafka/pull/15311#discussion_r1480405280
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##########
@@ -106,6 +106,13 @@ public void onSendAttempt(final long currentTimeMs) {
this.lastSentMs = currentTimeMs;
}
+ /**
+ * Update the lastReceivedTime in milliseconds, indicating that a response has been received.
+ */
+ public void updateLastReceivedTime(final long lastReceivedMs) {
+ this.lastReceivedMs = lastReceivedMs;
Review Comment:
Exactly, I did see the skip backoff logic in produce when a new leader is discovered, related to what you described. I think that the concept of "progress" here would be more abstract but still applicable, depending on the exact error we know that the action it triggers is based on some progress (send HB as new member, send HB when coord available)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #15311:
URL: https://github.com/apache/kafka/pull/15311#discussion_r1476728141
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##########
@@ -106,6 +106,13 @@ public void onSendAttempt(final long currentTimeMs) {
this.lastSentMs = currentTimeMs;
}
+ /**
+ * Update the lastReceivedTime in milliseconds, indicating that a response has been received.
+ */
+ public void updateLastReceivedTime(final long lastReceivedMs) {
+ this.lastReceivedMs = lastReceivedMs;
+ }
+
Review Comment:
This is not a problem in all managers because they end up calling `onSuccessfulAttempt` or `onFailedAttempt` in all cases, with clear paths for success, fail, or retry with backoff. The issue was with the heartbeat Mgr because it has the particularity that it wants to "retry" immediately on some errors, so it does not need the `onFailedAttempt` backoff strategy. Ex. fence (should just rejoin right after), not_coordinator (should discover the new one and retry when found). Then it does has some errors where it wants to backoff, like the coord_load_in_progress.
There, btw, was the answer to your 2nd point. The `onSuccessfulAttempt` and `onFailedAttempt` do update the last received, but it's more than that (they set values for backoff and attempts), so not exactly what's needed in this HB case, that we want to record the response time, and then apply backoff strategy on some errors (still errors, so not a successful attempt either)
Makes sense? I get the point that we could change how other managers use the success and fail and update the receive time with the new func separately, but their path seemed clear, as it's always success or fail (if retry then always with backoff)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #15311:
URL: https://github.com/apache/kafka/pull/15311#issuecomment-1930740037
Hey @dajac , I updated this to ensure we record a failed attempt for all errors in HB. That will effectively update the received time and backoff, with the ability to skip backoff (0 backoff) in some specific errors where the next HB depends on other conditions (coordinator discovered, assignment released) and we won't add any extra backoff to that.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #15311:
URL: https://github.com/apache/kafka/pull/15311#discussion_r1476536785
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##########
@@ -106,6 +106,13 @@ public void onSendAttempt(final long currentTimeMs) {
this.lastSentMs = currentTimeMs;
}
+ /**
+ * Update the lastReceivedTime in milliseconds, indicating that a response has been received.
+ */
+ public void updateLastReceivedTime(final long lastReceivedMs) {
+ this.lastReceivedMs = lastReceivedMs;
+ }
+
Review Comment:
We have `onSuccessfulAttempt()` and `onFailedAttempt()` that also update the `lastReceivedMs`. Why do we not want to just use those?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15311:
URL: https://github.com/apache/kafka/pull/15311#discussion_r1483161952
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##########
@@ -132,6 +132,16 @@ public void onFailedAttempt(final long currentTimeMs) {
this.numAttempts++;
}
+ /**
+ * Set backoff and number of attempts to 0. This will ensure that the request is sent out
+ * again right away. Expected to be used when receiving errors responses that lead to a new
+ * request that can be sent without backing off (ex. member rejoining after fencing)
+ */
+ public void resetBackoff() {
Review Comment:
I just realized that we already have `reset` which is used for a similar purpose. Could we reuse it?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##########
@@ -132,6 +132,16 @@ public void onFailedAttempt(final long currentTimeMs) {
this.numAttempts++;
}
+ /**
+ * Set backoff and number of attempts to 0. This will ensure that the request is sent out
+ * again right away. Expected to be used when receiving errors responses that lead to a new
+ * request that can be sent without backing off (ex. member rejoining after fencing)
+ */
+ public void resetBackoff() {
+ this.backoffMs = 0;
Review Comment:
Should we use `exponentialBackoff.backoff(0)` as initial value? This seems to be what we do on successful requests and in `reset`.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #15311:
URL: https://github.com/apache/kafka/pull/15311#discussion_r1476728141
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##########
@@ -106,6 +106,13 @@ public void onSendAttempt(final long currentTimeMs) {
this.lastSentMs = currentTimeMs;
}
+ /**
+ * Update the lastReceivedTime in milliseconds, indicating that a response has been received.
+ */
+ public void updateLastReceivedTime(final long lastReceivedMs) {
+ this.lastReceivedMs = lastReceivedMs;
+ }
+
Review Comment:
This is not a problem in all managers because they end up calling `onSuccessfulAttempt` or `onFailedAttempt` in all cases, with clear paths for success, fail, or retry with backoff. The issue was with the heartbeat Mgr because it has the particularity that it wants to "retry" immediately on some errors, so it does not need the `onFailedAttempt` backoff strategy. Ex. fence (should just rejoin right after), not_coordinator (should discover the new one and retry when found). Then it does has some errors where it wants to backoff, like the coord_load_in_progress.
There, btw, was the answer to your 2nd point. The `onSuccessfulAttempt` and `onFailedAttempt` do update the last received, but it's more than that (they set values for backoff and attempts), so not exactly what's needed in this HB case, that we want to record the response time, and then apply backoff strategy on some errors (still errors, so not a successful attempt either)
Makes sense?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #15311:
URL: https://github.com/apache/kafka/pull/15311#discussion_r1483072035
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##########
@@ -127,8 +127,16 @@ public void onSuccessfulAttempt(final long currentTimeMs) {
* @param currentTimeMs Current time in milliseconds
*/
public void onFailedAttempt(final long currentTimeMs) {
+ onFailedAttempt(currentTimeMs, false);
+ }
+
+ public void onFailedAttempt(final long currentTimeMs, final boolean skipBackoff) {
this.lastReceivedMs = currentTimeMs;
- this.backoffMs = exponentialBackoff.backoff(numAttempts);
+ if (skipBackoff) {
+ this.backoffMs = 0;
+ } else {
+ this.backoffMs = exponentialBackoff.backoff(numAttempts);
+ }
this.numAttempts++;
Review Comment:
I did consider using the `onSuccessfulAttempt`, but did not like it too much because even though it does what we need, it did not seem conceptually right since it's not exactly a successful case when we get those errors, and that could maybe lead to trouble in the future if we start taking some more actions on the onSuccessful path I expect. I did not think about the`resetBackoff` one, and I actually like it, will go for it and include the `numAttempts` in the reset, makes sense.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #15311:
URL: https://github.com/apache/kafka/pull/15311#discussion_r1483184589
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##########
@@ -132,6 +132,16 @@ public void onFailedAttempt(final long currentTimeMs) {
this.numAttempts++;
}
+ /**
+ * Set backoff and number of attempts to 0. This will ensure that the request is sent out
+ * again right away. Expected to be used when receiving errors responses that lead to a new
+ * request that can be sent without backing off (ex. member rejoining after fencing)
+ */
+ public void resetBackoff() {
+ this.backoffMs = 0;
Review Comment:
sure, got it from the existing reset now.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #15311:
URL: https://github.com/apache/kafka/pull/15311#discussion_r1480305816
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##########
@@ -106,6 +106,13 @@ public void onSendAttempt(final long currentTimeMs) {
this.lastSentMs = currentTimeMs;
}
+ /**
+ * Update the lastReceivedTime in milliseconds, indicating that a response has been received.
+ */
+ public void updateLastReceivedTime(final long lastReceivedMs) {
+ this.lastReceivedMs = lastReceivedMs;
Review Comment:
The implementation of exponential backoff for KIP-580 introduced the idea of "equivalent responses". The idea is that exponential backoff applies when metadata responses are "equivalent", but when things are progressing, the backoff is avoided. I think a similar principle could be helpful here. Essentially, if "progress" is being made such as a coordinator change, do not apply the backoff.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #15311:
URL: https://github.com/apache/kafka/pull/15311#discussion_r1476752511
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##########
@@ -106,6 +106,13 @@ public void onSendAttempt(final long currentTimeMs) {
this.lastSentMs = currentTimeMs;
}
+ /**
+ * Update the lastReceivedTime in milliseconds, indicating that a response has been received.
+ */
+ public void updateLastReceivedTime(final long lastReceivedMs) {
+ this.lastReceivedMs = lastReceivedMs;
+ }
+
Review Comment:
Thanks for the explanation, @lianetm!
Could you add a comment to the `updateLastReceivedTime()` method documentation so that we can minimize confusion about when to use `updateLastReceivedTime()` vs. `onSuccessfulAttempt()`/`onFailedAttempt()`?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #15311:
URL: https://github.com/apache/kafka/pull/15311#discussion_r1480240755
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##########
@@ -106,6 +106,13 @@ public void onSendAttempt(final long currentTimeMs) {
this.lastSentMs = currentTimeMs;
}
+ /**
+ * Update the lastReceivedTime in milliseconds, indicating that a response has been received.
+ */
+ public void updateLastReceivedTime(final long lastReceivedMs) {
+ this.lastReceivedMs = lastReceivedMs;
Review Comment:
Thanks for the feedback. Agree with @dajac's point that we might be carrying on with a non-desired backoff by leaving it unchanged in this situation, so we definitely need to be setting one, and as @AndrewJSchofield said, all errors would update the last receive time and apply "some" backoff.
That leads to deciding which one, and my intention was 0 backoff in a subset of errors. As I see it, there are some specific errors in HB where we don't just retry the same request, but rather do a different one, so it makes sense to skip backoff as an optimization, ex.
1. HB to rejoin as new member after fence error. We would send it right away after the fence, skipping backoff
2. HB to a new coordinator after not_coordinator error. We would send it as soon as the new coordinator is discovered, without applying any backoff.
Makes sense? I will update to make sure we always set a backoff, but with custom 0 backoff for these cases as the initial intention was, if it makes sense.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15311:
URL: https://github.com/apache/kafka/pull/15311#discussion_r1481205214
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##########
@@ -127,8 +127,16 @@ public void onSuccessfulAttempt(final long currentTimeMs) {
* @param currentTimeMs Current time in milliseconds
*/
public void onFailedAttempt(final long currentTimeMs) {
+ onFailedAttempt(currentTimeMs, false);
+ }
+
+ public void onFailedAttempt(final long currentTimeMs, final boolean skipBackoff) {
this.lastReceivedMs = currentTimeMs;
- this.backoffMs = exponentialBackoff.backoff(numAttempts);
+ if (skipBackoff) {
+ this.backoffMs = 0;
+ } else {
+ this.backoffMs = exponentialBackoff.backoff(numAttempts);
+ }
this.numAttempts++;
Review Comment:
Don't we need to also skip incrementing `numAttempts`? Conceptually, I would say so but I may have missed something.
I was also wondering whether it would make sense to keep `onFailedAttempt` as it was before and to always call it on failures and to add a `resetBackoff` method that we could call in the places where we don't want to backoff. `resetBackoff` would set both `backoffMs` and `numAttempts` to zero. Another alternative would have been to call `onSuccessfulAttempt` for the cases where we don't want to retry with a backoff. Have you already considered those?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #15311:
URL: https://github.com/apache/kafka/pull/15311
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #15311:
URL: https://github.com/apache/kafka/pull/15311#discussion_r1484254595
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -397,13 +400,17 @@ private void onErrorResponse(final ConsumerGroupHeartbeatResponse response,
membershipManager.memberId(), membershipManager.memberEpoch());
logInfo(message, response, currentTimeMs);
membershipManager.transitionToFenced();
+ // Skip backoff so that a next HB to rejoin is sent as soon as the fenced member releases it assignment
Review Comment:
nit: "its" assignment. And below also
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #15311:
URL: https://github.com/apache/kafka/pull/15311#discussion_r1479506790
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##########
@@ -106,6 +106,13 @@ public void onSendAttempt(final long currentTimeMs) {
this.lastSentMs = currentTimeMs;
}
+ /**
+ * Update the lastReceivedTime in milliseconds, indicating that a response has been received.
+ */
+ public void updateLastReceivedTime(final long lastReceivedMs) {
+ this.lastReceivedMs = lastReceivedMs;
Review Comment:
You're probably right @dajac. Should we be applying exponential backoff for heartbeats? Does it only apply to a subset of errors? As a starting position, I would have said that any response or error updates the last receive time, and we would apply exponential backoff in all cases.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #15311:
URL: https://github.com/apache/kafka/pull/15311#issuecomment-1924313236
Hey @dajac , this fixes the issue we were discussing about the 2 console consumers fenced but never rejoining. This same bug was surely behind the other issue, of consumer not rejoining after re-discovering a coordinator.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #15311:
URL: https://github.com/apache/kafka/pull/15311#issuecomment-1934281383
Thanks for the feedback @dajac! Comments addressed
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16215; KAFKA-16178: Fix member not rejoining after error [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #15311:
URL: https://github.com/apache/kafka/pull/15311#issuecomment-1934860453
Build broken with unrelated scala compilation failures being fixed already. I'll rebase as soon as the fix https://github.com/apache/kafka/pull/15343 goes in.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org