You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "jeffkbkim (via GitHub)" <gi...@apache.org> on 2024/08/13 23:28:22 UTC
[PR] KAFKA-16106: revert classic state transitions if deletion fails [kafka]
jeffkbkim opened a new pull request, #16874:
URL: https://github.com/apache/kafka/pull/16874
An expire-group-metadata operation generates tombstone records, updates the groups state and decrements group size counters, then performs a write to the log. If there is a __consumer_offsets partition reassignment, for instance, this operation fails. The groups state is reverted to an earlier snapshot but classic group size counters are not. This begins an inconsistency between the metrics and the actual groups size. This applies to all unsuccessful write operations that alter the classic group state.
However, some operations that alter the classic group state does not produce records. This means that we cannot rely on timeline data structures as we do for consumer group states.
### 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-16106: revert classic state transitions if deletion fails [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #16874:
URL: https://github.com/apache/kafka/pull/16874#discussion_r1716537013
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3984,6 +3984,8 @@ CoordinatorResult<Void, CoordinatorRecord> classicGroupJoinToClassicGroup(
Review Comment:
We have another case in this file where we delete a classic group. Take a look at `maybeDeleteEmptyClassicGroup` method.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -621,18 +623,32 @@ public CoordinatorResult<OffsetDeleteResponseData, CoordinatorRecord> deleteOffs
public CoordinatorResult<Void, CoordinatorRecord> cleanupGroupMetadata() {
long startMs = time.milliseconds();
List<CoordinatorRecord> records = new ArrayList<>();
+ AtomicInteger deletedClassicGroupCount = new AtomicInteger(0);
groupMetadataManager.groupIds().forEach(groupId -> {
boolean allOffsetsExpired = offsetMetadataManager.cleanupExpiredOffsets(groupId, records);
if (allOffsetsExpired) {
- groupMetadataManager.maybeDeleteGroup(groupId, records);
+ if (groupMetadataManager.maybeDeleteGroup(groupId, records)) {
+ if (groupMetadataManager.group(groupId).type() == Group.GroupType.CLASSIC) {
+ deletedClassicGroupCount.incrementAndGet();
Review Comment:
I wonder if we need this logic in `deleteGroups` too. What do you think?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -621,18 +623,32 @@ public CoordinatorResult<OffsetDeleteResponseData, CoordinatorRecord> deleteOffs
public CoordinatorResult<Void, CoordinatorRecord> cleanupGroupMetadata() {
long startMs = time.milliseconds();
List<CoordinatorRecord> records = new ArrayList<>();
+ AtomicInteger deletedClassicGroupCount = new AtomicInteger(0);
groupMetadataManager.groupIds().forEach(groupId -> {
boolean allOffsetsExpired = offsetMetadataManager.cleanupExpiredOffsets(groupId, records);
if (allOffsetsExpired) {
- groupMetadataManager.maybeDeleteGroup(groupId, records);
+ if (groupMetadataManager.maybeDeleteGroup(groupId, records)) {
+ if (groupMetadataManager.group(groupId).type() == Group.GroupType.CLASSIC) {
+ deletedClassicGroupCount.incrementAndGet();
+ }
+ }
}
});
log.info("Generated {} tombstone records while cleaning up group metadata in {} milliseconds.",
records.size(), time.milliseconds() - startMs);
// Reschedule the next cycle.
scheduleGroupMetadataExpiration();
- return new CoordinatorResult<>(records, false);
+
+ // If the append operation fails, revert classic group state transitions. Groups were only deleted
+ // if they were in Empty state.
+ CompletableFuture<Void> appendFuture = new CompletableFuture<>();
+ appendFuture.exceptionally(__ -> {
+ IntStream.range(0, deletedClassicGroupCount.get()).forEach(___ ->
+ metricsShard.onClassicGroupStateTransition(null, ClassicGroupState.EMPTY));
+ return null;
+ });
+ return new CoordinatorResult<>(records, null, appendFuture, true, false);
Review Comment:
nit: Let's add an empty line before this one.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3984,6 +3984,8 @@ CoordinatorResult<Void, CoordinatorRecord> classicGroupJoinToClassicGroup(
responseFuture.complete(new JoinGroupResponseData()
.setErrorCode(appendGroupMetadataErrorToResponseError(Errors.forException(t)).code()));
+
+ metrics.onClassicGroupStateTransition(EMPTY, null);
Review Comment:
Should we add a unit test to cover this one too?
--
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