You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2023/12/13 08:03:18 UTC
(kafka) branch 3.7 updated: KAFKA-15981: update Group size only when groups size changes (#14988)
This is an automated email from the ASF dual-hosted git repository.
dajac pushed a commit to branch 3.7
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.7 by this push:
new c47468fb812 KAFKA-15981: update Group size only when groups size changes (#14988)
c47468fb812 is described below
commit c47468fb81220e8004cd1579c24d8bcd576a93ca
Author: Jeff Kim <ki...@gmail.com>
AuthorDate: Wed Dec 13 03:01:56 2023 -0500
KAFKA-15981: update Group size only when groups size changes (#14988)
Currently, we increment generic group metrics whenever we create a new Group object when we load a partition. This is incorrect as the partition may contain several records for the same group if in the active segment or if the segment has not yet been compacted.
The same applies to removing groups; we can possibly have multiple group tombstone records. Instead, only increment the metric if we created a new group and only decrement the metric if the group exists.
Reviewers: David Jacot <dj...@confluent.io>
---
.../coordinator/group/GroupMetadataManager.java | 27 ++++--
.../coordinator/group/consumer/ConsumerGroup.java | 2 -
.../coordinator/group/generic/GenericGroup.java | 1 -
.../group/GroupMetadataManagerTest.java | 98 ++++++++++++++++++++++
.../group/consumer/ConsumerGroupTest.java | 43 ++++++++++
.../group/generic/GenericGroupTest.java | 24 ++++++
.../metrics/GroupCoordinatorMetricsShardTest.java | 11 +++
7 files changed, 195 insertions(+), 11 deletions(-)
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
index c394680ac80..6f9f4a92d52 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
@@ -575,6 +575,7 @@ public class GroupMetadataManager {
if (group == null) {
ConsumerGroup consumerGroup = new ConsumerGroup(snapshotRegistry, groupId, metrics);
groups.put(groupId, consumerGroup);
+ metrics.onConsumerGroupStateTransition(null, consumerGroup.state());
return consumerGroup;
} else {
if (group.type() == CONSUMER) {
@@ -613,6 +614,7 @@ public class GroupMetadataManager {
if (group == null) {
GenericGroup genericGroup = new GenericGroup(logContext, groupId, GenericGroupState.EMPTY, time, metrics);
groups.put(groupId, genericGroup);
+ metrics.onGenericGroupStateTransition(null, genericGroup.currentState());
return genericGroup;
} else {
if (group.type() == GENERIC) {
@@ -684,7 +686,19 @@ public class GroupMetadataManager {
private void removeGroup(
String groupId
) {
- groups.remove(groupId);
+ Group group = groups.remove(groupId);
+ if (group != null) {
+ switch (group.type()) {
+ case CONSUMER:
+ ConsumerGroup consumerGroup = (ConsumerGroup) group;
+ metrics.onConsumerGroupStateTransition(consumerGroup.state(), null);
+ break;
+ case GENERIC:
+ GenericGroup genericGroup = (GenericGroup) group;
+ metrics.onGenericGroupStateTransition(genericGroup.currentState(), null);
+ break;
+ }
+ }
}
/**
@@ -1594,7 +1608,6 @@ public class GroupMetadataManager {
+ " but did not receive ConsumerGroupTargetAssignmentMetadataValue tombstone.");
}
removeGroup(groupId);
- metrics.onConsumerGroupStateTransition(consumerGroup.state(), null);
}
}
@@ -1803,11 +1816,6 @@ public class GroupMetadataManager {
if (value == null) {
// Tombstone. Group should be removed.
- Group group = groups.get(groupId);
- if (group != null && group.type() == GENERIC) {
- GenericGroup genericGroup = (GenericGroup) group;
- metrics.onGenericGroupStateTransition(genericGroup.currentState(), null);
- }
removeGroup(groupId);
} else {
List<GenericGroupMember> loadedMembers = new ArrayList<>();
@@ -1851,7 +1859,10 @@ public class GroupMetadataManager {
);
loadedMembers.forEach(member -> genericGroup.add(member, null));
- groups.put(groupId, genericGroup);
+ Group prevGroup = groups.put(groupId, genericGroup);
+ if (prevGroup == null) {
+ metrics.onGenericGroupStateTransition(null, genericGroup.currentState());
+ }
genericGroup.setSubscribedTopics(
genericGroup.computeSubscribedTopics()
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java
index 3beaa5de1ef..216b06ac253 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java
@@ -190,8 +190,6 @@ public class ConsumerGroup implements Group {
this.targetAssignment = new TimelineHashMap<>(snapshotRegistry, 0);
this.currentPartitionEpoch = new TimelineHashMap<>(snapshotRegistry, 0);
this.metrics = Objects.requireNonNull(metrics);
-
- metrics.onConsumerGroupStateTransition(null, this.state.get());
}
/**
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroup.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroup.java
index 8903ef93213..9ff3c16a0f4 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroup.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroup.java
@@ -236,7 +236,6 @@ public class GenericGroup implements Group {
this.protocolName = protocolName;
this.leaderId = leaderId;
this.currentStateTimestamp = currentStateTimestamp;
- metrics.onGenericGroupStateTransition(null, initialState);
}
/**
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
index 1bfbf4ff8ac..3fde13937f6 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
@@ -159,6 +159,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -10682,6 +10683,103 @@ public class GroupMetadataManagerTest {
verify(context.metrics).record(CONSUMER_GROUP_REBALANCES_SENSOR_NAME);
}
+ @Test
+ public void testOnGenericGroupStateTransition() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // Creating a generic group should increment metric.
+ GenericGroup group = context.createGenericGroup("group-id");
+ verify(context.metrics, times(1)).onGenericGroupStateTransition(null, EMPTY);
+
+ // Replaying a new group should not increment metric as the group was already created.
+ context.replay(RecordHelpers.newGroupMetadataRecord(group, Collections.emptyMap(), MetadataVersion.LATEST_PRODUCTION));
+ verify(context.metrics, times(1)).onGenericGroupStateTransition(null, EMPTY);
+
+ // Loading a tombstone should remove group and decrement metric.
+ context.createGenericGroup("group-id");
+ context.replay(RecordHelpers.newGroupMetadataTombstoneRecord("group-id"));
+ verify(context.metrics, times(1)).onGenericGroupStateTransition(EMPTY, null);
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.group("group-id"));
+
+ // Replaying a tombstone for a group that has already been deleted should not decrement metric.
+ context.replay(RecordHelpers.newGroupMetadataTombstoneRecord("group-id"));
+ verify(context.metrics, times(1)).onGenericGroupStateTransition(EMPTY, null);
+ }
+
+ @Test
+ public void testOnGenericGroupStateTransitionOnLoading() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ GenericGroup group = new GenericGroup(
+ new LogContext(),
+ "group-id",
+ EMPTY,
+ context.time,
+ context.metrics
+ );
+
+ // Even if there are more group metadata records loaded than tombstone records, the last replayed record
+ // (tombstone in this test) is the latest state of the group. Hence, the overall metric count should be 0.
+ IntStream.range(0, 5).forEach(__ -> {
+ context.replay(RecordHelpers.newGroupMetadataRecord(group, Collections.emptyMap(), MetadataVersion.LATEST_PRODUCTION));
+ });
+ IntStream.range(0, 4).forEach(__ -> {
+ context.replay(RecordHelpers.newGroupMetadataTombstoneRecord("group-id"));
+ });
+
+ verify(context.metrics, times(1)).onGenericGroupStateTransition(null, EMPTY);
+ verify(context.metrics, times(1)).onGenericGroupStateTransition(EMPTY, null);
+ }
+
+ @Test
+ public void testOnConsumerGroupStateTransition() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // Replaying a consumer group epoch record should increment metric.
+ context.replay(RecordHelpers.newGroupEpochRecord("group-id", 1));
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(null, ConsumerGroup.ConsumerGroupState.EMPTY);
+
+ // Replaying a consumer group epoch record for a group that has already been created should not increment metric.
+ context.replay(RecordHelpers.newGroupEpochRecord("group-id", 1));
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(null, ConsumerGroup.ConsumerGroupState.EMPTY);
+
+ // Creating and replaying tombstones for a group should remove group and decrement metric.
+ List<Record> tombstones = new ArrayList<>();
+ Group group = context.groupMetadataManager.group("group-id");
+ group.createGroupTombstoneRecords(tombstones);
+ tombstones.forEach(context::replay);
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.group("group-id"));
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.EMPTY, null);
+
+ // Replaying a tombstone for a group that has already been removed should not decrement metric.
+ tombstones.forEach(tombstone -> assertThrows(GroupIdNotFoundException.class, () -> context.replay(tombstone)));
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.EMPTY, null);
+ }
+
+ @Test
+ public void testOnConsumerGroupStateTransitionOnLoading() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // Even if there are more group epoch records loaded than tombstone records, the last replayed record
+ // (tombstone in this test) is the latest state of the group. Hence, the overall metric count should be 0.
+ IntStream.range(0, 5).forEach(__ -> {
+ context.replay(RecordHelpers.newGroupEpochRecord("group-id", 0));
+ });
+ context.replay(RecordHelpers.newTargetAssignmentEpochTombstoneRecord("group-id"));
+ context.replay(RecordHelpers.newGroupEpochTombstoneRecord("group-id"));
+ IntStream.range(0, 3).forEach(__ -> {
+ assertThrows(GroupIdNotFoundException.class, () -> context.replay(RecordHelpers.newTargetAssignmentEpochTombstoneRecord("group-id")));
+ assertThrows(GroupIdNotFoundException.class, () -> context.replay(RecordHelpers.newGroupEpochTombstoneRecord("group-id")));
+ });
+
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(null, ConsumerGroup.ConsumerGroupState.EMPTY);
+ verify(context.metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.EMPTY, null);
+ }
+
private static void assertNoOrEmptyResult(List<ExpiredTimeout<Void, Record>> timeouts) {
assertTrue(timeouts.size() <= 1);
timeouts.forEach(timeout -> assertEquals(EMPTY_RESULT, timeout.result));
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupTest.java
index 7263181322f..71cbf3d6afb 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupTest.java
@@ -53,6 +53,8 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
public class ConsumerGroupTest {
@@ -875,4 +877,45 @@ public class ConsumerGroupTest {
assertEquals(expected, actual);
}
+
+ @Test
+ public void testStateTransitionMetrics() {
+ // Confirm metrics is not updated when a new ConsumerGroup is created but only when the group transitions
+ // its state.
+ GroupCoordinatorMetricsShard metrics = mock(GroupCoordinatorMetricsShard.class);
+ ConsumerGroup consumerGroup = new ConsumerGroup(
+ new SnapshotRegistry(new LogContext()),
+ "group-id",
+ metrics
+ );
+
+ assertEquals(ConsumerGroup.ConsumerGroupState.EMPTY, consumerGroup.state());
+ verify(metrics, times(0)).onConsumerGroupStateTransition(null, ConsumerGroup.ConsumerGroupState.EMPTY);
+
+ ConsumerGroupMember member = new ConsumerGroupMember.Builder("member")
+ .setMemberEpoch(1)
+ .setPreviousMemberEpoch(0)
+ .setTargetMemberEpoch(1)
+ .build();
+
+ consumerGroup.updateMember(member);
+
+ assertEquals(ConsumerGroup.ConsumerGroupState.RECONCILING, consumerGroup.state());
+ verify(metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.EMPTY, ConsumerGroup.ConsumerGroupState.RECONCILING);
+
+ consumerGroup.setGroupEpoch(1);
+
+ assertEquals(ConsumerGroup.ConsumerGroupState.ASSIGNING, consumerGroup.state());
+ verify(metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.RECONCILING, ConsumerGroup.ConsumerGroupState.ASSIGNING);
+
+ consumerGroup.setTargetAssignmentEpoch(1);
+
+ assertEquals(ConsumerGroup.ConsumerGroupState.STABLE, consumerGroup.state());
+ verify(metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.ASSIGNING, ConsumerGroup.ConsumerGroupState.STABLE);
+
+ consumerGroup.removeMember("member");
+
+ assertEquals(ConsumerGroup.ConsumerGroupState.EMPTY, consumerGroup.state());
+ verify(metrics, times(1)).onConsumerGroupStateTransition(ConsumerGroup.ConsumerGroupState.STABLE, ConsumerGroup.ConsumerGroupState.EMPTY);
+ }
}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/generic/GenericGroupTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/generic/GenericGroupTest.java
index 6aa36541a5b..d956cda4b33 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/generic/GenericGroupTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/generic/GenericGroupTest.java
@@ -62,7 +62,10 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
public class GenericGroupTest {
private final String protocolType = "consumer";
@@ -1251,6 +1254,27 @@ public class GenericGroupTest {
assertTrue(group.isSubscribedToTopic("topic"));
}
+ @Test
+ public void testStateTransitionMetrics() {
+ // Confirm metrics is not updated when a new GenericGroup is created but only when the group transitions
+ // its state.
+ GroupCoordinatorMetricsShard metrics = mock(GroupCoordinatorMetricsShard.class);
+ GenericGroup group = new GenericGroup(new LogContext(), "groupId", EMPTY, Time.SYSTEM, metrics);
+ verify(metrics, times(0)).onGenericGroupStateTransition(any(), any());
+
+ group.transitionTo(PREPARING_REBALANCE);
+ verify(metrics, times(1)).onGenericGroupStateTransition(EMPTY, PREPARING_REBALANCE);
+
+ group.transitionTo(COMPLETING_REBALANCE);
+ verify(metrics, times(1)).onGenericGroupStateTransition(PREPARING_REBALANCE, COMPLETING_REBALANCE);
+
+ group.transitionTo(STABLE);
+ verify(metrics, times(1)).onGenericGroupStateTransition(COMPLETING_REBALANCE, STABLE);
+
+ group.transitionTo(DEAD);
+ verify(metrics, times(1)).onGenericGroupStateTransition(STABLE, DEAD);
+ }
+
private void assertState(GenericGroup group, GenericGroupState targetState) {
Set<GenericGroupState> otherStates = new HashSet<>();
otherStates.add(STABLE);
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsShardTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsShardTest.java
index b9d9df6e6b3..371aa944f52 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsShardTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsShardTest.java
@@ -29,6 +29,7 @@ import org.apache.kafka.timeline.SnapshotRegistry;
import org.junit.jupiter.api.Test;
import java.util.Collections;
+import java.util.stream.IntStream;
import static org.apache.kafka.coordinator.group.generic.GenericGroupState.COMPLETING_REBALANCE;
import static org.apache.kafka.coordinator.group.generic.GenericGroupState.DEAD;
@@ -130,6 +131,11 @@ public class GroupCoordinatorMetricsShardTest {
GenericGroup group2 = new GenericGroup(logContext, "groupId2", EMPTY, Time.SYSTEM, shard);
GenericGroup group3 = new GenericGroup(logContext, "groupId3", EMPTY, Time.SYSTEM, shard);
+ IntStream.range(0, 4).forEach(__ -> {
+ shard.incrementLocalGauge(NUM_GENERIC_GROUPS);
+ shard.incrementGlobalGauge(NUM_GENERIC_GROUPS_EMPTY);
+ });
+
snapshotRegistry.getOrCreateSnapshot(1000);
shard.commitUpTo(1000);
assertEquals(4, shard.localGaugeValue(NUM_GENERIC_GROUPS));
@@ -198,6 +204,11 @@ public class GroupCoordinatorMetricsShardTest {
shard
);
+ IntStream.range(0, 4).forEach(__ -> {
+ shard.incrementLocalGauge(NUM_CONSUMER_GROUPS);
+ shard.incrementLocalGauge(NUM_CONSUMER_GROUPS_EMPTY);
+ });
+
snapshotRegistry.getOrCreateSnapshot(1000);
shard.commitUpTo(1000);
assertEquals(4, shard.localGaugeValue(NUM_CONSUMER_GROUPS));