showuon commented on a change in pull request #11451:
URL: https://github.com/apache/kafka/pull/11451#discussion_r748032208
##########
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<TopicPartition> 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:
We should use the `currentGeneration` snapshot to do the check.
Otherwise, user might see the unexpected callback got called when comparing the
log. Also, we should compare with `equals` here.
##########
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:
We only care about the `protocolName` here, so only check `protocolName`
not null
##########
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<TopicPartition> revokedPartitions;
- if (generation == Generation.NO_GENERATION.generationId &&
+ if (generation == Generation.NO_GENERATION.generationId ||
memberId.equals(Generation.NO_GENERATION.memberId)) {
revokedPartitions = new
HashSet<>(subscriptions.assignedPartitions());
if (!revokedPartitions.isEmpty()) {
- log.info("Giving away all assigned partitions as lost since
generation has been reset," +
- "indicating that consumer is no longer part of the group");
+ log.info("Giving away all assigned partitions as lost since
generation/memberID has been reset," +
+ "indicating that consumer is in old state or no longer
part of the group");
Review comment:
After this change, either no generation ID or no member ID, we'll clear
all their ownedPartitions since they are out-of-date.
##########
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
##########
@@ -684,7 +686,7 @@ public void handle(JoinGroupResponse joinResponse,
RequestFuture<ByteBuffer> fut
.setGenerationId(generation.generationId)
.setAssignments(Collections.emptyList())
);
- log.debug("Sending follower SyncGroup to coordinator {} at generation
{}: {}", this.coordinator, this.generation, requestBuilder);
+ log.debug("Sending follower SyncGroup to coordinator {}: {}",
this.coordinator, requestBuilder);
Review comment:
`requestBuilder` already log the `generation` info. So, remove it.
Before this change, log is like below (with duplicated generation info
output):
```
Sending leader SyncGroup to coordinator localhost:55644 (id: 2147483647
rack: null) at generation Generation{generationId=2,
memberId='consumer-test.group-15-e79d4f58-f8cc-4f98-897d-f711fb3385d8',
protocol='range'}: SyncGroupRequestData(groupId='test.group', generationId=2,
memberId='consumer-test.group-15-e79d4f58-f8cc-4f98-897d-f711fb3385d8',
groupInstanceId=null, protocolType='consumer', protocolName='range',
assignments=[SyncGroupRequestAssignment(memberId='consumer-test.group-16-ba76c722-d177-4cad-8251-2e7ece935e7d',
assignment=[0, 1, 0, 0, 0, 0, -1, -1, -1, -1]),
SyncGroupRequestAssignment(memberId='consumer-test.group-15-e79d4f58-f8cc-4f98-897d-f711fb3385d8',
assignment=[0, 1, 0, 0, 0, 1, 0, 3, 102, 111, 111, 0, 0, 0, 1, 0, 0, 0, 0, -1,
-1, -1, -1])])
```
--
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
