Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
kirktrue commented on PR #15723: URL: https://github.com/apache/kafka/pull/15723#issuecomment-2086027819 Thanks everyone for the reviews and @lucasbru for the merge! -- 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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
philipnee commented on PR #15723: URL: https://github.com/apache/kafka/pull/15723#issuecomment-2085841809 Hey sorry for the delay, the changes look good to me. -- 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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
lucasbru merged PR #15723: URL: https://github.com/apache/kafka/pull/15723 -- 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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
lucasbru commented on PR #15723: URL: https://github.com/apache/kafka/pull/15723#issuecomment-2084640215 @philipnee Since the unresolved conversation is only around an internal method name, I'm merging this. Feel free to ask for a follow-up PR -- 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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
kirktrue commented on code in PR #15723:
URL: https://github.com/apache/kafka/pull/15723#discussion_r1583450102
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##
@@ -98,12 +93,11 @@ public boolean canSendRequest(final long currentTimeMs) {
* is a request in-flight.
*/
public boolean requestInFlight() {
-return this.lastSentMs > -1 && this.lastReceivedMs < this.lastSentMs;
+return requestInFlight;
Review Comment:
@philipnee—let us know if you're OK to leave the method name as is. Thanks!
--
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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
kirktrue commented on code in PR #15723:
URL: https://github.com/apache/kafka/pull/15723#discussion_r1579862658
##
clients/src/test/java/org/apache/kafka/clients/consumer/internals/RequestStateTest.java:
##
@@ -48,4 +50,40 @@ public void testRequestStateSimple() {
state.reset();
assertTrue(state.canSendRequest(200));
}
+
+@Test
+public void testTrackInflightOnSuccessfulAttempt() {
+testTrackInflight(RequestState::onSuccessfulAttempt);
+}
+
+@Test
+public void testTrackInflightOnFailedAttempt() {
+testTrackInflight(RequestState::onFailedAttempt);
+}
+
+private void testTrackInflight(BiConsumer
onCompletedAttempt) {
+RequestState state = new RequestState(
+new LogContext(),
+this.getClass().getSimpleName(),
+100,
+2,
+1000,
+0);
+
+// This is just being paranoid...
+assertFalse(state.requestInFlight());
+
+// When we've sent a request, the flag should update from false to
true.
+state.onSendAttempt();
+assertTrue(state.requestInFlight());
+
+// Now we've received the response.
+onCompletedAttempt.accept(state, 236);
+
+// When we've sent a second request with THE SAME TIMESTAMP as the
previous response,
Review Comment:
I added back the timestamp so we could use it in the `lastSentMs` value for
debugging. So the comment should make sense again.
--
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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
kirktrue commented on code in PR #15723:
URL: https://github.com/apache/kafka/pull/15723#discussion_r1579725251
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##
@@ -98,12 +93,11 @@ public boolean canSendRequest(final long currentTimeMs) {
* is a request in-flight.
*/
public boolean requestInFlight() {
-return this.lastSentMs > -1 && this.lastReceivedMs < this.lastSentMs;
+return requestInFlight;
Review Comment:
@philipnee—I didn't make any changes to the name as `requestInFlight` was
the existing method name. Are you OK to leave this for 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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
lucasbru commented on PR #15723: URL: https://github.com/apache/kafka/pull/15723#issuecomment-2076731964 Still looking good to me. I was waiting for the comments from philipp to be commented on / addressed. Do you want to skip ahead an merge? -- 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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
kirktrue commented on PR #15723: URL: https://github.com/apache/kafka/pull/15723#issuecomment-2071054599 @lucasbru, @lianetm, @philipnee—I have re-introduced the `lastSentMs` instance variable, not as a driver for logic, but for informational purposes in the logs. This is ready for re-review. Thanks! -- 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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
philipnee commented on code in PR #15723:
URL: https://github.com/apache/kafka/pull/15723#discussion_r1571359996
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##
@@ -98,12 +93,11 @@ public boolean canSendRequest(final long currentTimeMs) {
* is a request in-flight.
*/
public boolean requestInFlight() {
-return this.lastSentMs > -1 && this.lastReceivedMs < this.lastSentMs;
+return requestInFlight;
Review Comment:
sorry - I meant to say `hasInflightRequest`. Either way is fine.
--
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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
lucasbru commented on code in PR #15723:
URL: https://github.com/apache/kafka/pull/15723#discussion_r1571335784
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##
@@ -98,12 +93,11 @@ public boolean canSendRequest(final long currentTimeMs) {
* is a request in-flight.
*/
public boolean requestInFlight() {
-return this.lastSentMs > -1 && this.lastReceivedMs < this.lastSentMs;
+return requestInFlight;
Review Comment:
`inflightRequested` seems to change the meaning. Who requested an inflight?
In my head it's clearly the noun "request", but if you are bothered, may I
suggest "isRequestInFlight"?
--
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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
philipnee commented on code in PR #15723:
URL: https://github.com/apache/kafka/pull/15723#discussion_r1570988398
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##
@@ -98,12 +93,11 @@ public boolean canSendRequest(final long currentTimeMs) {
* is a request in-flight.
*/
public boolean requestInFlight() {
-return this.lastSentMs > -1 && this.lastReceivedMs < this.lastSentMs;
+return requestInFlight;
Review Comment:
do you think it would make sense to rename this to: `inflightRequested`?
`requestInflight` sounds like a verb/action to me.
--
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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
philipnee commented on code in PR #15723:
URL: https://github.com/apache/kafka/pull/15723#discussion_r1570986831
##
clients/src/test/java/org/apache/kafka/clients/consumer/internals/RequestStateTest.java:
##
@@ -48,4 +50,40 @@ public void testRequestStateSimple() {
state.reset();
assertTrue(state.canSendRequest(200));
}
+
+@Test
+public void testTrackInflightOnSuccessfulAttempt() {
+testTrackInflight(RequestState::onSuccessfulAttempt);
+}
+
+@Test
+public void testTrackInflightOnFailedAttempt() {
+testTrackInflight(RequestState::onFailedAttempt);
+}
+
+private void testTrackInflight(BiConsumer
onCompletedAttempt) {
+RequestState state = new RequestState(
+new LogContext(),
+this.getClass().getSimpleName(),
+100,
+2,
+1000,
+0);
+
+// This is just being paranoid...
+assertFalse(state.requestInFlight());
+
+// When we've sent a request, the flag should update from false to
true.
+state.onSendAttempt();
+assertTrue(state.requestInFlight());
+
+// Now we've received the response.
+onCompletedAttempt.accept(state, 236);
+
+// When we've sent a second request with THE SAME TIMESTAMP as the
previous response,
Review Comment:
The requestInFlight doesn't update the timestamp internally. I wonder if we
should just say "making consecutive requests" instead of mentioning the
timestamp.
--
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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
lucasbru commented on PR #15723: URL: https://github.com/apache/kafka/pull/15723#issuecomment-2063300823 @lianetm did you want to make another pass or good to go? -- 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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
philipnee commented on code in PR #15723:
URL: https://github.com/apache/kafka/pull/15723#discussion_r1569554430
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java:
##
@@ -98,15 +98,15 @@ public NetworkClientDelegate.PollResult poll(final long
currentTimeMs) {
return EMPTY;
if (coordinatorRequestState.canSendRequest(currentTimeMs)) {
-NetworkClientDelegate.UnsentRequest request =
makeFindCoordinatorRequest(currentTimeMs);
+NetworkClientDelegate.UnsentRequest request =
makeFindCoordinatorRequest();
return new NetworkClientDelegate.PollResult(request);
}
return new
NetworkClientDelegate.PollResult(coordinatorRequestState.remainingBackoffMs(currentTimeMs));
}
-NetworkClientDelegate.UnsentRequest makeFindCoordinatorRequest(final long
currentTimeMs) {
-coordinatorRequestState.onSendAttempt(currentTimeMs);
+NetworkClientDelegate.UnsentRequest makeFindCoordinatorRequest() {
+coordinatorRequestState.onSendAttempt();
Review Comment:
Thanks. This will fix some of the no-backoff issues.
--
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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
kirktrue commented on PR #15723: URL: https://github.com/apache/kafka/pull/15723#issuecomment-2059706910 @lianetm & @lucasbru—I've addressed your comments, and this is ready for another review. Thanks! -- 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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
kirktrue commented on code in PR #15723:
URL: https://github.com/apache/kafka/pull/15723#discussion_r1567791850
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##
@@ -66,6 +67,7 @@ public RequestState(final LogContext logContext,
* and the backoff is restored to its minimal configuration.
*/
public void reset() {
+this.requestInFlight = false;
this.lastSentMs = -1;
Review Comment:
I've removed `lastSentMs` as it is no longer needed.
--
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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
kirktrue commented on code in PR #15723:
URL: https://github.com/apache/kafka/pull/15723#discussion_r1567722373
##
clients/src/test/java/org/apache/kafka/clients/consumer/internals/RequestStateTest.java:
##
@@ -48,4 +50,51 @@ public void testRequestStateSimple() {
state.reset();
assertTrue(state.canSendRequest(200));
}
+
+@Test
+public void testTrackInflightOnSuccessfulAttempt() {
+testTrackInflight(RequestState::onSuccessfulAttempt);
+}
+
+@Test
+public void testTrackInflightOnFailedAttempt() {
+testTrackInflight(RequestState::onFailedAttempt);
+}
+
+/**
+ * In some cases, the network layer is very fast and can send out
a second request within the same
+ * millisecond timestamp as receiving the first response.
+ *
+ *
+ *
+ * The previous logic for tracking inflight status used timestamps: if the
timestamp from the last received
+ * response was less than the timestamp from the last sent
request, we'd interpret that as having an
+ * inflight request. However, this approach would incorrectly return
false from
+ * {@link RequestState#requestInFlight()} if the two timestamps were
equal.
+ */
Review Comment:
I figure the existing PR description covers the basics. I'll just remove the
test comment wholesale.
--
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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
lucasbru commented on code in PR #15723:
URL: https://github.com/apache/kafka/pull/15723#discussion_r1567484668
##
clients/src/test/java/org/apache/kafka/clients/consumer/internals/RequestStateTest.java:
##
@@ -48,4 +50,51 @@ public void testRequestStateSimple() {
state.reset();
assertTrue(state.canSendRequest(200));
}
+
+@Test
+public void testTrackInflightOnSuccessfulAttempt() {
+testTrackInflight(RequestState::onSuccessfulAttempt);
+}
+
+@Test
+public void testTrackInflightOnFailedAttempt() {
+testTrackInflight(RequestState::onFailedAttempt);
+}
+
+/**
+ * In some cases, the network layer is very fast and can send out
a second request within the same
+ * millisecond timestamp as receiving the first response.
+ *
+ *
+ *
+ * The previous logic for tracking inflight status used timestamps: if the
timestamp from the last received
+ * response was less than the timestamp from the last sent
request, we'd interpret that as having an
+ * inflight request. However, this approach would incorrectly return
false from
+ * {@link RequestState#requestInFlight()} if the two timestamps were
equal.
+ */
Review Comment:
yeah I also noticed this. I think this parapraph fits best in the commit
message / PR description.
--
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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
lianetm commented on PR #15723: URL: https://github.com/apache/kafka/pull/15723#issuecomment-2059130202 Thanks for the changes @kirktrue , LGTM. Just thinking out loud to share a concern and reasoning about it. The flag based approach always makes me think if we would break managers out there that could be using the `canSend` (returns false based on the flag), but not using the other funcs that flip the flag when a response is received/not-received/succeeds/fails. My reasoning then was that given than the previous approach was based in numeric variables that were being set on the same places where the flag is flipped now I would say we can expect that no managers will be affected, so we should be good with it I expect. Thanks for the changes! -- 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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
lianetm commented on code in PR #15723:
URL: https://github.com/apache/kafka/pull/15723#discussion_r1567368858
##
clients/src/test/java/org/apache/kafka/clients/consumer/internals/RequestStateTest.java:
##
@@ -48,4 +50,51 @@ public void testRequestStateSimple() {
state.reset();
assertTrue(state.canSendRequest(200));
}
+
+@Test
+public void testTrackInflightOnSuccessfulAttempt() {
+testTrackInflight(RequestState::onSuccessfulAttempt);
+}
+
+@Test
+public void testTrackInflightOnFailedAttempt() {
+testTrackInflight(RequestState::onFailedAttempt);
+}
+
+/**
+ * In some cases, the network layer is very fast and can send out
a second request within the same
+ * millisecond timestamp as receiving the first response.
+ *
+ *
+ *
+ * The previous logic for tracking inflight status used timestamps: if the
timestamp from the last received
+ * response was less than the timestamp from the last sent
request, we'd interpret that as having an
+ * inflight request. However, this approach would incorrectly return
false from
+ * {@link RequestState#requestInFlight()} if the two timestamps were
equal.
+ */
Review Comment:
nit: Do we think it's helpful to explain so much about a "previous logic"
that was wrong here? I find it can be confusing for others. Also here we're
actually just testing the approach of checking inflights based on a flag (the
why we chose that approach makes more sense to me in the `requestInflight()`
func itself)
--
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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
lucasbru commented on code in PR #15723:
URL: https://github.com/apache/kafka/pull/15723#discussion_r1567025166
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestState.java:
##
@@ -66,6 +67,7 @@ public RequestState(final LogContext logContext,
* and the backoff is restored to its minimal configuration.
*/
public void reset() {
+this.requestInFlight = false;
this.lastSentMs = -1;
Review Comment:
what's the purpose of `lastSentMs` now? Can we remove it?
--
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
Re: [PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
kirktrue commented on PR #15723: URL: https://github.com/apache/kafka/pull/15723#issuecomment-2057988468 @lucasbru—please kindly take a look at this issue that's occasionally causing duplicate heartbeat requests. cc @lianetm @philipnee -- 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
[PR] KAFKA-16555: Consumer's RequestState has incorrect logic to determine if inflight [kafka]
kirktrue opened a new pull request, #15723: URL: https://github.com/apache/kafka/pull/15723 In some cases, the network layer is _very_ fast and can send out multiple requests within the same millisecond timestamp. The previous logic for tracking inflight status used timestamps: if the timestamp from the last received response was less than the timestamp from the last sent request, we'd interpret that as having an inflight request. However, this approach would incorrectly return `false` from `RequestState.requestInFlight()` if the two timestamps were _equal_. One result of this faulty logic is that in such cases, the consumer would accidentally send multiple heartbeat requests to the consumer group coordinator. The consumer group coordinator would interpret these requests as 'join group' requests and create members for each request. Therefore, the coordinator was under the false understanding that there were more members in the group than there really were. Consequently, if your luck was _really_ bad, the coordinator might assign partitions to one of the duplicate members. Those partitions would be assigned to a phantom consumer that was not reading any data, and this led to flaky tests. The implementation in `RequestState` has a stupid simple flag that is set in `onSendAttempt` and cleared in `onSuccessfulAttempt`, `onFailedAttempt`, and `reset`. A new unit test has been added and this has been tested against all of the consumer unit and integration tests, and has removed all known occurrences of phantom consumer group members in the system tests. ### Committer Checklist (excluded from commit message) - [ ] Verify design and implementation - [ ] Verify test coverage and CI build status - [ ] Verify documentation (including upgrade notes) -- 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
