You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "frankvicky (via GitHub)" <gi...@apache.org> on 2024/10/19 13:04:55 UTC
[PR] KAFKA-17116: New consumer may not send effective leave group if member ID received after close [kafka]
frankvicky opened a new pull request, #17549:
URL: https://github.com/apache/kafka/pull/17549
JIRA: [KAFKA-17116](https://issues.apache.org/jira/browse/KAFKA-17116)
This PR is aim to deliver [KIP-1082](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1082%3A+Require+Client-Generated+IDs+over+the+ConsumerGroupHeartbeat+RPC) and solves KAFKA-17716.
### 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1807323559
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMembershipManager.java:
##########
@@ -229,16 +227,6 @@ public void onHeartbeatSuccess(ConsumerGroupHeartbeatResponse response) {
return;
}
- // Update the group member id label in the client telemetry reporter if the member id has
- // changed. Initially the member id is empty, and it is updated when the member joins the
- // group. This is done here to avoid updating the label on every heartbeat response. Also
- // check if the member id is null, as the schema defines it as nullable.
- if (responseData.memberId() != null && !responseData.memberId().equals(memberId)) {
- clientTelemetryReporter.ifPresent(reporter -> reporter.updateMetricsLabels(
- Collections.singletonMap(ClientTelemetryProvider.GROUP_MEMBER_ID, responseData.memberId())));
- }
Review Comment:
see https://github.com/apache/kafka/pull/17549/files#r1807323514
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareMembershipManager.java:
##########
@@ -155,16 +153,6 @@ public void onHeartbeatSuccess(ShareGroupHeartbeatResponse response) {
return;
}
- // Update the group member id label in the client telemetry reporter if the member id has
- // changed. Initially the member id is empty, and it is updated when the member joins the
- // group. This is done here to avoid updating the label on every heartbeat response. Also
- // check if the member id is null, as the schema defines it as nullable.
- if (responseData.memberId() != null && !responseData.memberId().equals(memberId)) {
- clientTelemetryReporter.ifPresent(reporter -> reporter.updateMetricsLabels(
- Collections.singletonMap(ClientTelemetryProvider.GROUP_MEMBER_ID, responseData.memberId())));
- }
Review Comment:
see https://github.com/apache/kafka/pull/17549/files#r1807323514
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1808866121
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3199,7 +3201,7 @@ public CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord>
RequestContext context,
ConsumerGroupHeartbeatRequestData request
) throws ApiException {
- throwIfConsumerGroupHeartbeatRequestIsInvalid(request);
+ throwIfConsumerGroupHeartbeatRequestIsInvalid(request, context.apiVersion());
if (request.memberEpoch() == LEAVE_GROUP_MEMBER_EPOCH || request.memberEpoch() == LEAVE_GROUP_STATIC_MEMBER_EPOCH) {
// -1 means that the member wants to leave the group.
Review Comment:
Do we need to update share group logic too?
##########
clients/src/main/resources/common/message/ConsumerGroupHeartbeatRequest.json:
##########
@@ -19,7 +19,8 @@
"listeners": ["zkBroker", "broker"],
"name": "ConsumerGroupHeartbeatRequest",
// Version 1 adds SubscribedTopicRegex (KIP-848).
- "validVersions": "0-1",
+ // Version 2 requires the consumer to generate their own Member ID
+ "validVersions": "0-2",
Review Comment:
We must be careful with this change. Version 1 is actually unstable so if we bump it, it would become stable. This is not what we want. Here, I suggest to not bump and to modify version 1. We can do this because it is not stable yet.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1293,13 +1293,12 @@ private void throwIfNull(
* Validates the request.
*
* @param request The request to validate.
- *
+ * @param apiVersion The version of ConsumerGroupHeartbeat RPC
* @throws InvalidRequestException if the request is not valid.
* @throws UnsupportedAssignorException if the assignor is not supported.
*/
private void throwIfConsumerGroupHeartbeatRequestIsInvalid(
- ConsumerGroupHeartbeatRequestData request
- ) throws InvalidRequestException, UnsupportedAssignorException {
+ ConsumerGroupHeartbeatRequestData request, short apiVersion) throws InvalidRequestException, UnsupportedAssignorException {
Review Comment:
nit: Let's keep the style of the code coherent with the existing code please.
```
private void throwIfConsumerGroupHeartbeatRequestIsInvalid(
ConsumerGroupHeartbeatRequestData request,
short apiVersion
) throws InvalidRequestException, UnsupportedAssignorException {
```
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10225,53 +10251,61 @@ public void testConsumerGroupHeartbeatToClassicGroupFromExistingStaticMember() {
.setClassicMemberMetadata(null)
.build();
- List<CoordinatorRecord> expectedRecords = Arrays.asList(
- // The existing classic group tombstone.
- GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord(groupId),
-
- // Create the new consumer group with the static member.
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedClassicMember),
- GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 0),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId, expectedClassicMember.assignedPartitions()),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentEpochRecord(groupId, 0),
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedClassicMember),
-
- // Remove the static member because the rejoining member replaces it.
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentTombstoneRecord(groupId, memberId),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentTombstoneRecord(groupId, memberId),
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionTombstoneRecord(groupId, memberId),
-
- // Create the new static member.
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedReplacingConsumerMember),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, newMemberId, mkAssignment(mkTopicAssignment(fooTopicId, 0))),
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedReplacingConsumerMember),
-
- // The static member rejoins the new consumer group.
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedFinalConsumerMember),
-
- // The subscription metadata hasn't been updated during the conversion, so a new one is computed.
- GroupCoordinatorRecordHelpers.newConsumerGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
- {
- put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 1));
- }
- }),
+ List<CoordinatorRecord> expectedRecords = new ArrayList<>();
+ // The existing classic group tombstone.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord(groupId));
+
+ // Create the new consumer group with the static member.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedClassicMember));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 0));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId,
+ expectedClassicMember.assignedPartitions()));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentEpochRecord(groupId, 0));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedClassicMember));
+
+ // Remove the static member because the rejoining member replaces it.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentTombstoneRecord(groupId, memberId));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentTombstoneRecord(groupId, memberId));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionTombstoneRecord(groupId, memberId));
+
+ // Create the new static member.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedReplacingConsumerMember));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, newMemberId,
+ mkAssignment(mkTopicAssignment(fooTopicId, 0))));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedReplacingConsumerMember));
+
+ // The static member rejoins the new consumer group.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedFinalConsumerMember));
+
+ // The subscription metadata hasn't been updated during the conversion, so a new one is computed.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupSubscriptionMetadataRecord(groupId,
+ Map.of(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 1))));
+
+ // Newly joining static member bumps the group epoch. A new target assignment is computed.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 1));
+
+ // If the memberId is generated by the consumer itself, the consumer should retain this memberId.
+ // As a result, the record won't contain a new target assignment record.
+ // If the memberId is not consumer-generated, add a new target assignment record to the expected records,
+ // since a different memberId will be considered as a new member.
+ if (!isConsumerGeneratedMemberId) {
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, newMemberId,
+ mkAssignment(mkTopicAssignment(fooTopicId, 0))));
+ }
- // Newly joining static member bumps the group epoch. A new target assignment is computed.
- GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 1),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, newMemberId, mkAssignment(mkTopicAssignment(fooTopicId, 0))),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentEpochRecord(groupId, 1),
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentEpochRecord(groupId, 1));
- // The newly created static member takes the assignment from the existing member.
- // Bump its member epoch and transition to STABLE.
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedFinalConsumerMember)
- );
+ // The newly created static member takes the assignment from the existing member.
+ // Bump its member epoch and transition to STABLE.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedFinalConsumerMember));
assertRecordsEquals(expectedRecords, result.records());
context.assertSessionTimeout(groupId, newMemberId, 45000);
}
- @Test
- public void testConsumerGroupHeartbeatFromExistingClassicStaticMember() {
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testConsumerGroupHeartbeatFromExistingClassicStaticMember(boolean isConsumerGeneratedMemberId) {
Review Comment:
We should perhaps use the last version in this test and generate the member id on the client side.
##########
core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala:
##########
@@ -569,7 +569,7 @@ class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) {
protected def consumerGroupHeartbeat(
groupId: String,
- memberId: String = "",
+ memberId: String = Uuid.randomUuid().toString,
Review Comment:
You can pass the version and use it. See `commitOffset` in this file as an example. We should also update ConsumerGroupHeartbeatRequestTest.scala btw.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java:
##########
@@ -599,13 +599,20 @@ public MemberState consumerGroupMemberState(
.state();
}
+
public CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord> consumerGroupHeartbeat(
ConsumerGroupHeartbeatRequestData request
+ ) {
+ return this.consumerGroupHeartbeat(ApiKeys.CONSUMER_GROUP_HEARTBEAT.latestVersion(), request);
+ }
+
+ public CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord> consumerGroupHeartbeat(
+ short apiVersion, ConsumerGroupHeartbeatRequestData request
Review Comment:
nit1: Let's put apiVersion as the second argument. This is usually what we do.
nit2: Let's put one argument per line to follow the style of the code in this file.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -266,7 +275,9 @@ public String groupId() {
}
/**
- * @return Member ID assigned by the server to this member when it joins the consumer group.
+ * Returns the Member ID that is generated at startup and remains unchanged for the entire lifetime of the process.
+ *
+ * @return Member ID that does not change during the process's lifetime.
Review Comment:
nit: Let's keep the old style to reduce the comment.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10117,8 +10130,9 @@ public void testConsumerGroupHeartbeatWithPreparingRebalanceClassicGroup() throw
assertEquals(group, context.groupMetadataManager.getOrMaybeCreateClassicGroup("group-id", false));
}
- @Test
- public void testConsumerGroupHeartbeatToClassicGroupFromExistingStaticMember() {
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testConsumerGroupHeartbeatToClassicGroupFromExistingStaticMember(boolean isConsumerGeneratedMemberId) {
Review Comment:
Why are we changing those tests in particular?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1317,6 +1316,9 @@ private void throwIfConsumerGroupHeartbeatRequestIsInvalid(
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
+ if (apiVersion >= 2) {
Review Comment:
nit: Let's add a constant to ConsumerGroupHeartbeatRequest for the required version. It will make the code clearer too.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java:
##########
@@ -599,13 +599,20 @@ public MemberState consumerGroupMemberState(
.state();
}
+
Review Comment:
nit: We can remove this empty line.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10225,53 +10251,61 @@ public void testConsumerGroupHeartbeatToClassicGroupFromExistingStaticMember() {
.setClassicMemberMetadata(null)
.build();
- List<CoordinatorRecord> expectedRecords = Arrays.asList(
- // The existing classic group tombstone.
- GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord(groupId),
-
- // Create the new consumer group with the static member.
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedClassicMember),
- GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 0),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId, expectedClassicMember.assignedPartitions()),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentEpochRecord(groupId, 0),
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedClassicMember),
-
- // Remove the static member because the rejoining member replaces it.
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentTombstoneRecord(groupId, memberId),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentTombstoneRecord(groupId, memberId),
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionTombstoneRecord(groupId, memberId),
-
- // Create the new static member.
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedReplacingConsumerMember),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, newMemberId, mkAssignment(mkTopicAssignment(fooTopicId, 0))),
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedReplacingConsumerMember),
-
- // The static member rejoins the new consumer group.
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedFinalConsumerMember),
-
- // The subscription metadata hasn't been updated during the conversion, so a new one is computed.
- GroupCoordinatorRecordHelpers.newConsumerGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
- {
- put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 1));
- }
- }),
+ List<CoordinatorRecord> expectedRecords = new ArrayList<>();
+ // The existing classic group tombstone.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord(groupId));
+
+ // Create the new consumer group with the static member.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedClassicMember));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 0));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId,
+ expectedClassicMember.assignedPartitions()));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentEpochRecord(groupId, 0));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedClassicMember));
+
+ // Remove the static member because the rejoining member replaces it.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentTombstoneRecord(groupId, memberId));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentTombstoneRecord(groupId, memberId));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionTombstoneRecord(groupId, memberId));
+
+ // Create the new static member.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedReplacingConsumerMember));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, newMemberId,
+ mkAssignment(mkTopicAssignment(fooTopicId, 0))));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedReplacingConsumerMember));
+
+ // The static member rejoins the new consumer group.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedFinalConsumerMember));
+
+ // The subscription metadata hasn't been updated during the conversion, so a new one is computed.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupSubscriptionMetadataRecord(groupId,
+ Map.of(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 1))));
+
+ // Newly joining static member bumps the group epoch. A new target assignment is computed.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 1));
+
+ // If the memberId is generated by the consumer itself, the consumer should retain this memberId.
+ // As a result, the record won't contain a new target assignment record.
+ // If the memberId is not consumer-generated, add a new target assignment record to the expected records,
+ // since a different memberId will be considered as a new member.
+ if (!isConsumerGeneratedMemberId) {
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, newMemberId,
+ mkAssignment(mkTopicAssignment(fooTopicId, 0))));
+ }
Review Comment:
Hum... I am not sure about this change. My understanding is that we wanted to test a static member rejoining, assuming after a restart so the member id must be different.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1810581161
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1351,6 +1355,7 @@ private void throwIfShareGroupHeartbeatRequestIsInvalid(
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
+ throwIfEmptyString(request.memberId(), "MemberId can't be empty.");
Review Comment:
Since consumers are now required to generate their own `memberId` and the `ShareGroupHeartbeat RPC` is still under development, it seems feasible to check the `memberId` at the beginning:
```java
private void throwIfShareGroupHeartbeatRequestIsInvalid(
ShareGroupHeartbeatRequestData request
) throws InvalidRequestException, UnsupportedAssignorException {
throwIfEmptyString(request.groupId(), "GroupId can't be empty.");
throwIfEmptyString(request.rackId(), "RackId can't be empty.");
throwIfEmptyString(request.memberId(), "MemberId can't be empty.");
if (request.memberEpoch() == 0) {
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
} else {
throw new InvalidRequestException("MemberEpoch is invalid.");
}
}
Dose this make 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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1822185721
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -198,14 +209,14 @@ public void testConsumerHeartbeatRequestValidation() {
// SubscribedTopicNames must be present and empty in the first request (epoch == 0).
ex = assertThrows(InvalidRequestException.class, () -> context.consumerGroupHeartbeat(
new ConsumerGroupHeartbeatRequestData()
+ .setMemberId(memberId)
.setGroupId("foo")
.setMemberEpoch(0)
.setRebalanceTimeoutMs(5000)
.setTopicPartitions(Collections.emptyList())));
assertEquals("SubscribedTopicNames must be set in first request.", ex.getMessage());
- // MemberId must be non-empty in all requests except for the first one where it
- // could be empty (epoch != 0).
+ // MemberId must be non-empty in all requests
Review Comment:
Yes.
Since we have a similar one in `ConsumerGroupHeartbeatRequestTest`, we could remove 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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1820699580
##########
clients/src/main/resources/common/message/ConsumerGroupHeartbeatResponse.json:
##########
@@ -39,7 +39,7 @@
{ "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The top-level error message, or null if there was no error." },
{ "name": "MemberId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
- "about": "The member id generated by the coordinator. Only provided when the member joins with MemberEpoch == 0." },
+ "about": "The member id is generated by the consumer and provided by the consumer for all requests." },
Review Comment:
hmm... I feel that we don't need to specify the behavior in version `0` since it's an unstable version.
If users are using the 4.0 release, they should have the latest behavior.
> However, in version 1+, it is basically the value received from the consumer?
I think yes.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1809943339
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10225,53 +10251,61 @@ public void testConsumerGroupHeartbeatToClassicGroupFromExistingStaticMember() {
.setClassicMemberMetadata(null)
.build();
- List<CoordinatorRecord> expectedRecords = Arrays.asList(
- // The existing classic group tombstone.
- GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord(groupId),
-
- // Create the new consumer group with the static member.
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedClassicMember),
- GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 0),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId, expectedClassicMember.assignedPartitions()),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentEpochRecord(groupId, 0),
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedClassicMember),
-
- // Remove the static member because the rejoining member replaces it.
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentTombstoneRecord(groupId, memberId),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentTombstoneRecord(groupId, memberId),
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionTombstoneRecord(groupId, memberId),
-
- // Create the new static member.
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedReplacingConsumerMember),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, newMemberId, mkAssignment(mkTopicAssignment(fooTopicId, 0))),
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedReplacingConsumerMember),
-
- // The static member rejoins the new consumer group.
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedFinalConsumerMember),
-
- // The subscription metadata hasn't been updated during the conversion, so a new one is computed.
- GroupCoordinatorRecordHelpers.newConsumerGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
- {
- put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 1));
- }
- }),
+ List<CoordinatorRecord> expectedRecords = new ArrayList<>();
+ // The existing classic group tombstone.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord(groupId));
+
+ // Create the new consumer group with the static member.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedClassicMember));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 0));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId,
+ expectedClassicMember.assignedPartitions()));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentEpochRecord(groupId, 0));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedClassicMember));
+
+ // Remove the static member because the rejoining member replaces it.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentTombstoneRecord(groupId, memberId));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentTombstoneRecord(groupId, memberId));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionTombstoneRecord(groupId, memberId));
+
+ // Create the new static member.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedReplacingConsumerMember));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, newMemberId,
+ mkAssignment(mkTopicAssignment(fooTopicId, 0))));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedReplacingConsumerMember));
+
+ // The static member rejoins the new consumer group.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedFinalConsumerMember));
+
+ // The subscription metadata hasn't been updated during the conversion, so a new one is computed.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupSubscriptionMetadataRecord(groupId,
+ Map.of(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 1))));
+
+ // Newly joining static member bumps the group epoch. A new target assignment is computed.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 1));
+
+ // If the memberId is generated by the consumer itself, the consumer should retain this memberId.
+ // As a result, the record won't contain a new target assignment record.
+ // If the memberId is not consumer-generated, add a new target assignment record to the expected records,
+ // since a different memberId will be considered as a new member.
+ if (!isConsumerGeneratedMemberId) {
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, newMemberId,
+ mkAssignment(mkTopicAssignment(fooTopicId, 0))));
+ }
Review Comment:
Got it.
I will try to simulate this behavior and update the test case accordingly.👍🏼
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1819372748
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10226,48 +10256,48 @@ public void testConsumerGroupHeartbeatToClassicGroupFromExistingStaticMember() {
.build();
List<CoordinatorRecord> expectedRecords = Arrays.asList(
- // The existing classic group tombstone.
- GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord(groupId),
-
- // Create the new consumer group with the static member.
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedClassicMember),
- GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 0),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId, expectedClassicMember.assignedPartitions()),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentEpochRecord(groupId, 0),
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedClassicMember),
-
- // Remove the static member because the rejoining member replaces it.
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentTombstoneRecord(groupId, memberId),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentTombstoneRecord(groupId, memberId),
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionTombstoneRecord(groupId, memberId),
+ // The existing classic group tombstone.
Review Comment:
Sure, I will fix indentation in next commit 😺
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1821035370
##########
core/src/test/scala/unit/kafka/server/ConsumerGroupHeartbeatRequestTest.scala:
##########
@@ -36,54 +35,68 @@ import scala.collection.Map
import scala.jdk.CollectionConverters._
@ExtendWith(value = Array(classOf[ClusterTestExtensions]))
-class ConsumerGroupHeartbeatRequestTest(cluster: ClusterInstance) {
-
- @ClusterTest(
+@ClusterTestDefaults(
types = Array(Type.KRAFT),
serverProperties = Array(
- new ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, value = "classic"),
new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1")
)
+)
+class ConsumerGroupHeartbeatRequestTest(cluster: ClusterInstance) {
+
+ @ClusterTests(value = Array(
+ new ClusterTest(
+ serverProperties = Array(new ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, value = "classic")),
+ tags = Array("0")
+ ),
+ new ClusterTest(
+ serverProperties = Array(new ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, value = "classic")),
+ tags = Array("1")
+ )
+ )
)
Review Comment:
We don't need to parameterize all the existing tests. How about the following? We make all the existing tests use the last version (so they provided the member id). We add one test which ensure that version 0 generate the member id on the server side. We add one test which ensure that versions > 0 fails if the member id is not provided. This should give us enough coverage. What do you think?
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1824460356
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MemberStateListener.java:
##########
@@ -20,18 +20,18 @@
import java.util.Optional;
/**
- * Listener for getting notified of member ID and epoch changes.
+ * Listener for getting notified of member epoch changes.
*/
public interface MemberStateListener {
/**
- * Called whenever member ID or epoch change with new values received from the broker or
+ * Called whenever epoch change with new values received from the broker or
* cleared if the member is not part of the group anymore (when it gets fenced, leaves the
* group or fails).
*
* @param memberEpoch New member epoch received from the broker. Empty if the member is
* not part of the group anymore.
- * @param memberId Current member ID. Empty if the member is not part of the group.
+ * @param memberId Current member ID. It won't change until the process terminated.
Review Comment:
```suggestion
* @param memberId Current member ID. It won't change until the process is terminated.
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MemberStateListener.java:
##########
@@ -20,18 +20,18 @@
import java.util.Optional;
/**
- * Listener for getting notified of member ID and epoch changes.
+ * Listener for getting notified of member epoch changes.
*/
public interface MemberStateListener {
/**
- * Called whenever member ID or epoch change with new values received from the broker or
+ * Called whenever epoch change with new values received from the broker or
Review Comment:
```suggestion
* Called whenever epoch changes with new values received from the broker or
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -95,10 +95,9 @@ public class CommitRequestManager implements RequestManager, MemberStateListener
private Optional<Integer> lastEpochSentOnCommit;
/**
- * Latest member ID and epoch received via the {@link #onMemberEpochUpdated(Optional, Optional)},
- * to be included in the OffsetFetch and OffsetCommit requests if present. This will have
- * the latest values received from the broker, or empty of the member is not part of the
- * group anymore.
+ * The member ID and latest member epoch received via the {@link MemberStateListener#onMemberEpochUpdated(Optional, String)},
+ * to be included in the OffsetFetch and OffsetCommit requests. This will have
+ * the latest memberEpoch values received from the broker.
Review Comment:
```suggestion
* the latest memberEpoch value received from the broker.
```
(or simply remove the "value")
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1810581161
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1351,6 +1355,7 @@ private void throwIfShareGroupHeartbeatRequestIsInvalid(
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
+ throwIfEmptyString(request.memberId(), "MemberId can't be empty.");
Review Comment:
Since consumers are now required to generate their own `memberId` and the `ShareGroupHeartbeat RPC` is still under development, it seems feasible to check the `memberId` at the beginning:
```java
private void throwIfShareGroupHeartbeatRequestIsInvalid(
ShareGroupHeartbeatRequestData request
) throws InvalidRequestException, UnsupportedAssignorException {
throwIfEmptyString(request.groupId(), "GroupId can't be empty.");
throwIfEmptyString(request.rackId(), "RackId can't be empty.");
throwIfEmptyString(request.memberId(), "MemberId can't be empty."); // Always check memberId
if (request.memberEpoch() == 0) {
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
} else if (request.memberEpoch() < ShareGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH) {
// Handle any negative memberEpoch except the LEAVE_GROUP_MEMBER_EPOCH value
throw new InvalidRequestException("MemberEpoch is invalid.");
}
}
```
Dose this make 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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1807564866
##########
core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala:
##########
@@ -569,7 +569,7 @@ class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) {
protected def consumerGroupHeartbeat(
groupId: String,
- memberId: String = "",
+ memberId: String = Uuid.randomUuid().toString,
Review Comment:
Hi @dajac
I have a question about test in server-side. Could you please take a look ?
Many 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1819213379
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1293,21 +1294,28 @@ private void throwIfNull(
* Validates the request.
*
* @param request The request to validate.
- *
+ * @param apiVersion The version of ConsumerGroupHeartbeat RPC
* @throws InvalidRequestException if the request is not valid.
* @throws UnsupportedAssignorException if the assignor is not supported.
*/
private void throwIfConsumerGroupHeartbeatRequestIsInvalid(
- ConsumerGroupHeartbeatRequestData request
+ ConsumerGroupHeartbeatRequestData request,
+ short apiVersion
) throws InvalidRequestException, UnsupportedAssignorException {
+ if (apiVersion >= CONSUMER_GENERATED_MEMBER_ID_REQUIRED_VERSION ||
+ request.memberEpoch() > 0 ||
+ request.memberEpoch() == LEAVE_GROUP_MEMBER_EPOCH
+ ) {
+ throwIfNull(request.memberId(), "MemberId can't be null.");
Review Comment:
Yes, you're right.
There is no `nullableVersions` and `"default": null` for `memberId`. I will remove this null check.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1819301417
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10226,48 +10256,48 @@ public void testConsumerGroupHeartbeatToClassicGroupFromExistingStaticMember() {
.build();
List<CoordinatorRecord> expectedRecords = Arrays.asList(
- // The existing classic group tombstone.
- GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord(groupId),
-
- // Create the new consumer group with the static member.
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedClassicMember),
- GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 0),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId, expectedClassicMember.assignedPartitions()),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentEpochRecord(groupId, 0),
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedClassicMember),
-
- // Remove the static member because the rejoining member replaces it.
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentTombstoneRecord(groupId, memberId),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentTombstoneRecord(groupId, memberId),
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionTombstoneRecord(groupId, memberId),
+ // The existing classic group tombstone.
Review Comment:
The only change is variable name (`newMemberId` -> `serverReturnedMemberId`) since there is no new member id generated.
https://github.com/frankvicky/kafka/blob/51e19205be60e3baf6c861575a989713247cdc37/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java#L10229-L10236
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1824606302
##########
clients/src/main/resources/common/message/ConsumerGroupHeartbeatRequest.json:
##########
@@ -18,15 +18,15 @@
"type": "request",
"listeners": ["zkBroker", "broker"],
Review Comment:
I see, I will revert it shortly
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1820686572
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -944,23 +941,20 @@ public boolean sameRequest(final OffsetFetchRequestState request) {
public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
OffsetFetchRequest.Builder builder;
- if (memberInfo.memberId.isPresent() && memberInfo.memberEpoch.isPresent()) {
- builder = new OffsetFetchRequest.Builder(
- groupId,
- memberInfo.memberId.get(),
- memberInfo.memberEpoch.get(),
- true,
- new ArrayList<>(this.requestedPartitions),
- throwOnFetchStableOffsetUnsupported);
- } else {
- // Building request without passing member ID/epoch to leave the logic to choose
- // default values when not present on the request builder.
- builder = new OffsetFetchRequest.Builder(
- groupId,
- true,
- new ArrayList<>(this.requestedPartitions),
- throwOnFetchStableOffsetUnsupported);
- }
+ // Building request without passing member ID/epoch to leave the logic to choose
+ // default values when not present on the request builder.
+ builder = memberInfo.memberEpoch.map(epoch -> new OffsetFetchRequest.Builder(
+ groupId,
+ memberInfo.memberId,
+ epoch,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported))
+ .orElseGet(() -> new OffsetFetchRequest.Builder(
+ groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported));
Review Comment:
I'm not sure about this issue is related to following code or not. [0]
If so, I think yes，we could call the method if `memberEpoch` is present.
As my previous comment[1]，the `groupMetadata` should represent the broker acknowledgements since we're moving coordinator logic to broker side.
If I don't misunderstand, I will update it in next commit.
[0]:
https://github.com/apache/kafka/blob/8c071b02e9908d9facf10c0a18e7e0f9d1b0825f/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java#L636-L647
[1]:
>Hi @lianetm,
>
>Here is my understanding. Please correct me if I’m wrong:
>Since the group management has now moved to the server-side, the ConsumerGroupMetadata basically represents the data acknowledged by the broker. In this case, I think we should keep the ConsumerGroupMetadata.memberId empty at startup. This is because, at that moment, the broker still has no knowledge of the memberId. Once the consumer polls, the broker will then recognize the memberId and update >the assignment, allowing the consumer to update the ConsumerGroupMetadata via MemberStateListener. After >that, ConsumerGroupMetadata will have the memberId.
>
> Does this make 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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1820686572
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -944,23 +941,20 @@ public boolean sameRequest(final OffsetFetchRequestState request) {
public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
OffsetFetchRequest.Builder builder;
- if (memberInfo.memberId.isPresent() && memberInfo.memberEpoch.isPresent()) {
- builder = new OffsetFetchRequest.Builder(
- groupId,
- memberInfo.memberId.get(),
- memberInfo.memberEpoch.get(),
- true,
- new ArrayList<>(this.requestedPartitions),
- throwOnFetchStableOffsetUnsupported);
- } else {
- // Building request without passing member ID/epoch to leave the logic to choose
- // default values when not present on the request builder.
- builder = new OffsetFetchRequest.Builder(
- groupId,
- true,
- new ArrayList<>(this.requestedPartitions),
- throwOnFetchStableOffsetUnsupported);
- }
+ // Building request without passing member ID/epoch to leave the logic to choose
+ // default values when not present on the request builder.
+ builder = memberInfo.memberEpoch.map(epoch -> new OffsetFetchRequest.Builder(
+ groupId,
+ memberInfo.memberId,
+ epoch,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported))
+ .orElseGet(() -> new OffsetFetchRequest.Builder(
+ groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported));
Review Comment:
I'm not sure about this issue is related to following code or not. [0]
If so, I think yes，we could call the method if `memberEpoch` is present.
As my previous comment[1]，the `groupMetadata` should represent the broker acknowledgements since we're moving coordinator logic to broker side.
If I don't misunderstand, I will update it in next commit.
[0]
https://github.com/apache/kafka/blob/8c071b02e9908d9facf10c0a18e7e0f9d1b0825f/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java#L636-L647
[1]
>Hi @lianetm,
>
>Here is my understanding. Please correct me if I’m wrong:
>Since the group management has now moved to the server-side, the ConsumerGroupMetadata basically represents the data acknowledged by the broker. In this case, I think we should keep the ConsumerGroupMetadata.memberId empty at startup. This is because, at that moment, the broker still has no knowledge of the memberId. Once the consumer polls, the broker will then recognize the memberId and update >the assignment, allowing the consumer to update the ConsumerGroupMetadata via MemberStateListener. After >that, ConsumerGroupMetadata will have the memberId.
>
> Does this make 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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1821049715
##########
core/src/test/scala/unit/kafka/server/ConsumerGroupHeartbeatRequestTest.scala:
##########
@@ -36,54 +35,68 @@ import scala.collection.Map
import scala.jdk.CollectionConverters._
@ExtendWith(value = Array(classOf[ClusterTestExtensions]))
-class ConsumerGroupHeartbeatRequestTest(cluster: ClusterInstance) {
-
- @ClusterTest(
+@ClusterTestDefaults(
types = Array(Type.KRAFT),
serverProperties = Array(
- new ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, value = "classic"),
new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1")
)
+)
+class ConsumerGroupHeartbeatRequestTest(cluster: ClusterInstance) {
+
+ @ClusterTests(value = Array(
+ new ClusterTest(
+ serverProperties = Array(new ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, value = "classic")),
+ tags = Array("0")
+ ),
+ new ClusterTest(
+ serverProperties = Array(new ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, value = "classic")),
+ tags = Array("1")
+ )
+ )
)
Review Comment:
Thanks for feedback.
I think this approach is very elegant. 👍🏼
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1809998105
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10117,8 +10130,9 @@ public void testConsumerGroupHeartbeatWithPreparingRebalanceClassicGroup() throw
assertEquals(group, context.groupMetadataManager.getOrMaybeCreateClassicGroup("group-id", false));
}
- @Test
- public void testConsumerGroupHeartbeatToClassicGroupFromExistingStaticMember() {
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testConsumerGroupHeartbeatToClassicGroupFromExistingStaticMember(boolean isConsumerGeneratedMemberId) {
Review Comment:
Originally, I was thinking covering both scenarios for client-generated memberId and server-generated memberId. However, based on the previous comment, I will focus on simulating the behavior of the latest version and update the test case accordingly.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1807324925
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -1234,8 +1245,8 @@ protected void resetEpoch() {
protected void updateMemberEpoch(int newEpoch) {
boolean newEpochReceived = this.memberEpoch != newEpoch;
this.memberEpoch = newEpoch;
- // Simply notify based on epoch change only, given that the member will never receive a
- // new member ID without an epoch (member ID is only assigned when it joins the group).
+ // Simply notify based on epoch changes only, since the member will generate a member ID
+ // at startup, and it will remain unchanged for its entire lifetime.
if (newEpochReceived) {
Review Comment:
I'm not sure if we need to update the logic here accordingly.
https://github.com/apache/kafka/blob/2267518d5bed60e2a1731712ae7cb455ea0bf8c4/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java#L1239-L1245
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1821015724
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -944,23 +941,20 @@ public boolean sameRequest(final OffsetFetchRequestState request) {
public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
OffsetFetchRequest.Builder builder;
- if (memberInfo.memberId.isPresent() && memberInfo.memberEpoch.isPresent()) {
- builder = new OffsetFetchRequest.Builder(
- groupId,
- memberInfo.memberId.get(),
- memberInfo.memberEpoch.get(),
- true,
- new ArrayList<>(this.requestedPartitions),
- throwOnFetchStableOffsetUnsupported);
- } else {
- // Building request without passing member ID/epoch to leave the logic to choose
- // default values when not present on the request builder.
- builder = new OffsetFetchRequest.Builder(
- groupId,
- true,
- new ArrayList<>(this.requestedPartitions),
- throwOnFetchStableOffsetUnsupported);
- }
+ // Building request without passing member ID/epoch to leave the logic to choose
+ // default values when not present on the request builder.
+ builder = memberInfo.memberEpoch.map(epoch -> new OffsetFetchRequest.Builder(
+ groupId,
+ memberInfo.memberId,
+ epoch,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported))
+ .orElseGet(() -> new OffsetFetchRequest.Builder(
+ groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported));
Review Comment:
makes sense to me to align and return `memberId` in the consumer group metadata info only if the member is in a group. So after `unsubscribe` (this was my concern btw, not startup), even if internally the `membershipMgr` knows the member ID, we're saying that the consumer group metadata won't include it (change already in place). Agreed.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1824516687
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -95,10 +95,9 @@ public class CommitRequestManager implements RequestManager, MemberStateListener
private Optional<Integer> lastEpochSentOnCommit;
/**
- * Latest member ID and epoch received via the {@link #onMemberEpochUpdated(Optional, Optional)},
- * to be included in the OffsetFetch and OffsetCommit requests if present. This will have
- * the latest values received from the broker, or empty of the member is not part of the
- * group anymore.
+ * The member ID and latest member epoch received via the {@link MemberStateListener#onMemberEpochUpdated(Optional, String)},
+ * to be included in the OffsetFetch and OffsetCommit requests. This will have
+ * the latest memberEpoch values received from the broker.
Review Comment:
I think `value` is a bit redundant, I will 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1824546147
##########
clients/src/main/resources/common/message/ConsumerGroupHeartbeatRequest.json:
##########
@@ -18,15 +18,15 @@
"type": "request",
"listeners": ["zkBroker", "broker"],
Review Comment:
Nope it does not. It returns the unsupported version error. The zk stuff will be removed in 4.0 anyway.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1820671218
##########
core/src/test/scala/unit/kafka/server/ConsumerGroupHeartbeatRequestTest.scala:
##########
@@ -36,54 +35,68 @@ import scala.collection.Map
import scala.jdk.CollectionConverters._
@ExtendWith(value = Array(classOf[ClusterTestExtensions]))
-class ConsumerGroupHeartbeatRequestTest(cluster: ClusterInstance) {
-
- @ClusterTest(
+@ClusterTestDefaults(
types = Array(Type.KRAFT),
serverProperties = Array(
- new ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, value = "classic"),
new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1")
)
+)
+class ConsumerGroupHeartbeatRequestTest(cluster: ClusterInstance) {
+
+ @ClusterTests(value = Array(
+ new ClusterTest(
+ serverProperties = Array(new ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, value = "classic")),
+ tags = Array("0")
+ ),
+ new ClusterTest(
+ serverProperties = Array(new ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, value = "classic")),
+ tags = Array("1")
+ )
+ )
)
Review Comment:
I’m currently testing different versions of `ConsumerGroupHeartbeatRequest` using the `tags` property as a workaround. It feels quite clunky, but I haven’t found a way to do parameterized tests within the cluster test context. 🥵
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1815322199
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMembershipManager.java:
##########
@@ -188,6 +186,7 @@ public ConsumerMembershipManager(String groupId,
this.rebalanceTimeoutMs = rebalanceTimeoutMs;
this.serverAssignor = serverAssignor;
this.commitRequestManager = commitRequestManager;
+ this.commitRequestManager.memberInfo.memberId = this.memberId;
Review Comment:
This line will cause tests to fail in `ConsumerMembershipManagerTest` because `commitRequestManager` is a mock, meaning its `memberInfo` is never initialized. As a result, assigning a value will lead to a NPE.
I'm considering how to handle this issue.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1824442610
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -14086,16 +14123,20 @@ public void testShareGroupMemberIdGeneration() {
Collections.emptyMap()
));
+ String memberId = Uuid.randomUuid().toString();
CoordinatorResult<ShareGroupHeartbeatResponseData, CoordinatorRecord> result = context.shareGroupHeartbeat(
new ShareGroupHeartbeatRequestData()
.setGroupId("group-foo")
+ .setMemberId(memberId)
.setMemberEpoch(0)
.setSubscribedTopicNames(Arrays.asList("foo", "bar")));
- // Verify that a member id was generated for the new member.
- String memberId = result.response().memberId();
- assertNotNull(memberId);
- assertNotEquals("", memberId);
+ assertNotNull(result.response().memberId());
Review Comment:
nit: This one is not needed because assertEquals already cover 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #17549:
URL: https://github.com/apache/kafka/pull/17549#issuecomment-2449940365
@chia7712 Do you want to do another pass on 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on PR #17549:
URL: https://github.com/apache/kafka/pull/17549#issuecomment-2451153826
I really appreciate everyone's review and patience. 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1821182275
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -433,7 +442,7 @@ public void transitionToFatal() {
MemberState previousState = state;
transitionTo(MemberState.FATAL);
log.error("Member {} with epoch {} transitioned to fatal state", memberIdInfoForLog(), memberEpoch);
Review Comment:
+1 to remove memberIdInfoForLog, as I left the same comment here: [link](https://github.com/apache/kafka/pull/17549#discussion_r1821135581) :smile:
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1820756222
##########
core/src/test/scala/unit/kafka/server/ConsumerGroupHeartbeatRequestTest.scala:
##########
@@ -36,54 +35,68 @@ import scala.collection.Map
import scala.jdk.CollectionConverters._
@ExtendWith(value = Array(classOf[ClusterTestExtensions]))
-class ConsumerGroupHeartbeatRequestTest(cluster: ClusterInstance) {
-
- @ClusterTest(
+@ClusterTestDefaults(
types = Array(Type.KRAFT),
serverProperties = Array(
- new ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, value = "classic"),
new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1")
)
+)
+class ConsumerGroupHeartbeatRequestTest(cluster: ClusterInstance) {
+
+ @ClusterTests(value = Array(
+ new ClusterTest(
+ serverProperties = Array(new ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, value = "classic")),
+ tags = Array("0")
+ ),
+ new ClusterTest(
+ serverProperties = Array(new ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, value = "classic")),
+ tags = Array("1")
+ )
+ )
)
Review Comment:
I have tried using a `for` loop locally, but since this approach doesn’t provide a 'clean' environment for each iteration, the assertions might get polluted.
Is it acceptable ?
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1814431666
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1317,6 +1319,9 @@ private void throwIfConsumerGroupHeartbeatRequestIsInvalid(
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
+ if (apiVersion >= CONSUMER_GENERATED_MEMBER_ID_REQUIRED_VERSION) {
+ throwIfEmptyString(request.memberId(), "MemberId can't be empty.");
Review Comment:
Make 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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1819183418
##########
core/src/test/scala/unit/kafka/server/DeleteGroupsRequestTest.scala:
##########
@@ -36,6 +37,7 @@ class DeleteGroupsRequestTest(cluster: ClusterInstance) extends GroupCoordinator
)
)
def testDeleteGroupsWithNewConsumerGroupProtocolAndNewGroupCoordinator(): Unit = {
+ // TODO fix
Review Comment:
This comment was accidentally left here and should be deleted. I will remove it in the next commit.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1819213379
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1293,21 +1294,28 @@ private void throwIfNull(
* Validates the request.
*
* @param request The request to validate.
- *
+ * @param apiVersion The version of ConsumerGroupHeartbeat RPC
* @throws InvalidRequestException if the request is not valid.
* @throws UnsupportedAssignorException if the assignor is not supported.
*/
private void throwIfConsumerGroupHeartbeatRequestIsInvalid(
- ConsumerGroupHeartbeatRequestData request
+ ConsumerGroupHeartbeatRequestData request,
+ short apiVersion
) throws InvalidRequestException, UnsupportedAssignorException {
+ if (apiVersion >= CONSUMER_GENERATED_MEMBER_ID_REQUIRED_VERSION ||
+ request.memberEpoch() > 0 ||
+ request.memberEpoch() == LEAVE_GROUP_MEMBER_EPOCH
+ ) {
+ throwIfNull(request.memberId(), "MemberId can't be null.");
Review Comment:
Yes, you're right.
There is no `nullableVersions` and `"default: null"` for `memberId`. I will remove this null check.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1807326273
##########
core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala:
##########
@@ -569,7 +569,7 @@ class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) {
protected def consumerGroupHeartbeat(
groupId: String,
- memberId: String = "",
+ memberId: String = Uuid.randomUuid().toString,
Review Comment:
Since we are upgrading the `ConsumerGroupHeartbeat`, the `memberId` cannot be empty or null.
It would be nice to test the older version of the protocol, but I'm not quite sure how to achieve 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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1814848836
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -1293,17 +1287,16 @@ public void setInflightCommitStatus(final boolean inflightCommitStatus) {
}
static class MemberInfo {
- Optional<String> memberId;
+ String memberId = "";
Review Comment:
Yes, I think it's very ideal! But after quickly going through `RequestManagers`, it seems that `MembershipManager` relies on `CommitRequestManager`. If that's the case, it could be tricky. (L213)
https://github.com/apache/kafka/blob/553e6b4c6d912b3b619b5a1a5263b616b6d3f62e/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java#L197-L219
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1814557313
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -1234,8 +1245,8 @@ protected void resetEpoch() {
protected void updateMemberEpoch(int newEpoch) {
boolean newEpochReceived = this.memberEpoch != newEpoch;
this.memberEpoch = newEpoch;
- // Simply notify based on epoch change only, given that the member will never receive a
- // new member ID without an epoch (member ID is only assigned when it joins the group).
+ // Simply notify based on epoch changes only, since the member will generate a member ID
+ // at startup, and it will remain unchanged for its entire lifetime.
if (newEpochReceived) {
Review Comment:
Thanks for feedback. 😺
I’m wondering about the initial value of `MemberInfo.memberId` if we replace it with a `String` instead of `Optional<String>`. Should we use an empty string as its initial value ?
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1814794971
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -95,10 +95,9 @@ public class CommitRequestManager implements RequestManager, MemberStateListener
private Optional<Integer> lastEpochSentOnCommit;
/**
- * Latest member ID and epoch received via the {@link #onMemberEpochUpdated(Optional, Optional)},
- * to be included in the OffsetFetch and OffsetCommit requests if present. This will have
- * the latest values received from the broker, or empty of the member is not part of the
- * group anymore.
+ * Latest member ID and epoch received via the {@link MemberStateListener#onMemberEpochUpdated(Optional, String)},
Review Comment:
nit: the concept of "latest member ID received" is confusing now given that we don't receive it from the broker anymore. Maybe something like "The member ID and latest epoch..." (used in other places here too, let's fix them all)
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1815213715
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -1293,17 +1287,16 @@ public void setInflightCommitStatus(final boolean inflightCommitStatus) {
}
static class MemberInfo {
- Optional<String> memberId;
+ String memberId = "";
Review Comment:
I reviewed the code and found that in `AsyncKafkaConsumer`, the `updateGroupMetadata` method uses `memberId` to update `ConsumerGroupMetadata`[0]. Unfortunately, this method is passed as a lambda to implement `onMemberEpochUpdated`[1]. As a result, we can't remove the `memberId` parameter from `onMemberEpochUpdated` since `memberId` cannot be accessed directly within `AsyncKafkaConsumer`.
The best we could do now is to make sure we only set `MemberInfo.memberId` once at `membershipManager` constructor.
[0]
https://github.com/apache/kafka/blob/140d35c5459f4c7a91b23a238467cb5aa01b59fb/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java#L635-L646
[1]
https://github.com/apache/kafka/blob/140d35c5459f4c7a91b23a238467cb5aa01b59fb/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java#L354
It's pass as lambda to L159:
https://github.com/apache/kafka/blob/140d35c5459f4c7a91b23a238467cb5aa01b59fb/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java#L145-L159
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1815339271
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMembershipManager.java:
##########
@@ -188,6 +186,7 @@ public ConsumerMembershipManager(String groupId,
this.rebalanceTimeoutMs = rebalanceTimeoutMs;
this.serverAssignor = serverAssignor;
this.commitRequestManager = commitRequestManager;
+ this.commitRequestManager.memberInfo.memberId = this.memberId;
Review Comment:
well from your [comment](https://github.com/apache/kafka/pull/17549#discussion_r1815213715) above it seems that even if we set the member ID explicitly, it is still needed to keep it on the `onMemberEpochUpdated` as param right? if so then it's probably not worth the hassle, we can just remove this line and stay with the memberID updated on the listeners as part of the onMemberEpochUpdated
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on PR #17549:
URL: https://github.com/apache/kafka/pull/17549#issuecomment-2434708086
Hi @lianetm
Thanks for point out.
I have correct the typo and url. 😺
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1824570188
##########
clients/src/main/resources/common/message/ConsumerGroupHeartbeatRequest.json:
##########
@@ -18,15 +18,15 @@
"type": "request",
"listeners": ["zkBroker", "broker"],
Review Comment:
I will remove `zkBroker` 😺
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #17549:
URL: https://github.com/apache/kafka/pull/17549
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1810407560
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1351,6 +1355,7 @@ private void throwIfShareGroupHeartbeatRequestIsInvalid(
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
+ throwIfEmptyString(request.memberId(), "MemberId can't be empty.");
Review Comment:
Since `ShareGroupHeartbeat RPC` is still under development, I assume this change is acceptable without a version check.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1814943048
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -1293,17 +1287,16 @@ public void setInflightCommitStatus(final boolean inflightCommitStatus) {
}
static class MemberInfo {
- Optional<String> memberId;
+ String memberId = "";
Review Comment:
uhm true. Then what about we simply do a kind of commitManager.setMemberId(memberId) from the membershipMgr constructor (right after assigning the commitRequestManager)? The point I'm after in the commitMgr is to clearly separate the `memberId` (set once, never updated) from the `memberEpoch` (may be updated and the commitMgr needs to know about those 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1814788055
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -1293,17 +1287,16 @@ public void setInflightCommitStatus(final boolean inflightCommitStatus) {
}
static class MemberInfo {
- Optional<String> memberId;
+ String memberId = "";
Review Comment:
taking this a step further, I wonder if it would be clearer if this commitManager (and this MemberInfo), just take the memberId as constructor parameter. The memberId is generated by the membershipMgr once at startup, won't change, so it could pass it on to the commitManager upon creation here:
https://github.com/apache/kafka/blob/553e6b4c6d912b3b619b5a1a5263b616b6d3f62e/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java#L197 (creating membershipMgr first)
Then the `onMemberEpochUpdated` would only receive the epoch, which is what changes. What do you think?
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #17549:
URL: https://github.com/apache/kafka/pull/17549#issuecomment-2435743145
One concern, the unsubscribe() logic in the consumer includes logic for resetting the group metadata (will clear the member ID).
https://github.com/apache/kafka/blob/140d35c5459f4c7a91b23a238467cb5aa01b59fb/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java#L630
I expect this needs to be updated now, to keep the memberId (otherwise a running consumer that calls unsubscribe would loose the member ID that is only generated now when the consumer starts). 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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1820674708
##########
core/src/test/scala/unit/kafka/server/ConsumerProtocolMigrationTest.scala:
##########
@@ -242,50 +243,54 @@ class ConsumerProtocolMigrationTest(cluster: ClusterInstance) extends GroupCoord
assignment = assignment(List(0, 1, 2))
)
- // The joining request with a consumer group member 2 is accepted.
- val memberId2 = consumerGroupHeartbeat(
- groupId = groupId,
- rebalanceTimeoutMs = 5 * 60 * 1000,
- subscribedTopicNames = List("foo"),
- topicPartitions = List.empty,
- expectedError = Errors.NONE
- ).memberId
-
- // The group has become a consumer group.
- assertEquals(
- List(
- new ListGroupsResponseData.ListedGroup()
- .setGroupId(groupId)
- .setProtocolType("consumer")
- .setGroupState(ConsumerGroupState.RECONCILING.toString)
- .setGroupType(Group.GroupType.CONSUMER.toString)
- ),
- listGroups(
- statesFilter = List.empty,
- typesFilter = List(Group.GroupType.CONSUMER.toString)
+ for (version <- ApiKeys.CONSUMER_GROUP_HEARTBEAT.oldestVersion() to ApiKeys.CONSUMER_GROUP_HEARTBEAT.latestVersion(isUnstableApiEnabled)) {
Review Comment:
I have test all the versions in `ConsumerGroupHeartbeatRequestTest` but I have a question, see:
https://github.com/apache/kafka/pull/17549#discussion_r1820671218
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1820736948
##########
core/src/test/scala/unit/kafka/server/ConsumerGroupHeartbeatRequestTest.scala:
##########
@@ -36,54 +35,68 @@ import scala.collection.Map
import scala.jdk.CollectionConverters._
@ExtendWith(value = Array(classOf[ClusterTestExtensions]))
-class ConsumerGroupHeartbeatRequestTest(cluster: ClusterInstance) {
-
- @ClusterTest(
+@ClusterTestDefaults(
types = Array(Type.KRAFT),
serverProperties = Array(
- new ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, value = "classic"),
new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1")
)
+)
+class ConsumerGroupHeartbeatRequestTest(cluster: ClusterInstance) {
+
+ @ClusterTests(value = Array(
+ new ClusterTest(
+ serverProperties = Array(new ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, value = "classic")),
+ tags = Array("0")
+ ),
+ new ClusterTest(
+ serverProperties = Array(new ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, value = "classic")),
+ tags = Array("1")
+ )
+ )
)
Review Comment:
Yeah, this is a bit too extreme. We have ApiVersionsSource but you cannot combine it with ClusterTest. Honestly, the best may be to just loop over the version within the test 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1824591348
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMembershipManager.java:
##########
@@ -158,10 +157,15 @@ public ConsumerMembershipManager(String groupId,
commitRequestManager,
metadata,
logContext,
- clientTelemetryReporter,
backgroundEventHandler,
time,
new ConsumerRebalanceMetricsManager(metrics));
+
+ // Update the group member ID label in the client telemetry reporter.
Review Comment:
I also apply this change to `shareMembershipManager`
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1820699580
##########
clients/src/main/resources/common/message/ConsumerGroupHeartbeatResponse.json:
##########
@@ -39,7 +39,7 @@
{ "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The top-level error message, or null if there was no error." },
{ "name": "MemberId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
- "about": "The member id generated by the coordinator. Only provided when the member joins with MemberEpoch == 0." },
+ "about": "The member id is generated by the consumer and provided by the consumer for all requests." },
Review Comment:
~hmm... I feel that we don't need to specify the behavior in version `0` since it's an unstable version.~
~If users are using the 4.0 release, they should have the latest behavior.~
Update: I think we should precise it since it's response from server.
> However, in version 1+, it is basically the value received from the consumer?
I think yes.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1809780391
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3199,7 +3201,7 @@ public CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord>
RequestContext context,
ConsumerGroupHeartbeatRequestData request
) throws ApiException {
- throwIfConsumerGroupHeartbeatRequestIsInvalid(request);
+ throwIfConsumerGroupHeartbeatRequestIsInvalid(request, context.apiVersion());
if (request.memberEpoch() == LEAVE_GROUP_MEMBER_EPOCH || request.memberEpoch() == LEAVE_GROUP_STATIC_MEMBER_EPOCH) {
// -1 means that the member wants to leave the group.
Review Comment:
Yep, I will update it accordingly
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1810764018
##########
core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala:
##########
@@ -606,7 +608,7 @@ class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) {
protected def shareGroupHeartbeat(
groupId: String,
- memberId: String = "",
+ memberId: String = Uuid.randomUuid.toString,
Review Comment:
Yeah, I'm keeping an eye on all this and will make sure KIP-932 incorporates all of the latest refinements before it is completed.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1810581161
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1351,6 +1355,7 @@ private void throwIfShareGroupHeartbeatRequestIsInvalid(
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
+ throwIfEmptyString(request.memberId(), "MemberId can't be empty.");
Review Comment:
Since consumers are now required to generate their own `memberId` and the `ShareGroupHeartbeat RPC` is still under development, it seems feasible to check the `memberId` at the beginning:
```java
private void throwIfShareGroupHeartbeatRequestIsInvalid(
ShareGroupHeartbeatRequestData request
) throws InvalidRequestException, UnsupportedAssignorException {
throwIfEmptyString(request.groupId(), "GroupId can't be empty.");
throwIfEmptyString(request.rackId(), "RackId can't be empty.");
throwIfEmptyString(request.memberId(), "MemberId can't be empty."); // Always check memberId
if (request.memberEpoch() == 0) {
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
} else if (request.memberEpoch() < 0 && request.memberEpoch() != ShareGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH) {
// Handle any negative memberEpoch except the LEAVE_GROUP_MEMBER_EPOCH value
throw new InvalidRequestException("MemberEpoch is invalid.");
}
}
```
Dose this make 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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1814798029
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -944,23 +941,20 @@ public boolean sameRequest(final OffsetFetchRequestState request) {
public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
OffsetFetchRequest.Builder builder;
- if (memberInfo.memberId.isPresent() && memberInfo.memberEpoch.isPresent()) {
- builder = new OffsetFetchRequest.Builder(
- groupId,
- memberInfo.memberId.get(),
- memberInfo.memberEpoch.get(),
- true,
- new ArrayList<>(this.requestedPartitions),
- throwOnFetchStableOffsetUnsupported);
- } else {
- // Building request without passing member ID/epoch to leave the logic to choose
- // default values when not present on the request builder.
- builder = new OffsetFetchRequest.Builder(
- groupId,
- true,
- new ArrayList<>(this.requestedPartitions),
- throwOnFetchStableOffsetUnsupported);
- }
+ // Building request without passing member ID/epoch to leave the logic to choose
+ // default values when not present on the request builder.
+ builder = memberInfo.memberEpoch.map(integer -> new OffsetFetchRequest.Builder(
+ groupId,
+ memberInfo.memberId,
+ integer,
Review Comment:
nit: let's rename this var from "integer" to something clearer, maybe "epoch"
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1810581161
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1351,6 +1355,7 @@ private void throwIfShareGroupHeartbeatRequestIsInvalid(
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
+ throwIfEmptyString(request.memberId(), "MemberId can't be empty.");
Review Comment:
Since consumers are now required to generate their own `memberId` and the `ShareGroupHeartbeat RPC` is still under development, it seems feasible to check the `memberId` at the beginning:
```java
private void throwIfShareGroupHeartbeatRequestIsInvalid(
ShareGroupHeartbeatRequestData request
) throws InvalidRequestException, UnsupportedAssignorException {
throwIfEmptyString(request.groupId(), "GroupId can't be empty.");
throwIfEmptyString(request.rackId(), "RackId can't be empty.");
throwIfEmptyString(request.memberId(), "MemberId can't be empty.");
if (request.memberEpoch() == 0) {
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
} else {
throw new InvalidRequestException("MemberEpoch is invalid.");
}
}
```
Dose this make 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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1810680947
##########
core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala:
##########
@@ -606,7 +608,7 @@ class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) {
protected def shareGroupHeartbeat(
groupId: String,
- memberId: String = "",
+ memberId: String = Uuid.randomUuid.toString,
Review Comment:
Since `ShareGroupHeartbeat RPC` is still under development, I think it's acceptable to change the default value from an empty string to a UUID.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1820686572
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -944,23 +941,20 @@ public boolean sameRequest(final OffsetFetchRequestState request) {
public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
OffsetFetchRequest.Builder builder;
- if (memberInfo.memberId.isPresent() && memberInfo.memberEpoch.isPresent()) {
- builder = new OffsetFetchRequest.Builder(
- groupId,
- memberInfo.memberId.get(),
- memberInfo.memberEpoch.get(),
- true,
- new ArrayList<>(this.requestedPartitions),
- throwOnFetchStableOffsetUnsupported);
- } else {
- // Building request without passing member ID/epoch to leave the logic to choose
- // default values when not present on the request builder.
- builder = new OffsetFetchRequest.Builder(
- groupId,
- true,
- new ArrayList<>(this.requestedPartitions),
- throwOnFetchStableOffsetUnsupported);
- }
+ // Building request without passing member ID/epoch to leave the logic to choose
+ // default values when not present on the request builder.
+ builder = memberInfo.memberEpoch.map(epoch -> new OffsetFetchRequest.Builder(
+ groupId,
+ memberInfo.memberId,
+ epoch,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported))
+ .orElseGet(() -> new OffsetFetchRequest.Builder(
+ groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported));
Review Comment:
I'm not sure about this issue is related to following code or not：
https://github.com/apache/kafka/blob/8c071b02e9908d9facf10c0a18e7e0f9d1b0825f/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java#L636-L647
If so, I think yes，we could call the method if `memberEpoch` is present.
As my previous comment，the `groupMetadata` should represent the broker acknowledgements since we're moving coordinator logic to broker side.
If I don't misunderstand, I will update it in next commit.
>Hi @lianetm,
>
>Here is my understanding. Please correct me if I’m wrong:
>Since the group management has now moved to the server-side, the ConsumerGroupMetadata basically represents the data acknowledged by the broker. In this case, I think we should keep the ConsumerGroupMetadata.memberId empty at startup. This is because, at that moment, the broker still has no knowledge of the memberId. Once the consumer polls, the broker will then recognize the memberId and update >the assignment, allowing the consumer to update the ConsumerGroupMetadata via MemberStateListener. After >that, ConsumerGroupMetadata will have the memberId.
>
> Does this make 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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1821126889
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -224,6 +227,12 @@ public abstract class AbstractMembershipManager<R extends AbstractResponse> impl
this.clientTelemetryReporter = clientTelemetryReporter;
this.time = time;
this.metricsManager = metricsManager;
+
+ // Update the group member ID label in the client telemetry reporter.
+ // According to KIP-1082, the consumer will generate the member ID as the incarnation ID of the process.
+ // Therefore, we can update the group member ID during initialization.
+ clientTelemetryReporter.ifPresent(reporter -> reporter.updateMetricsLabels(
+ Collections.singletonMap(ClientTelemetryProvider.GROUP_MEMBER_ID, memberId)));
Review Comment:
`Map.of`
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -224,6 +227,12 @@ public abstract class AbstractMembershipManager<R extends AbstractResponse> impl
this.clientTelemetryReporter = clientTelemetryReporter;
this.time = time;
this.metricsManager = metricsManager;
+
+ // Update the group member ID label in the client telemetry reporter.
+ // According to KIP-1082, the consumer will generate the member ID as the incarnation ID of the process.
+ // Therefore, we can update the group member ID during initialization.
+ clientTelemetryReporter.ifPresent(reporter -> reporter.updateMetricsLabels(
Review Comment:
we don't need to pass `clientTelemetryReporter` to `AbstractMembershipManager` as we can call `updateMetricsLabels` after initializing `AbstractMembershipManager`, right? The member is fixed now
##########
clients/src/main/resources/common/message/ConsumerGroupHeartbeatResponse.json:
##########
@@ -39,7 +39,7 @@
{ "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The top-level error message, or null if there was no error." },
{ "name": "MemberId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
- "about": "The member id generated by the coordinator. Only provided when the member joins with MemberEpoch == 0." },
+ "about": "The member id is generated by the consumer and provided by the consumer for all requests starting from version 1. In version 0, however, the member ID is generated by the group coordinator." },
Review Comment:
`In version 0, however, the member ID is generated by the group coordinator.`
This description is inaccurate; in version 0, users can also generate a member ID.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -80,10 +81,12 @@ public abstract class AbstractMembershipManager<R extends AbstractResponse> impl
protected final String groupId;
/**
- * Member ID assigned by the server to the member, received in a heartbeat response when
- * joining the group specified in {@link #groupId}
+ * Member ID generated by the consumer at startup, which is unique within the group and remains consistent
+ * for the entire lifetime of the process. This ID acts as an incarnation identifier for the consumer process
+ * and does not reset or change, even if the consumer leaves and rejoins the group.
+ * The Member ID remains the same until the process is completely stopped or terminated.
*/
- protected String memberId = "";
+ protected final String memberId = Uuid.randomUuid().toString();
Review Comment:
`memberIdInfoForLog` is useless, correct?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -622,10 +631,10 @@ public void transitionToSendingLeaveGroup(boolean dueToExpiredPollTimer) {
/**
* Call all listeners that are registered to get notified when the member epoch is updated.
- * This also includes the latest member ID in the notification. If the member fails or leaves
- * the group, this will be invoked with empty epoch and member ID.
+ * This also includes the member ID in the notification. If the member fails or leaves
+ * the group, this will be invoked with empty epoch.
*/
- void notifyEpochChange(Optional<Integer> epoch, Optional<String> memberId) {
+ void notifyEpochChange(Optional<Integer> epoch, String memberId) {
Review Comment:
`memberId` is immutable now so we can remove `memberId` from `notifyEpochChange`
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -1293,17 +1287,16 @@ public void setInflightCommitStatus(final boolean inflightCommitStatus) {
}
static class MemberInfo {
- Optional<String> memberId;
+ String memberId = "";
Optional<Integer> memberEpoch;
MemberInfo() {
Review Comment:
we can initialize the `memberEpoch` without default constructor
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1813043564
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -1234,8 +1245,8 @@ protected void resetEpoch() {
protected void updateMemberEpoch(int newEpoch) {
boolean newEpochReceived = this.memberEpoch != newEpoch;
this.memberEpoch = newEpoch;
- // Simply notify based on epoch change only, given that the member will never receive a
- // new member ID without an epoch (member ID is only assigned when it joins the group).
+ // Simply notify based on epoch changes only, since the member will generate a member ID
+ // at startup, and it will remain unchanged for its entire lifetime.
if (newEpochReceived) {
Review Comment:
I think we should update it to align, and it's actually a simplification here and in the `CommitRequestManager` now that the memberId is always known.
The `CommitRequestManager` keeps a reference to the `memberId` and `memberEpoch`, and both were optional given that we could land here without having received the memberId from the broker. Every time the epoch changed, we provided both (to include them in the requests). Now we can probably simply have a memberId in the commitMgr as a String, not optional, here:
https://github.com/apache/kafka/blob/2d896d9130f121e75ccba2d913bdffa358cf3867/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java#L1296
and then in this membershipMgr.notifyEpochChange we always provide the known memberId and the optional epoch.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1317,6 +1319,9 @@ private void throwIfConsumerGroupHeartbeatRequestIsInvalid(
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
+ if (apiVersion >= CONSUMER_GENERATED_MEMBER_ID_REQUIRED_VERSION) {
+ throwIfEmptyString(request.memberId(), "MemberId can't be empty.");
Review Comment:
don't we need to `throwIfNull` also? if memberId is not included in the request, the `throwIfEmptyString` will not throw.
Also, with this addition, we end up validating the member ID in 3 places here (depending on the epoch), but couldn't we simplify and validate once at the beginning of this func? (checking the api version). In the end, with this KIP, if version>1 the member ID must be included in all requests, no matter the epoch.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1351,6 +1355,7 @@ private void throwIfShareGroupHeartbeatRequestIsInvalid(
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
+ throwIfEmptyString(request.memberId(), "MemberId can't be empty.");
Review Comment:
makes sense to me, and it aligns with my comment above for the consumer. Seems sensible to check at the beginning since we require member ID for all requests, regardless the epoch. And for the share case we shoudn't need the version check given that the RPC is still at v0 unstable.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1810581161
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1351,6 +1355,7 @@ private void throwIfShareGroupHeartbeatRequestIsInvalid(
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
+ throwIfEmptyString(request.memberId(), "MemberId can't be empty.");
Review Comment:
Since consumers are now required to generate their own `memberId` and the `ShareGroupHeartbeat RPC` is still under development, it seems feasible to check the `memberId` at the beginning:
```java
private void throwIfShareGroupHeartbeatRequestIsInvalid(
ShareGroupHeartbeatRequestData request
) throws InvalidRequestException, UnsupportedAssignorException {
throwIfEmptyString(request.groupId(), "GroupId can't be empty.");
throwIfEmptyString(request.rackId(), "RackId can't be empty.");
throwIfEmptyString(request.memberId(), "MemberId can't be empty."); // Always check memberId
if (request.memberEpoch() == 0) {
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
} else if (request.memberEpoch() < 0) {
throw new InvalidRequestException("MemberEpoch is invalid.");
}
}
```
Dose this make 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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1819370203
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10226,48 +10256,48 @@ public void testConsumerGroupHeartbeatToClassicGroupFromExistingStaticMember() {
.build();
List<CoordinatorRecord> expectedRecords = Arrays.asList(
- // The existing classic group tombstone.
- GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord(groupId),
-
- // Create the new consumer group with the static member.
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedClassicMember),
- GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 0),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId, expectedClassicMember.assignedPartitions()),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentEpochRecord(groupId, 0),
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedClassicMember),
-
- // Remove the static member because the rejoining member replaces it.
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentTombstoneRecord(groupId, memberId),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentTombstoneRecord(groupId, memberId),
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionTombstoneRecord(groupId, memberId),
+ // The existing classic group tombstone.
Review Comment:
I see. I suppose that the diff will get better when the indentation is fixed.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1819075308
##########
clients/src/main/java/org/apache/kafka/common/requests/ConsumerGroupHeartbeatRequest.java:
##########
@@ -37,6 +37,14 @@ public class ConsumerGroupHeartbeatRequest extends AbstractRequest {
*/
public static final int JOIN_GROUP_MEMBER_EPOCH = 0;
+ /**
+ * The version from which consumers are required to generate their own memberId.
Review Comment:
nit: Let's use `member id` here and below.
##########
clients/src/main/resources/common/message/DescribeGroupsResponse.json:
##########
@@ -48,7 +48,7 @@
{ "name": "Members", "type": "[]DescribedGroupMember", "versions": "0+",
"about": "The group members.", "fields": [
{ "name": "MemberId", "type": "string", "versions": "0+",
- "about": "The member ID assigned by the group coordinator." },
+ "about": "The member ID is either assigned by the group coordinator or generated by consumer itself." },
Review Comment:
nit: I would simplify it to `The member id`.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1293,21 +1294,28 @@ private void throwIfNull(
* Validates the request.
*
* @param request The request to validate.
- *
+ * @param apiVersion The version of ConsumerGroupHeartbeat RPC
* @throws InvalidRequestException if the request is not valid.
* @throws UnsupportedAssignorException if the assignor is not supported.
*/
private void throwIfConsumerGroupHeartbeatRequestIsInvalid(
- ConsumerGroupHeartbeatRequestData request
+ ConsumerGroupHeartbeatRequestData request,
+ short apiVersion
) throws InvalidRequestException, UnsupportedAssignorException {
+ if (apiVersion >= CONSUMER_GENERATED_MEMBER_ID_REQUIRED_VERSION ||
+ request.memberEpoch() > 0 ||
+ request.memberEpoch() == LEAVE_GROUP_MEMBER_EPOCH
+ ) {
+ throwIfNull(request.memberId(), "MemberId can't be null.");
Review Comment:
Can it be null? It does not seem to be possible from the RPC definition.
##########
clients/src/main/resources/common/message/ShareGroupHeartbeatRequest.json:
##########
@@ -28,7 +28,7 @@
{ "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
"about": "The group identifier." },
{ "name": "MemberId", "type": "string", "versions": "0+",
- "about": "The member ID generated by the coordinator. The member ID must be kept during the entire lifetime of the member." },
+ "about": "The member ID generated by the consumer. The member ID must be kept during the entire lifetime of the consumer process." },
Review Comment:
ditto.
##########
core/src/test/scala/unit/kafka/server/DeleteGroupsRequestTest.scala:
##########
@@ -82,15 +84,17 @@ class DeleteGroupsRequestTest(cluster: ClusterInstance) extends GroupCoordinator
// We test DeleteGroups on empty and non-empty groups. Here we create the non-empty group.
joinConsumerGroup(
groupId = "grp-non-empty",
- useNewProtocol = useNewProtocol
+ useNewProtocol = useNewProtocol,
+ memberId = Uuid.randomUuid.toString
Review Comment:
It is a bit weird to have the member id here because the classic protocol does not use it. I wonder if we should just hide this detail and generate the member id within the joinConsumerGroup. What do you think?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -944,23 +941,20 @@ public boolean sameRequest(final OffsetFetchRequestState request) {
public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
OffsetFetchRequest.Builder builder;
- if (memberInfo.memberId.isPresent() && memberInfo.memberEpoch.isPresent()) {
- builder = new OffsetFetchRequest.Builder(
- groupId,
- memberInfo.memberId.get(),
- memberInfo.memberEpoch.get(),
- true,
- new ArrayList<>(this.requestedPartitions),
- throwOnFetchStableOffsetUnsupported);
- } else {
- // Building request without passing member ID/epoch to leave the logic to choose
- // default values when not present on the request builder.
- builder = new OffsetFetchRequest.Builder(
- groupId,
- true,
- new ArrayList<>(this.requestedPartitions),
- throwOnFetchStableOffsetUnsupported);
- }
+ // Building request without passing member ID/epoch to leave the logic to choose
+ // default values when not present on the request builder.
+ builder = memberInfo.memberEpoch.map(epoch -> new OffsetFetchRequest.Builder(
+ groupId,
+ memberInfo.memberId,
+ epoch,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported))
+ .orElseGet(() -> new OffsetFetchRequest.Builder(
+ groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported));
Review Comment:
For my understanding, this means that we only set the member id when we have a valid epoch too. In other words, we don't set the member id if we are not actively part of a group. Is my understanding correct?
I wonder if we should do the same for the groupMetadata in order to be consistent. What do you think?
##########
clients/src/main/resources/common/message/ConsumerGroupHeartbeatResponse.json:
##########
@@ -39,7 +39,7 @@
{ "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The top-level error message, or null if there was no error." },
{ "name": "MemberId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
- "about": "The member id generated by the coordinator. Only provided when the member joins with MemberEpoch == 0." },
+ "about": "The member id is generated by the consumer and provided by the consumer for all requests." },
Review Comment:
I wonder if we should precise that in version 0, it is generated by the group coordinator. However in version 1+, it is basically the value received from the consumer?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1293,21 +1294,28 @@ private void throwIfNull(
* Validates the request.
*
* @param request The request to validate.
- *
+ * @param apiVersion The version of ConsumerGroupHeartbeat RPC
* @throws InvalidRequestException if the request is not valid.
* @throws UnsupportedAssignorException if the assignor is not supported.
*/
private void throwIfConsumerGroupHeartbeatRequestIsInvalid(
- ConsumerGroupHeartbeatRequestData request
+ ConsumerGroupHeartbeatRequestData request,
+ short apiVersion
Review Comment:
nit: The indentation is incorrect. It should be 4 spaces.
##########
clients/src/main/resources/common/message/OffsetFetchRequest.json:
##########
@@ -54,7 +54,7 @@
{ "name": "GroupId", "type": "string", "versions": "8+", "entityType": "groupId",
"about": "The group ID."},
{ "name": "MemberId", "type": "string", "versions": "9+", "nullableVersions": "9+", "default": "null", "ignorable": true,
- "about": "The member ID assigned by the group coordinator if using the new consumer protocol (KIP-848)." },
+ "about": "The member ID generated by the consumer if using the new consumer protocol (KIP-848, KIP-1082)." },
Review Comment:
ditto.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10226,48 +10256,48 @@ public void testConsumerGroupHeartbeatToClassicGroupFromExistingStaticMember() {
.build();
List<CoordinatorRecord> expectedRecords = Arrays.asList(
- // The existing classic group tombstone.
- GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord(groupId),
-
- // Create the new consumer group with the static member.
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedClassicMember),
- GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 0),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId, expectedClassicMember.assignedPartitions()),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentEpochRecord(groupId, 0),
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedClassicMember),
-
- // Remove the static member because the rejoining member replaces it.
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentTombstoneRecord(groupId, memberId),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentTombstoneRecord(groupId, memberId),
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionTombstoneRecord(groupId, memberId),
+ // The existing classic group tombstone.
Review Comment:
Are they any changes in this block?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -281,13 +294,17 @@ public void testMemberIdGeneration() {
));
CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord> result = context.consumerGroupHeartbeat(
- new ConsumerGroupHeartbeatRequestData()
- .setGroupId("group-foo")
- .setMemberEpoch(0)
- .setServerAssignor("range")
- .setRebalanceTimeoutMs(5000)
- .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
- .setTopicPartitions(Collections.emptyList()));
+ // The consumer generates its own Member ID starting from version 1 of the ConsumerGroupHeartbeat RPC.
+ // Therefore, this test case is specific to earlier versions of the RPC.
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId("group-foo")
+ .setMemberEpoch(0)
+ .setServerAssignor("range")
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setTopicPartitions(Collections.emptyList()),
+ (short) (CONSUMER_GENERATED_MEMBER_ID_REQUIRED_VERSION - 1)
Review Comment:
nit: Let's use `0`. It is more explicit.
##########
core/src/test/scala/unit/kafka/server/ConsumerProtocolMigrationTest.scala:
##########
@@ -242,50 +243,54 @@ class ConsumerProtocolMigrationTest(cluster: ClusterInstance) extends GroupCoord
assignment = assignment(List(0, 1, 2))
)
- // The joining request with a consumer group member 2 is accepted.
- val memberId2 = consumerGroupHeartbeat(
- groupId = groupId,
- rebalanceTimeoutMs = 5 * 60 * 1000,
- subscribedTopicNames = List("foo"),
- topicPartitions = List.empty,
- expectedError = Errors.NONE
- ).memberId
-
- // The group has become a consumer group.
- assertEquals(
- List(
- new ListGroupsResponseData.ListedGroup()
- .setGroupId(groupId)
- .setProtocolType("consumer")
- .setGroupState(ConsumerGroupState.RECONCILING.toString)
- .setGroupType(Group.GroupType.CONSUMER.toString)
- ),
- listGroups(
- statesFilter = List.empty,
- typesFilter = List(Group.GroupType.CONSUMER.toString)
+ for (version <- ApiKeys.CONSUMER_GROUP_HEARTBEAT.oldestVersion() to ApiKeys.CONSUMER_GROUP_HEARTBEAT.latestVersion(isUnstableApiEnabled)) {
Review Comment:
Testing all the versions in this suite does not seem necessary. We can focus on testing the last one. However, let's ensure that we test all the versions in ConsumerGroupHeartbeatRequestTest.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1335,23 +1342,22 @@ private void throwIfConsumerGroupHeartbeatRequestIsInvalid(
* Validates the ShareGroupHeartbeat request.
*
* @param request The request to validate.
- *
* @throws InvalidRequestException if the request is not valid.
* @throws UnsupportedAssignorException if the assignor is not supported.
*/
private void throwIfShareGroupHeartbeatRequestIsInvalid(
ShareGroupHeartbeatRequestData request
) throws InvalidRequestException, UnsupportedAssignorException {
+ throwIfNull(request.memberId(), "MemberId can't be null.");
Review Comment:
ditto. I am not sure that it is necessary.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -281,13 +294,17 @@ public void testMemberIdGeneration() {
));
CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord> result = context.consumerGroupHeartbeat(
- new ConsumerGroupHeartbeatRequestData()
- .setGroupId("group-foo")
- .setMemberEpoch(0)
- .setServerAssignor("range")
- .setRebalanceTimeoutMs(5000)
- .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
- .setTopicPartitions(Collections.emptyList()));
+ // The consumer generates its own Member ID starting from version 1 of the ConsumerGroupHeartbeat RPC.
+ // Therefore, this test case is specific to earlier versions of the RPC.
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId("group-foo")
+ .setMemberEpoch(0)
+ .setServerAssignor("range")
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(Arrays.asList("foo", "bar"))
+ .setTopicPartitions(Collections.emptyList()),
+ (short) (CONSUMER_GENERATED_MEMBER_ID_REQUIRED_VERSION - 1)
Review Comment:
nit: Indentation seems incorrect. There are other cases in this file. I let you check them.
##########
core/src/test/scala/unit/kafka/server/DeleteGroupsRequestTest.scala:
##########
@@ -36,6 +37,7 @@ class DeleteGroupsRequestTest(cluster: ClusterInstance) extends GroupCoordinator
)
)
def testDeleteGroupsWithNewConsumerGroupProtocolAndNewGroupCoordinator(): Unit = {
+ // TODO fix
Review Comment:
Is it still relevant?
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1807321405
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -80,10 +81,12 @@ public abstract class AbstractMembershipManager<R extends AbstractResponse> impl
protected final String groupId;
/**
- * Member ID assigned by the server to the member, received in a heartbeat response when
- * joining the group specified in {@link #groupId}
+ * Member ID generated by the consumer at startup, which is unique within the group and remains consistent
+ * for the entire lifetime of the process. This ID acts as an incarnation identifier for the consumer process
+ * and does not reset or change, even if the consumer leaves and rejoins the group.
+ * The Member ID remains the same until the process is completely stopped or terminated.
*/
- protected String memberId = "";
+ protected final String memberId = Uuid.randomUuid().toString();
Review Comment:
According to KIP-1082, the `memberId` should remain unchanged until the process terminates, so I added the `final` modifier.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1807323514
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -224,6 +227,12 @@ public abstract class AbstractMembershipManager<R extends AbstractResponse> impl
this.clientTelemetryReporter = clientTelemetryReporter;
this.time = time;
this.metricsManager = metricsManager;
+
+ // Update the group member ID label in the client telemetry reporter.
+ // According to KIP-1082, the consumer will generate the member ID as the incarnation ID of the process.
+ // Therefore, we can update the group member ID during initialization.
+ clientTelemetryReporter.ifPresent(reporter -> reporter.updateMetricsLabels(
+ Collections.singletonMap(ClientTelemetryProvider.GROUP_MEMBER_ID, memberId)));
Review Comment:
Originally, this logic was triggered when the consumer received the `memberId` assigned by the server.
Since the consumer no longer relies on this mechanism, I moved the logic to the constructor.
This also guarantees that the logic is only called once.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1810581161
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1351,6 +1355,7 @@ private void throwIfShareGroupHeartbeatRequestIsInvalid(
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
+ throwIfEmptyString(request.memberId(), "MemberId can't be empty.");
Review Comment:
Since consumers are now required to generate their own `memberId` and the `ShareGroupHeartbeat RPC` is still under development, it seems feasible to check the `memberId` at the beginning:
```java
private void throwIfShareGroupHeartbeatRequestIsInvalid(
ShareGroupHeartbeatRequestData request
) throws InvalidRequestException, UnsupportedAssignorException {
throwIfEmptyString(request.groupId(), "GroupId can't be empty.");
throwIfEmptyString(request.rackId(), "RackId can't be empty.");
throwIfEmptyString(request.memberId(), "MemberId can't be empty."); // Always check memberId
if (request.memberEpoch() == 0) {
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
} else if (request.memberEpoch() < ShareGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH) {
throw new InvalidRequestException("MemberEpoch is invalid.");
}
}
```
Dose this make sense ?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1351,6 +1355,7 @@ private void throwIfShareGroupHeartbeatRequestIsInvalid(
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
+ throwIfEmptyString(request.memberId(), "MemberId can't be empty.");
Review Comment:
Since consumers are now required to generate their own `memberId` and the `ShareGroupHeartbeat RPC` is still under development, it seems feasible to check the `memberId` at the beginning:
```java
private void throwIfShareGroupHeartbeatRequestIsInvalid(
ShareGroupHeartbeatRequestData request
) throws InvalidRequestException, UnsupportedAssignorException {
throwIfEmptyString(request.groupId(), "GroupId can't be empty.");
throwIfEmptyString(request.rackId(), "RackId can't be empty.");
throwIfEmptyString(request.memberId(), "MemberId can't be empty."); // Always check memberId
if (request.memberEpoch() == 0) {
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
} else if (request.memberEpoch() < ShareGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH) {
throw new InvalidRequestException("MemberEpoch is invalid.");
}
}
```
Dose this make 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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1824604318
##########
clients/src/main/resources/common/message/ConsumerGroupHeartbeatRequest.json:
##########
@@ -18,15 +18,15 @@
"type": "request",
"listeners": ["zkBroker", "broker"],
Review Comment:
@frankvicky thanks for the quick acknowledgment, but this is unrelated to this PR. We can handle the cleanup in a follow-up.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1824527415
##########
clients/src/main/resources/common/message/ConsumerGroupHeartbeatRequest.json:
##########
@@ -18,15 +18,15 @@
"type": "request",
"listeners": ["zkBroker", "broker"],
Review Comment:
@dajac Does zk broker support this request?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMembershipManager.java:
##########
@@ -158,10 +157,15 @@ public ConsumerMembershipManager(String groupId,
commitRequestManager,
metadata,
logContext,
- clientTelemetryReporter,
backgroundEventHandler,
time,
new ConsumerRebalanceMetricsManager(metrics));
+
+ // Update the group member ID label in the client telemetry reporter.
Review Comment:
Could you please remove `clientTelemetryReporter` from this class? `ConsumerMembershipManager` does not use it actually. We can update `clientTelemetryReporter` after creating `ConsumerMembershipManager`. for example:
```java
membershipManager = new ConsumerMembershipManager(
groupRebalanceConfig.groupId,
groupRebalanceConfig.groupInstanceId,
groupRebalanceConfig.rebalanceTimeoutMs,
serverAssignor,
subscriptions,
commitRequestManager,
metadata,
logContext,
clientTelemetryReporter,
backgroundEventHandler,
time,
metrics);
// Update the group member ID label in the client telemetry reporter.
// According to KIP-1082, the consumer will generate the member ID as the incarnation ID of the process.
// Therefore, we can update the group member ID during initialization.
if (clientTelemetryReporter.isPresent())
clientTelemetryReporter.get().updateMetricsLabels(Map.of(ClientTelemetryProvider.GROUP_MEMBER_ID,
membershipManager.memberId()));
```
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1809775339
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1317,6 +1316,9 @@ private void throwIfConsumerGroupHeartbeatRequestIsInvalid(
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
+ if (apiVersion >= 2) {
Review Comment:
Cool.
I will add and name this constant as `CONSUMER_GENERATED_MEMBER_ID_REQUIRED_VERSION`.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on PR #17549:
URL: https://github.com/apache/kafka/pull/17549#issuecomment-2436851353
Hi @lianetm,
Here is my understanding. Please correct me if I’m wrong:
Since the group management has now moved to the server-side, the `ConsumerGroupMetadata` basically represents the data acknowledged by the broker. In this case, I think we should keep the `ConsumerGroupMetadata.memberId` empty at startup. This is because, at that moment, the broker still has no knowledge of the memberId. Once the consumer polls, the broker will then recognize the `memberId` and update the assignment, allowing the consumer to update the `ConsumerGroupMetadata` via `MemberStateListener`. After that, `ConsumerGroupMetadata` will have the `memberId`.
Does this make 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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on PR #17549:
URL: https://github.com/apache/kafka/pull/17549#issuecomment-2428817599
I changed this PR to ‘ready for review’ to leverage CI.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1814848836
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -1293,17 +1287,16 @@ public void setInflightCommitStatus(final boolean inflightCommitStatus) {
}
static class MemberInfo {
- Optional<String> memberId;
+ String memberId = "";
Review Comment:
Yes, I think it's very ideal! But after quickly going through `RequestManagers`, it seems that `MembershipManager` relies on `CommitRequestManager`. If that's the case, it could be tricky.
https://github.com/apache/kafka/blob/553e6b4c6d912b3b619b5a1a5263b616b6d3f62e/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java#L197-L219
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1810779333
##########
core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala:
##########
@@ -606,7 +608,7 @@ class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) {
protected def shareGroupHeartbeat(
groupId: String,
- memberId: String = "",
+ memberId: String = Uuid.randomUuid.toString,
Review Comment:
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1810581161
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1351,6 +1355,7 @@ private void throwIfShareGroupHeartbeatRequestIsInvalid(
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
+ throwIfEmptyString(request.memberId(), "MemberId can't be empty.");
Review Comment:
Since consumers are now required to generate their own `memberId` and the `ShareGroupHeartbeat RPC` is still under development, it seems feasible to check the `memberId` at the beginning:
```java
private void throwIfShareGroupHeartbeatRequestIsInvalid(
ShareGroupHeartbeatRequestData request
) throws InvalidRequestException, UnsupportedAssignorException {
throwIfEmptyString(request.groupId(), "GroupId can't be empty.");
throwIfEmptyString(request.rackId(), "RackId can't be empty.");
throwIfEmptyString(request.memberId(), "MemberId can't be empty."); // Always check memberId
if (request.memberEpoch() == 0) {
if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
}
} else if (request.memberEpoch() < ShareGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH) {
throw new InvalidRequestException("MemberEpoch is invalid.");
}
}
```
Dose this make 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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1819318737
##########
core/src/test/scala/unit/kafka/server/DeleteGroupsRequestTest.scala:
##########
@@ -82,15 +84,17 @@ class DeleteGroupsRequestTest(cluster: ClusterInstance) extends GroupCoordinator
// We test DeleteGroups on empty and non-empty groups. Here we create the non-empty group.
joinConsumerGroup(
groupId = "grp-non-empty",
- useNewProtocol = useNewProtocol
+ useNewProtocol = useNewProtocol,
+ memberId = Uuid.randomUuid.toString
Review Comment:
Thanks for pointing this out. 😺
I agree, and it could indeed be misleading in the current form.
I'll make the changes based on your suggestion.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1821082552
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -433,7 +442,7 @@ public void transitionToFatal() {
MemberState previousState = state;
transitionTo(MemberState.FATAL);
log.error("Member {} with epoch {} transitioned to fatal state", memberIdInfoForLog(), memberEpoch);
Review Comment:
this makes me notice that we can probably remove this `memberIdInfoForLog` now and simply use `memberId` (It was needed given there could be no member ID, but that's not the case anymore in this manager)
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -622,10 +631,10 @@ public void transitionToSendingLeaveGroup(boolean dueToExpiredPollTimer) {
/**
* Call all listeners that are registered to get notified when the member epoch is updated.
- * This also includes the latest member ID in the notification. If the member fails or leaves
- * the group, this will be invoked with empty epoch and member ID.
+ * This also includes the member ID in the notification. If the member fails or leaves
+ * the group, this will be invoked with empty epoch.
*/
- void notifyEpochChange(Optional<Integer> epoch, Optional<String> memberId) {
+ void notifyEpochChange(Optional<Integer> epoch, String memberId) {
stateUpdatesListeners.forEach(stateListener -> stateListener.onMemberEpochUpdated(epoch, memberId));
Review Comment:
let's update the java doc for this `stateUpdatesListeners` that still refers to a memberID received from the broker
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1820868382
##########
core/src/test/scala/unit/kafka/server/ConsumerGroupHeartbeatRequestTest.scala:
##########
@@ -36,54 +35,68 @@ import scala.collection.Map
import scala.jdk.CollectionConverters._
@ExtendWith(value = Array(classOf[ClusterTestExtensions]))
-class ConsumerGroupHeartbeatRequestTest(cluster: ClusterInstance) {
-
- @ClusterTest(
+@ClusterTestDefaults(
types = Array(Type.KRAFT),
serverProperties = Array(
- new ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, value = "classic"),
new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1")
)
+)
+class ConsumerGroupHeartbeatRequestTest(cluster: ClusterInstance) {
+
+ @ClusterTests(value = Array(
+ new ClusterTest(
+ serverProperties = Array(new ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, value = "classic")),
+ tags = Array("0")
+ ),
+ new ClusterTest(
+ serverProperties = Array(new ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, value = "classic")),
+ tags = Array("1")
+ )
+ )
)
Review Comment:
For example:
in the `testStaticMemberRemovedAfterSessionTimeoutExpiryWhenNewGroupCoordinatorIsEnabled` test case，we validate the `memberEpoch`．If we use a `for` loop in this case，since we don't restart clusters, we would need to set the expected value in a tricky way．
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #17549:
URL: https://github.com/apache/kafka/pull/17549#issuecomment-2449206268
@chia7712 @lianetm Could you please review it again, especially the client side? It seems that we are close to being able to merge 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1822143292
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -198,14 +209,14 @@ public void testConsumerHeartbeatRequestValidation() {
// SubscribedTopicNames must be present and empty in the first request (epoch == 0).
ex = assertThrows(InvalidRequestException.class, () -> context.consumerGroupHeartbeat(
new ConsumerGroupHeartbeatRequestData()
+ .setMemberId(memberId)
.setGroupId("foo")
.setMemberEpoch(0)
.setRebalanceTimeoutMs(5000)
.setTopicPartitions(Collections.emptyList())));
assertEquals("SubscribedTopicNames must be set in first request.", ex.getMessage());
- // MemberId must be non-empty in all requests except for the first one where it
- // could be empty (epoch != 0).
+ // MemberId must be non-empty in all requests
Review Comment:
This test case seems redundant now. Should we remove it?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10204,8 +10225,16 @@ public void testConsumerGroupHeartbeatToClassicGroupFromExistingStaticMember() {
mkTopicAssignment(fooTopicId, 0)))
.build();
- String newMemberId = result.response().memberId();
- ConsumerGroupMember expectedReplacingConsumerMember = new ConsumerGroupMember.Builder(newMemberId)
+ // The memberId is generated by the consumer and should be retained
+ // for the entire lifetime of the process until termination.
+ String serverReturnedMemberId = result.response().memberId();
+ assertEquals(
+ newMemberId,
+ serverReturnedMemberId,
+ "Server should not generate a new memberId since the consumer has already generated its own."
+ );
Review Comment:
nit: How about `assertEquals(newMemberId, result.response().memberId(), ...)`? We don't really need the `serverReturnedMemberId` local variable as we can directly use newMemberId.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10177,15 +10193,20 @@ public void testConsumerGroupHeartbeatToClassicGroupFromExistingStaticMember() {
context.replay(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(group, assignments, metadataImage.features().metadataVersion()));
context.commit();
- // The static member rejoins with new protocol, triggering the upgrade.
+ // The static member rejoins with new protocol after a restart, triggering the upgrade.
+ String newMemberId = Uuid.randomUuid().toString();
+ assertNotEquals(newMemberId, memberId, "The consumer should generate a new memberId since the process has been terminated and restarted.");
Review Comment:
nit: I would remove this one.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -14086,23 +14133,28 @@ public void testShareGroupMemberIdGeneration() {
Collections.emptyMap()
));
+ String memberId = Uuid.randomUuid().toString();
CoordinatorResult<ShareGroupHeartbeatResponseData, CoordinatorRecord> result = context.shareGroupHeartbeat(
new ShareGroupHeartbeatRequestData()
.setGroupId("group-foo")
+ .setMemberId(memberId)
.setMemberEpoch(0)
.setSubscribedTopicNames(Arrays.asList("foo", "bar")));
- // Verify that a member id was generated for the new member.
- String memberId = result.response().memberId();
- assertNotNull(memberId);
- assertNotEquals("", memberId);
+ String serverReturnedMemberId = result.response().memberId();
Review Comment:
ditto.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10367,14 +10392,26 @@ public void testConsumerGroupHeartbeatFromExistingClassicStaticMember() {
CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord> result = context.consumerGroupHeartbeat(
new ConsumerGroupHeartbeatRequestData()
.setGroupId(groupId)
+ .setMemberId(memberId1)
.setInstanceId(instanceId1)
.setRebalanceTimeoutMs(5000)
.setServerAssignor(NoOpPartitionAssignor.NAME)
.setSubscribedTopicNames(new ArrayList<>(member1.subscribedTopicNames()))
- .setTopicPartitions(Collections.emptyList()));
+ .setTopicPartitions(Collections.emptyList()),
+ ApiKeys.CONSUMER_GROUP_HEARTBEAT.latestVersion()
+ );
+
+
+ // The memberId is generated by the consumer itself, the consumer should retain this memberId
+ // for its entire lifetime until the process terminates.
+ String serverReturnedMemberId = result.response().memberId();
Review Comment:
ditto.
--
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-17116: New consumer may not send effective leave group if member ID received after close [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17549:
URL: https://github.com/apache/kafka/pull/17549#discussion_r1807326709
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -10225,53 +10251,61 @@ public void testConsumerGroupHeartbeatToClassicGroupFromExistingStaticMember() {
.setClassicMemberMetadata(null)
.build();
- List<CoordinatorRecord> expectedRecords = Arrays.asList(
- // The existing classic group tombstone.
- GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord(groupId),
-
- // Create the new consumer group with the static member.
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedClassicMember),
- GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 0),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId, expectedClassicMember.assignedPartitions()),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentEpochRecord(groupId, 0),
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedClassicMember),
-
- // Remove the static member because the rejoining member replaces it.
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentTombstoneRecord(groupId, memberId),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentTombstoneRecord(groupId, memberId),
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionTombstoneRecord(groupId, memberId),
-
- // Create the new static member.
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedReplacingConsumerMember),
- GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, newMemberId, mkAssignment(mkTopicAssignment(fooTopicId, 0))),
- GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedReplacingConsumerMember),
-
- // The static member rejoins the new consumer group.
- GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedFinalConsumerMember),
-
- // The subscription metadata hasn't been updated during the conversion, so a new one is computed.
- GroupCoordinatorRecordHelpers.newConsumerGroupSubscriptionMetadataRecord(groupId, new HashMap<String, TopicMetadata>() {
- {
- put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 1));
- }
- }),
+ List<CoordinatorRecord> expectedRecords = new ArrayList<>();
+ // The existing classic group tombstone.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newGroupMetadataTombstoneRecord(groupId));
+
+ // Create the new consumer group with the static member.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedClassicMember));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 0));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId,
+ expectedClassicMember.assignedPartitions()));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentEpochRecord(groupId, 0));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedClassicMember));
+
+ // Remove the static member because the rejoining member replaces it.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentTombstoneRecord(groupId, memberId));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentTombstoneRecord(groupId, memberId));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionTombstoneRecord(groupId, memberId));
+
+ // Create the new static member.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedReplacingConsumerMember));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, newMemberId,
+ mkAssignment(mkTopicAssignment(fooTopicId, 0))));
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedReplacingConsumerMember));
+
+ // The static member rejoins the new consumer group.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedFinalConsumerMember));
+
+ // The subscription metadata hasn't been updated during the conversion, so a new one is computed.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupSubscriptionMetadataRecord(groupId,
+ Map.of(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 1))));
+
+ // Newly joining static member bumps the group epoch. A new target assignment is computed.
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 1));
+
+ // If the memberId is generated by the consumer itself, the consumer should retain this memberId.
+ // As a result, the record won't contain a new target assignment record.
+ // If the memberId is not consumer-generated, add a new target assignment record to the expected records,
+ // since a different memberId will be considered as a new member.
+ if (!isConsumerGeneratedMemberId) {
+ expectedRecords.add(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, newMemberId,
+ mkAssignment(mkTopicAssignment(fooTopicId, 0))));
+ }
Review Comment:
Since the consumer's `memberId` won't change, it also means that a new target assignment won't occur.
Therefore, we don't need a new target assignment record here.
--
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