frankvicky commented on code in PR #17008:
URL: https://github.com/apache/kafka/pull/17008#discussion_r1733986782
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1121,7 +1135,7 @@ private <T> CoordinatorResult<T, CoordinatorRecord>
convertToClassicGroup(
metrics.onClassicGroupStateTransition(classicGroup.currentState(),
null);
return null;
});
- return new CoordinatorResult<>(records, response, appendFuture, false);
+ return new CoordinatorResult<>(records, null, appendFuture, false);
Review Comment:
A constructor already handles the `response = null` situation.
`new CoordinatorResult<>(records, appendFuture, false);`
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -3202,25 +3214,34 @@ public void testSessionTimeoutExpirationStaticMember() {
List<ExpiredTimeout<Void, CoordinatorRecord>> timeouts =
context.sleep(45000 + 1);
// Verify the expired timeout.
+ assertEquals(1, timeouts.size());
+ ExpiredTimeout<Void, CoordinatorRecord> timeout = timeouts.get(0);
+ assertEquals(groupSessionTimeoutKey(groupId, memberId), timeout.key);
assertEquals(
- Collections.singletonList(new ExpiredTimeout<Void,
CoordinatorRecord>(
- groupSessionTimeoutKey(groupId, memberId),
- new CoordinatorResult<>(
- Arrays.asList(
-
GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentTombstoneRecord(groupId,
memberId),
-
GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentTombstoneRecord(groupId,
memberId),
-
GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionTombstoneRecord(groupId,
memberId),
-
GroupCoordinatorRecordHelpers.newConsumerGroupSubscriptionMetadataRecord(groupId,
Collections.emptyMap()),
-
GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 2)
- )
- )
- )),
- timeouts
+ Arrays.asList(
+
GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentTombstoneRecord(groupId,
memberId),
+
GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentTombstoneRecord(groupId,
memberId),
+
GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionTombstoneRecord(groupId,
memberId),
+
GroupCoordinatorRecordHelpers.newConsumerGroupSubscriptionMetadataRecord(groupId,
Collections.emptyMap()),
+
GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 2)
+ ),
+ timeout.result.records()
);
- // Verify that there are no timers.
+ // Verify that there is a downgrade timer scheduled if the append
future is completed without exception.
+ timeout.result.appendFuture().complete(null);
context.assertNoSessionTimeout(groupId, memberId);
context.assertNoRebalanceTimeout(groupId, memberId);
+ context.assertDowngradeTimeout(groupId);
+
+ // The downgrade is not triggered.
+ assertEquals(
+ new ExpiredTimeout<Void, CoordinatorRecord>(
+ consumerGroupDowngradeKey(groupId),
+ new CoordinatorResult<>(Collections.emptyList())
+ ),
+ context.sleep(0).get(0)
+ );
Review Comment:
ditto
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -3121,25 +3124,34 @@ public void testSessionTimeoutExpiration() {
List<ExpiredTimeout<Void, CoordinatorRecord>> timeouts =
context.sleep(45000 + 1);
// Verify the expired timeout.
+ assertEquals(1, timeouts.size());
+ ExpiredTimeout<Void, CoordinatorRecord> timeout = timeouts.get(0);
+ assertEquals(groupSessionTimeoutKey(groupId, memberId), timeout.key);
assertEquals(
- Collections.singletonList(new ExpiredTimeout<Void,
CoordinatorRecord>(
- groupSessionTimeoutKey(groupId, memberId),
- new CoordinatorResult<>(
- Arrays.asList(
-
GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentTombstoneRecord(groupId,
memberId),
-
GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentTombstoneRecord(groupId,
memberId),
-
GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionTombstoneRecord(groupId,
memberId),
-
GroupCoordinatorRecordHelpers.newConsumerGroupSubscriptionMetadataRecord(groupId,
Collections.emptyMap()),
-
GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 2)
- )
- )
- )),
- timeouts
+ Arrays.asList(
+
GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentTombstoneRecord(groupId,
memberId),
+
GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentTombstoneRecord(groupId,
memberId),
+
GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionTombstoneRecord(groupId,
memberId),
+
GroupCoordinatorRecordHelpers.newConsumerGroupSubscriptionMetadataRecord(groupId,
Collections.emptyMap()),
+
GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 2)
+ ),
+ timeout.result.records()
);
- // Verify that there are no timers.
+ // Verify that there is a downgrade timer scheduled if the append
future is completed without exception.
+ timeout.result.appendFuture().complete(null);
context.assertNoSessionTimeout(groupId, memberId);
context.assertNoRebalanceTimeout(groupId, memberId);
+ context.assertDowngradeTimeout(groupId);
+
+ // The downgrade is not triggered.
+ assertEquals(
+ new ExpiredTimeout<Void, CoordinatorRecord>(
+ consumerGroupDowngradeKey(groupId),
+ new CoordinatorResult<>(Collections.emptyList())
+ ),
+ context.sleep(0).get(0)
+ );
Review Comment:
nit: I think it will be more clear if we could utilize message argument
instead of comment.
For example:
```java
assertEquals(
new ExpiredTimeout<Void, CoordinatorRecord>(
consumerGroupDowngradeKey(groupId),
new CoordinatorResult<>(Collections.emptyList())
),
context.sleep(0).get(0),
"message of assertion fail"
);
```
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -3499,24 +3520,33 @@ public void testRebalanceTimeoutExpiration() {
List<ExpiredTimeout<Void, CoordinatorRecord>> timeouts =
context.sleep(10000 + 1);
// Verify the expired timeout.
+ assertEquals(1, timeouts.size());
+ ExpiredTimeout<Void, CoordinatorRecord> timeout = timeouts.get(0);
+ assertEquals(consumerGroupRebalanceTimeoutKey(groupId, memberId1),
timeout.key);
assertEquals(
- Collections.singletonList(new ExpiredTimeout<Void,
CoordinatorRecord>(
- consumerGroupRebalanceTimeoutKey(groupId, memberId1),
- new CoordinatorResult<>(
- Arrays.asList(
-
GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentTombstoneRecord(groupId,
memberId1),
-
GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentTombstoneRecord(groupId,
memberId1),
-
GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionTombstoneRecord(groupId,
memberId1),
-
GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 3)
- )
- )
- )),
- timeouts
+ Arrays.asList(
+
GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentTombstoneRecord(groupId,
memberId1),
+
GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentTombstoneRecord(groupId,
memberId1),
+
GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionTombstoneRecord(groupId,
memberId1),
+
GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 3)
+ ),
+ timeout.result.records()
);
- // Verify that there are no timers.
+ // Verify that there is a downgrade timer scheduled if the append
future is completed without exception.
+ timeout.result.appendFuture().complete(null);
context.assertNoSessionTimeout(groupId, memberId1);
context.assertNoRebalanceTimeout(groupId, memberId1);
+ context.assertDowngradeTimeout(groupId);
+
+ // The downgrade is not triggered.
+ assertEquals(
+ new ExpiredTimeout<Void, CoordinatorRecord>(
+ consumerGroupDowngradeKey(groupId),
+ new CoordinatorResult<>(Collections.emptyList())
+ ),
+ context.sleep(0).get(0)
+ );
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
