[GitHub] [kafka] showuon commented on a change in pull request #11451: KAFKA-13419: Only reset generation ID when ILLEGAL_GENERATION error
showuon commented on a change in pull request #11451:
URL: https://github.com/apache/kafka/pull/11451#discussion_r771181554
##
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
##
@@ -705,13 +705,13 @@ protected void onJoinPrepare(int generation, String
memberId) {
// so that users can still access the previously owned partitions to
commit offsets etc.
Exception exception = null;
final Set revokedPartitions;
-if (generation == Generation.NO_GENERATION.generationId &&
+if (generation == Generation.NO_GENERATION.generationId ||
Review comment:
Yes, I agree that it doesn't hurt to have it. Thank you.
--
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
[GitHub] [kafka] showuon commented on a change in pull request #11451: KAFKA-13419: Only reset generation ID when ILLEGAL_GENERATION error
showuon commented on a change in pull request #11451:
URL: https://github.com/apache/kafka/pull/11451#discussion_r771181062
##
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
##
@@ -755,18 +755,17 @@ protected void onJoinPrepare(int generation, String
memberId) {
@Override
public void onLeavePrepare() {
-// Save the current Generation and use that to get the memberId, as
the hb thread can change it at any time
+// Save the current Generation, as the hb thread can change it at any
time
final Generation currentGeneration = generation();
-final String memberId = currentGeneration.memberId;
-log.debug("Executing onLeavePrepare with generation {} and memberId
{}", currentGeneration, memberId);
+log.debug("Executing onLeavePrepare with generation {}",
currentGeneration);
// we should reset assignment and trigger the callback before leaving
group
Set droppedPartitions = new
HashSet<>(subscriptions.assignedPartitions());
if (subscriptions.hasAutoAssignedPartitions() &&
!droppedPartitions.isEmpty()) {
final Exception e;
-if (generation() == Generation.NO_GENERATION ||
rebalanceInProgress()) {
+if (currentGeneration.equals(Generation.NO_GENERATION) ||
rebalanceInProgress()) {
Review comment:
David, you're right! I was focusing on fixing the `==` error here. Yes,
we should be consistent with `onJoinPrepare`
[here](https://github.com/apache/kafka/blob/1beb3bd5160c9cc950a541b02684f2fd53ea8da2/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java#L708-L718),
to invoke PartitionsLost when
```
generation == Generation.NO_GENERATION.generationId ||
memberId.equals(Generation.NO_GENERATION.memberId)
```
Otherwise, invoke PartitionsRevoked.
I'll update it later. Thank you.
--
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
[GitHub] [kafka] showuon commented on a change in pull request #11451: KAFKA-13419: Only reset generation ID when ILLEGAL_GENERATION error
showuon commented on a change in pull request #11451:
URL: https://github.com/apache/kafka/pull/11451#discussion_r752901904
##
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
##
@@ -443,7 +443,9 @@ boolean joinGroupIfNeeded(final Timer timer) {
stateSnapshot = this.state;
}
-if (!generationSnapshot.equals(Generation.NO_GENERATION) &&
stateSnapshot == MemberState.STABLE) {
+if ((generationSnapshot.generationId !=
Generation.NO_GENERATION.generationId ||
+
!generationSnapshot.memberId.equals(Generation.NO_GENERATION.memberId)) &&
+stateSnapshot == MemberState.STABLE) {
Review comment:
You are right! Updated.
--
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
[GitHub] [kafka] showuon commented on a change in pull request #11451: KAFKA-13419: Only reset generation ID when ILLEGAL_GENERATION error
showuon commented on a change in pull request #11451:
URL: https://github.com/apache/kafka/pull/11451#discussion_r752900442
##
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
##
@@ -746,8 +748,8 @@ public void handle(SyncGroupResponse syncResponse,
sensors.syncSensor.record(response.requestLatencyMs());
synchronized (AbstractCoordinator.this) {
-if (!generation.equals(Generation.NO_GENERATION) &&
state == MemberState.COMPLETING_REBALANCE) {
-// check protocol name only if the generation is
not reset
+if (generation.protocolName != null && state ==
MemberState.COMPLETING_REBALANCE) {
+// check protocol name only if the generation is
not reset (protocol name is not null)
Review comment:
You are right, updated and explained below. 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
[GitHub] [kafka] showuon commented on a change in pull request #11451: KAFKA-13419: Only reset generation ID when ILLEGAL_GENERATION error
showuon commented on a change in pull request #11451:
URL: https://github.com/apache/kafka/pull/11451#discussion_r751200340
##
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
##
@@ -755,18 +755,17 @@ protected void onJoinPrepare(int generation, String
memberId) {
@Override
public void onLeavePrepare() {
-// Save the current Generation and use that to get the memberId, as
the hb thread can change it at any time
+// Save the current Generation, as the hb thread can change it at any
time
final Generation currentGeneration = generation();
-final String memberId = currentGeneration.memberId;
-log.debug("Executing onLeavePrepare with generation {} and memberId
{}", currentGeneration, memberId);
Review comment:
It's because the `memberId` info is already included in the `generation`
info. This is the log output currently:
`Executing onLeavePrepare with generation Generation{generationId=1,
memberId='consumer1', protocol='range'} and memberId consumer1`
Sorry, I should have mentioned it to make it clear.
--
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
