This is an automated email from the ASF dual-hosted git repository.
lucasbru pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new fb557a346a8 KAFKA-20717: Add broker-side set/get observability sensors
for streams topology description (KIP-1331) (#22664)
fb557a346a8 is described below
commit fb557a346a866113e212cb4f7c844222d54a9aa1
Author: Alieh Saeedi <[email protected]>
AuthorDate: Mon Jun 29 23:35:01 2026 +0200
KAFKA-20717: Add broker-side set/get observability sensors for streams
topology description (KIP-1331) (#22664)
## Summary
Implements the broker-side observability sensors for the streams-group
topology-description plugin's `setTopology` and `getTopology`
operations, as specified in
[KIP-1331](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1331%3A+Streams+Group+Topology+Description+Plugin)
([KAFKA-20717](https://issues.apache.org/jira/browse/KAFKA-20717)).
The `delete` and `cleanup` sensors (KAFKA-20623/KAFKA-20624) are already
in place; this change mirrors their pattern for the remaining
`set`/`get` sensors. New sensors registered under
`kafka.server:type=group-coordinator-metrics`:
- `streams-group-topology-description-set-success-{rate,count}`
- `streams-group-topology-description-set-error-{rate,count}`
- `streams-group-topology-description-get-success-{rate,count}`
- `streams-group-topology-description-get-error-{rate,count}`
## Implementation notes
- **Set**: recorded at the `invokeSetTopology` dispatch in
`GroupCoordinatorService`. `SUCCESS` increments `set-success`; both
`PERMANENT` and `TRANSIENT` fold into `set-error` (any failure type
counts as an error, per the KIP).
- **Get**: recorded in `StreamsGroupTopologyDescriptionManager`, where
`plugin.getTopology` resolves — the only place the per-call outcome
surfaces. A `null` return counts as a **success** (it maps to
`NOT_STORED`, not an error); exceptional/synchronous-throw/null-future
outcomes count as `get-error`.
## Testing
- `GroupCoordinatorMetricsTest`: added the 8 new metric names to the
registration assertions and a new test verifying `recordSensor` drives
the correct meters.
- `GroupCoordinatorServiceTopologyDescriptionTest`: 6 new tests covering
set success/permanent/transient and get available/null-return/error
outcomes.
- Full `GroupCoordinatorServiceTest`, checkstyle, and spotbugs pass.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Reviewers: Lucas Brutschy <[email protected]>, TengYao Chi
<[email protected]>
---------
Co-authored-by: Claude Opus 4.8 (1M context) <[email protected]>
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
---
.../coordinator/group/GroupCoordinatorService.java | 65 +++---
.../group/metrics/GroupCoordinatorMetrics.java | 60 +++++-
.../StreamsGroupTopologyDescriptionManager.java | 28 ++-
...pCoordinatorServiceTopologyDescriptionTest.java | 233 ++++++++++++++++++++-
.../group/metrics/GroupCoordinatorMetricsTest.java | 33 +++
5 files changed, 390 insertions(+), 29 deletions(-)
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
index 8d10af7dd32..579e9726f93 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
@@ -427,7 +427,8 @@ public class GroupCoordinatorService implements
GroupCoordinator {
this.streamsGroupTopologyDescriptionManager = new
StreamsGroupTopologyDescriptionManager(
logContext,
streamsGroupTopologyDescriptionPlugin,
- time
+ time,
+ groupCoordinatorMetrics
);
}
@@ -743,29 +744,32 @@ public class GroupCoordinatorService implements
GroupCoordinator {
.thenApply(__ ->
StreamsGroupTopologyDescriptionConverter.fromRequest(request.topologyDescription()))
.thenCompose(description ->
streamsGroupTopologyDescriptionManager.invokeSetTopology(
groupId, pushedEpoch, description))
- .thenCompose(pluginOutcome -> switch (pluginOutcome.kind()) {
- case SUCCESS -> runtime.scheduleWriteOperation(
- "streams-group-set-stored-topology-epoch",
- tp,
- coordinator ->
coordinator.setStoredDescriptionTopologyEpoch(groupId, pushedEpoch)
- ).handle((unused, throwable) ->
streamsGroupTopologyDescriptionManager.completeEpochWrite(
- groupId, pushedEpoch, throwable,
- new StreamsGroupTopologyDescriptionUpdateResponseData()));
- case PERMANENT -> runtime.scheduleWriteOperation(
- "streams-group-set-failed-topology-epoch",
- tp,
- coordinator ->
coordinator.setFailedDescriptionTopologyEpoch(groupId, pushedEpoch)
- ).handle((unused, throwable) ->
streamsGroupTopologyDescriptionManager.completeEpochWrite(
- groupId, pushedEpoch, throwable,
- new StreamsGroupTopologyDescriptionUpdateResponseData()
-
.setErrorCode(Errors.STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED.code())
- .setErrorMessage(pluginOutcome.message())));
- case TRANSIENT -> {
- streamsGroupTopologyDescriptionManager.armBackoff(groupId,
pushedEpoch);
- yield CompletableFuture.completedFuture(new
StreamsGroupTopologyDescriptionUpdateResponseData()
-
.setErrorCode(Errors.STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED.code())
- .setErrorMessage(pluginOutcome.message()));
- }
+ .thenCompose(pluginOutcome -> {
+ recordPluginSetOutcome(pluginOutcome.kind());
+ return switch (pluginOutcome.kind()) {
+ case SUCCESS -> runtime.scheduleWriteOperation(
+ "streams-group-set-stored-topology-epoch",
+ tp,
+ coordinator ->
coordinator.setStoredDescriptionTopologyEpoch(groupId, pushedEpoch)
+ ).handle((unused, throwable) ->
streamsGroupTopologyDescriptionManager.completeEpochWrite(
+ groupId, pushedEpoch, throwable,
+ new
StreamsGroupTopologyDescriptionUpdateResponseData()));
+ case PERMANENT -> runtime.scheduleWriteOperation(
+ "streams-group-set-failed-topology-epoch",
+ tp,
+ coordinator ->
coordinator.setFailedDescriptionTopologyEpoch(groupId, pushedEpoch)
+ ).handle((unused, throwable) ->
streamsGroupTopologyDescriptionManager.completeEpochWrite(
+ groupId, pushedEpoch, throwable,
+ new StreamsGroupTopologyDescriptionUpdateResponseData()
+
.setErrorCode(Errors.STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED.code())
+ .setErrorMessage(pluginOutcome.message())));
+ case TRANSIENT -> {
+
streamsGroupTopologyDescriptionManager.armBackoff(groupId, pushedEpoch);
+ yield CompletableFuture.completedFuture(new
StreamsGroupTopologyDescriptionUpdateResponseData()
+
.setErrorCode(Errors.STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED.code())
+ .setErrorMessage(pluginOutcome.message()));
+ }
+ };
})
.exceptionally(exception -> handleOperationException(
"streams-group-topology-description-update",
@@ -906,6 +910,19 @@ public class GroupCoordinatorService implements
GroupCoordinator {
}
}
+ /**
+ * Record the outcome of a single {@code plugin.setTopology} call against
the
+ * {@code set-success} / {@code set-error} sensors. A {@code SUCCESS}
outcome increments
+ * the success sensor; every failure outcome ({@code PERMANENT} or {@code
TRANSIENT},
+ * regardless of the underlying exception type) increments the error
sensor.
+ */
+ private void
recordPluginSetOutcome(StreamsGroupTopologyDescriptionManager.PluginOutcome.Kind
kind) {
+ String sensorName = kind ==
StreamsGroupTopologyDescriptionManager.PluginOutcome.Kind.SUCCESS
+ ?
GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_SET_SUCCESS_SENSOR_NAME
+ :
GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_SET_ERROR_SENSOR_NAME;
+ groupCoordinatorMetrics.recordSensor(sensorName);
+ }
+
/**
* Conditional metadata write that clears {@code
StoredDescriptionTopologyEpoch} for
* {@code groupId} only when the persisted value still equals {@code
expectedStoredEpoch}.
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java
index 185ea52d83a..1fb54b6309b 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java
@@ -126,6 +126,10 @@ public class GroupCoordinatorMetrics extends
CoordinatorMetrics implements AutoC
public static final String
STREAMS_GROUP_TOPOLOGY_DESCRIPTION_CLEANUP_ELIGIBLE_GROUPS_SENSOR_NAME =
"StreamsGroupTopologyDescriptionCleanupEligibleGroups";
public static final String
STREAMS_GROUP_TOPOLOGY_DESCRIPTION_DELETE_SUCCESS_SENSOR_NAME =
"StreamsGroupTopologyDescriptionDeleteSuccess";
public static final String
STREAMS_GROUP_TOPOLOGY_DESCRIPTION_DELETE_ERROR_SENSOR_NAME =
"StreamsGroupTopologyDescriptionDeleteError";
+ public static final String
STREAMS_GROUP_TOPOLOGY_DESCRIPTION_SET_SUCCESS_SENSOR_NAME =
"StreamsGroupTopologyDescriptionSetSuccess";
+ public static final String
STREAMS_GROUP_TOPOLOGY_DESCRIPTION_SET_ERROR_SENSOR_NAME =
"StreamsGroupTopologyDescriptionSetError";
+ public static final String
STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_SUCCESS_SENSOR_NAME =
"StreamsGroupTopologyDescriptionGetSuccess";
+ public static final String
STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_ERROR_SENSOR_NAME =
"StreamsGroupTopologyDescriptionGetError";
private final MetricName offsetCountMetricName;
private final MetricName classicGroupCountMetricName;
@@ -442,6 +446,46 @@ public class GroupCoordinatorMetrics extends
CoordinatorMetrics implements AutoC
METRICS_GROUP,
"The total number of failed plugin.deleteTopology calls
(DeleteGroups and periodic cleanup combined)")));
+ Sensor streamsGroupTopologyDescriptionSetSuccessSensor =
+
metrics.sensor(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_SET_SUCCESS_SENSOR_NAME);
+ streamsGroupTopologyDescriptionSetSuccessSensor.add(new Meter(
+
metrics.metricName("streams-group-topology-description-set-success-rate",
+ METRICS_GROUP,
+ "The rate of successful plugin.setTopology calls"),
+
metrics.metricName("streams-group-topology-description-set-success-count",
+ METRICS_GROUP,
+ "The total number of successful plugin.setTopology calls")));
+
+ Sensor streamsGroupTopologyDescriptionSetErrorSensor =
+
metrics.sensor(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_SET_ERROR_SENSOR_NAME);
+ streamsGroupTopologyDescriptionSetErrorSensor.add(new Meter(
+
metrics.metricName("streams-group-topology-description-set-error-rate",
+ METRICS_GROUP,
+ "The rate of failed plugin.setTopology calls (any failure:
permanent, transient, or otherwise)"),
+
metrics.metricName("streams-group-topology-description-set-error-count",
+ METRICS_GROUP,
+ "The total number of failed plugin.setTopology calls (any
failure: permanent, transient, or otherwise)")));
+
+ Sensor streamsGroupTopologyDescriptionGetSuccessSensor =
+
metrics.sensor(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_SUCCESS_SENSOR_NAME);
+ streamsGroupTopologyDescriptionGetSuccessSensor.add(new Meter(
+
metrics.metricName("streams-group-topology-description-get-success-rate",
+ METRICS_GROUP,
+ "The rate of successful getTopology calls (a call returning
null counts as success)"),
+
metrics.metricName("streams-group-topology-description-get-success-count",
+ METRICS_GROUP,
+ "The total number of successful plugin.getTopology calls (a
call returning null counts as success)")));
+
+ Sensor streamsGroupTopologyDescriptionGetErrorSensor =
+
metrics.sensor(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_ERROR_SENSOR_NAME);
+ streamsGroupTopologyDescriptionGetErrorSensor.add(new Meter(
+
metrics.metricName("streams-group-topology-description-get-error-rate",
+ METRICS_GROUP,
+ "The rate of failed getTopology operations (plugin errors, SPI
contract violations, or conversion failures)"),
+
metrics.metricName("streams-group-topology-description-get-error-count",
+ METRICS_GROUP,
+ "The total number of failed plugin.getTopology calls")));
+
globalSensors = Collections.unmodifiableMap(Utils.mkMap(
Utils.mkEntry(OFFSET_COMMITS_SENSOR_NAME, offsetCommitsSensor),
Utils.mkEntry(OFFSET_EXPIRED_SENSOR_NAME, offsetExpiredSensor),
@@ -457,7 +501,15 @@ public class GroupCoordinatorMetrics extends
CoordinatorMetrics implements AutoC
Utils.mkEntry(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_DELETE_SUCCESS_SENSOR_NAME,
streamsGroupTopologyDescriptionDeleteSuccessSensor),
Utils.mkEntry(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_DELETE_ERROR_SENSOR_NAME,
- streamsGroupTopologyDescriptionDeleteErrorSensor)
+ streamsGroupTopologyDescriptionDeleteErrorSensor),
+
Utils.mkEntry(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_SET_SUCCESS_SENSOR_NAME,
+ streamsGroupTopologyDescriptionSetSuccessSensor),
+
Utils.mkEntry(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_SET_ERROR_SENSOR_NAME,
+ streamsGroupTopologyDescriptionSetErrorSensor),
+
Utils.mkEntry(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_SUCCESS_SENSOR_NAME,
+ streamsGroupTopologyDescriptionGetSuccessSensor),
+
Utils.mkEntry(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_ERROR_SENSOR_NAME,
+ streamsGroupTopologyDescriptionGetErrorSensor)
));
}
@@ -566,7 +618,11 @@ public class GroupCoordinatorMetrics extends
CoordinatorMetrics implements AutoC
STREAMS_GROUP_TOPOLOGY_DESCRIPTION_CLEANUP_CYCLE_RUNS_SENSOR_NAME,
STREAMS_GROUP_TOPOLOGY_DESCRIPTION_CLEANUP_ELIGIBLE_GROUPS_SENSOR_NAME,
STREAMS_GROUP_TOPOLOGY_DESCRIPTION_DELETE_SUCCESS_SENSOR_NAME,
- STREAMS_GROUP_TOPOLOGY_DESCRIPTION_DELETE_ERROR_SENSOR_NAME
+ STREAMS_GROUP_TOPOLOGY_DESCRIPTION_DELETE_ERROR_SENSOR_NAME,
+ STREAMS_GROUP_TOPOLOGY_DESCRIPTION_SET_SUCCESS_SENSOR_NAME,
+ STREAMS_GROUP_TOPOLOGY_DESCRIPTION_SET_ERROR_SENSOR_NAME,
+ STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_SUCCESS_SENSOR_NAME,
+ STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_ERROR_SENSOR_NAME
).forEach(metrics::removeSensor);
}
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java
index 5955155b9fa..8a81bab8616 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java
@@ -31,6 +31,7 @@ import org.apache.kafka.common.utils.internals.LogContext;
import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription;
import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescriptionPlugin;
import
org.apache.kafka.coordinator.group.api.streams.StreamsTopologyDescriptionPermanentFailureException;
+import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics;
import org.apache.kafka.server.util.timer.Timer;
import org.apache.kafka.server.util.timer.TimerTask;
@@ -73,6 +74,9 @@ public class StreamsGroupTopologyDescriptionManager
implements AutoCloseable {
private final Logger log;
private final Optional<StreamsGroupTopologyDescriptionPlugin> plugin;
private final StreamsGroupTopologyDescriptionBackoff backoff;
+ // Get sensors are recorded here, not in GroupCoordinatorService like
set/delete:
+ // the get outcome is only finally classified inside
applyGetTopologyOutcome, so the metric lives next to that single source of
truth.
+ private final GroupCoordinatorMetrics metrics;
/**
* True between {@link #startCleanupCycle} and {@link #close}. The {@link
TimerTask}
@@ -100,11 +104,13 @@ public class StreamsGroupTopologyDescriptionManager
implements AutoCloseable {
public StreamsGroupTopologyDescriptionManager(
LogContext logContext,
Optional<StreamsGroupTopologyDescriptionPlugin> plugin,
- Time time
+ Time time,
+ GroupCoordinatorMetrics metrics
) {
this.log =
logContext.logger(StreamsGroupTopologyDescriptionManager.class);
this.plugin = plugin;
this.backoff = new StreamsGroupTopologyDescriptionBackoff(time);
+ this.metrics = metrics;
}
/**
@@ -511,10 +517,12 @@ public class StreamsGroupTopologyDescriptionManager
implements AutoCloseable {
pluginFuture = p.getTopology(describedGroup.groupId(),
topologyEpoch);
} catch (Exception e) {
// SPI contract violation: synchronous throw treated as ERROR.
+ recordGetError();
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_ERROR);
return CompletableFuture.completedFuture(null);
}
if (pluginFuture == null) {
+ recordGetError();
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_ERROR);
return CompletableFuture.completedFuture(null);
}
@@ -533,10 +541,15 @@ public class StreamsGroupTopologyDescriptionManager
implements AutoCloseable {
Throwable cause = Errors.maybeUnwrapException(throwable);
log.warn("Topology description plugin getTopology failed for group
{}.",
describedGroup.groupId(), cause != null ? cause : throwable);
+ recordGetError();
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_ERROR);
return;
}
+ // The plugin call itself completed normally. A null return is the
documented
+ // "plugin no longer has the data" path and surfaces as NOT_STORED,
not an error,
+ // so it still counts as a successful getTopology.
if (topology == null) {
+ recordGetSuccess();
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_NOT_STORED);
return;
}
@@ -544,13 +557,24 @@ public class StreamsGroupTopologyDescriptionManager
implements AutoCloseable {
describedGroup.setTopologyDescription(
StreamsGroupTopologyDescriptionConverter.toDescribeResponse(topology));
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_AVAILABLE);
+ recordGetSuccess();
} catch (Exception conversionError) {
- // Defensive catch, should be unreachable in practice
+ // Defensive catch, should be unreachable in practice. The outcome
surfaces as
+ // ERROR to the client, so count it as a failed getTopology to
stay consistent.
+ recordGetError();
describedGroup.setTopologyDescription(null);
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_ERROR);
}
}
+ private void recordGetSuccess() {
+
metrics.recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_SUCCESS_SENSOR_NAME);
+ }
+
+ private void recordGetError() {
+
metrics.recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_ERROR_SENSOR_NAME);
+ }
+
// Visible for testing.
StreamsGroupTopologyDescriptionBackoff backoff() {
return backoff;
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java
index 204e1e95509..668b7011590 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java
@@ -42,10 +42,12 @@ import
org.apache.kafka.coordinator.group.api.streams.StreamsTopologyDescription
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics;
import org.apache.kafka.coordinator.group.streams.StreamsGroupDescribeResult;
import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
+import
org.apache.kafka.coordinator.group.streams.StreamsGroupTopologyDescriptionConverter;
import org.apache.kafka.server.share.persister.NoOpStatePersister;
import org.apache.kafka.server.util.timer.MockTimer;
import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
import java.time.Duration;
import java.util.Collections;
@@ -70,7 +72,9 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -103,13 +107,23 @@ public class
GroupCoordinatorServiceTopologyDescriptionTest {
Optional<StreamsGroupTopologyDescriptionPlugin> plugin,
boolean startup,
MockTimer timer
+ ) {
+ return buildService(runtime, plugin, startup, timer, new
GroupCoordinatorMetrics());
+ }
+
+ private static GroupCoordinatorService buildService(
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime,
+ Optional<StreamsGroupTopologyDescriptionPlugin> plugin,
+ boolean startup,
+ MockTimer timer,
+ GroupCoordinatorMetrics metrics
) {
MockTime time = timer.time();
GroupCoordinatorService service = new GroupCoordinatorService(
new LogContext(),
GroupCoordinatorConfigTest.createGroupCoordinatorConfig(4096,
600000L, 24),
runtime,
- new GroupCoordinatorMetrics(),
+ metrics,
createConfigManager(),
new NoOpStatePersister(),
timer,
@@ -1413,6 +1427,223 @@ public class
GroupCoordinatorServiceTopologyDescriptionTest {
verify(plugin, never()).getTopology(anyString(), anyInt());
}
+ @Test
+ public void testUpdateSuccessRecordsSetSuccessSensor() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(plugin.setTopology(anyString(), anyInt(), any()))
+ .thenReturn(CompletableFuture.completedFuture(null));
+
when(runtime.scheduleReadOperation(eq("streams-group-topology-description-validate"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(null));
+
when(runtime.scheduleWriteOperation(eq("streams-group-set-stored-topology-epoch"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(null));
+
+ GroupCoordinatorMetrics metrics = mock(GroupCoordinatorMetrics.class);
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true, new MockTimer(), metrics);
+
+ service.streamsGroupTopologyDescriptionUpdate(
+ requestContext(ApiKeys.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_UPDATE),
validUpdateRequest()
+ ).get(5, TimeUnit.SECONDS);
+
+
verify(metrics).recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_SET_SUCCESS_SENSOR_NAME);
+ verify(metrics,
never()).recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_SET_ERROR_SENSOR_NAME);
+ }
+
+ @Test
+ public void testUpdatePermanentFailureRecordsSetErrorSensor() throws
Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(plugin.setTopology(anyString(), anyInt(), any()))
+ .thenReturn(CompletableFuture.failedFuture(
+ new StreamsTopologyDescriptionPermanentFailureException("too
large")));
+
when(runtime.scheduleReadOperation(eq("streams-group-topology-description-validate"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(null));
+
when(runtime.scheduleWriteOperation(eq("streams-group-set-failed-topology-epoch"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(null));
+
+ GroupCoordinatorMetrics metrics = mock(GroupCoordinatorMetrics.class);
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true, new MockTimer(), metrics);
+
+ service.streamsGroupTopologyDescriptionUpdate(
+ requestContext(ApiKeys.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_UPDATE),
validUpdateRequest()
+ ).get(5, TimeUnit.SECONDS);
+
+
verify(metrics).recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_SET_ERROR_SENSOR_NAME);
+ verify(metrics,
never()).recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_SET_SUCCESS_SENSOR_NAME);
+ }
+
+ @Test
+ public void testUpdateTransientFailureRecordsSetErrorSensor() throws
Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(plugin.setTopology(anyString(), anyInt(), any()))
+ .thenReturn(CompletableFuture.failedFuture(new
RuntimeException("backend offline")));
+
when(runtime.scheduleReadOperation(eq("streams-group-topology-description-validate"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(null));
+
+ GroupCoordinatorMetrics metrics = mock(GroupCoordinatorMetrics.class);
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true, new MockTimer(), metrics);
+
+ service.streamsGroupTopologyDescriptionUpdate(
+ requestContext(ApiKeys.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_UPDATE),
validUpdateRequest()
+ ).get(5, TimeUnit.SECONDS);
+
+ // A transient failure is still a failed plugin.setTopology call: it
folds into set-error,
+ // not a separate sensor.
+
verify(metrics).recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_SET_ERROR_SENSOR_NAME);
+ verify(metrics,
never()).recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_SET_SUCCESS_SENSOR_NAME);
+ }
+
+ @Test
+ public void testDescribeAvailableRecordsGetSuccessSensor() throws
Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ StreamsGroupTopologyDescription pojo = new
StreamsGroupTopologyDescription(
+ List.of(new StreamsGroupTopologyDescription.Subtopology("sub-0",
List.of(
+ new StreamsGroupTopologyDescription.Source("src",
Set.of("input"), Set.of())))),
+ List.of()
+ );
+ when(plugin.getTopology("foo",
5)).thenReturn(CompletableFuture.completedFuture(pojo));
+ when(runtime.scheduleReadOperation(eq("streams-group-describe"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(
+ new
StreamsGroupDescribeResult(List.of(describedGroupWithTopology("foo", 5)),
Map.of("foo", 5))));
+
+ GroupCoordinatorMetrics metrics = mock(GroupCoordinatorMetrics.class);
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true, new MockTimer(), metrics);
+
+ service.streamsGroupDescribe(
+ requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE), List.of("foo"),
true
+ ).get(5, TimeUnit.SECONDS);
+
+
verify(metrics).recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_SUCCESS_SENSOR_NAME);
+ verify(metrics,
never()).recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_ERROR_SENSOR_NAME);
+ }
+
+ @Test
+ public void testDescribeNullReturnRecordsGetSuccessSensor() throws
Exception {
+ // A getTopology returning null is a successful plugin call that found
no data
+ // (surfaces as NOT_STORED) and must count as get-success, not
get-error.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(plugin.getTopology("foo",
5)).thenReturn(CompletableFuture.completedFuture(null));
+ when(runtime.scheduleReadOperation(eq("streams-group-describe"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(
+ new
StreamsGroupDescribeResult(List.of(describedGroupWithTopology("foo", 5)),
Map.of("foo", 5))));
+
+ GroupCoordinatorMetrics metrics = mock(GroupCoordinatorMetrics.class);
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true, new MockTimer(), metrics);
+
+ service.streamsGroupDescribe(
+ requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE), List.of("foo"),
true
+ ).get(5, TimeUnit.SECONDS);
+
+
verify(metrics).recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_SUCCESS_SENSOR_NAME);
+ verify(metrics,
never()).recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_ERROR_SENSOR_NAME);
+ }
+
+ @Test
+ public void testDescribeErrorRecordsGetErrorSensor() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(plugin.getTopology("foo", 5))
+ .thenReturn(CompletableFuture.failedFuture(new
RuntimeException("plugin offline")));
+ when(runtime.scheduleReadOperation(eq("streams-group-describe"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(
+ new
StreamsGroupDescribeResult(List.of(describedGroupWithTopology("foo", 5)),
Map.of("foo", 5))));
+
+ GroupCoordinatorMetrics metrics = mock(GroupCoordinatorMetrics.class);
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true, new MockTimer(), metrics);
+
+ service.streamsGroupDescribe(
+ requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE), List.of("foo"),
true
+ ).get(5, TimeUnit.SECONDS);
+
+
verify(metrics).recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_ERROR_SENSOR_NAME);
+ verify(metrics,
never()).recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_SUCCESS_SENSOR_NAME);
+ }
+
+ @Test
+ public void testDescribeSynchronousThrowRecordsGetErrorSensor() throws
Exception {
+ // A getTopology that throws synchronously violates the SPI contract
and surfaces as
+ // ERROR; it must count as get-error.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(plugin.getTopology("foo", 5)).thenThrow(new
RuntimeException("plugin blew up"));
+ when(runtime.scheduleReadOperation(eq("streams-group-describe"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(
+ new
StreamsGroupDescribeResult(List.of(describedGroupWithTopology("foo", 5)),
Map.of("foo", 5))));
+
+ GroupCoordinatorMetrics metrics = mock(GroupCoordinatorMetrics.class);
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true, new MockTimer(), metrics);
+
+ service.streamsGroupDescribe(
+ requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE), List.of("foo"),
true
+ ).get(5, TimeUnit.SECONDS);
+
+
verify(metrics).recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_ERROR_SENSOR_NAME);
+ verify(metrics,
never()).recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_SUCCESS_SENSOR_NAME);
+ }
+
+ @Test
+ public void testDescribeNullFutureRecordsGetErrorSensor() throws Exception
{
+ // A getTopology that returns a null future violates the SPI contract
and surfaces as
+ // ERROR; it must count as get-error.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(plugin.getTopology("foo", 5)).thenReturn(null);
+ when(runtime.scheduleReadOperation(eq("streams-group-describe"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(
+ new
StreamsGroupDescribeResult(List.of(describedGroupWithTopology("foo", 5)),
Map.of("foo", 5))));
+
+ GroupCoordinatorMetrics metrics = mock(GroupCoordinatorMetrics.class);
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true, new MockTimer(), metrics);
+
+ service.streamsGroupDescribe(
+ requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE), List.of("foo"),
true
+ ).get(5, TimeUnit.SECONDS);
+
+
verify(metrics).recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_ERROR_SENSOR_NAME);
+ verify(metrics,
never()).recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_SUCCESS_SENSOR_NAME);
+ }
+
+ @Test
+ public void testDescribeConversionErrorRecordsGetErrorSensor() throws
Exception {
+ // The plugin returns a valid topology, but converting it to the wire
response throws.
+ // That defensive catch in applyGetTopologyOutcome is otherwise
unreachable (the Node type
+ // is sealed to Source/Processor/Sink, so a well-formed topology
always converts), so we
+ // force toDescribeResponse to throw to exercise it. The outcome
surfaces as ERROR to the
+ // client and must count as get-error, not get-success.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ StreamsGroupTopologyDescription pojo = new
StreamsGroupTopologyDescription(
+ List.of(new StreamsGroupTopologyDescription.Subtopology("sub-0",
List.of(
+ new StreamsGroupTopologyDescription.Source("src",
Set.of("input"), Set.of())))),
+ List.of()
+ );
+ when(plugin.getTopology("foo",
5)).thenReturn(CompletableFuture.completedFuture(pojo));
+ when(runtime.scheduleReadOperation(eq("streams-group-describe"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(
+ new
StreamsGroupDescribeResult(List.of(describedGroupWithTopology("foo", 5)),
Map.of("foo", 5))));
+
+ GroupCoordinatorMetrics metrics = mock(GroupCoordinatorMetrics.class);
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true, new MockTimer(), metrics);
+
+ // CALLS_REAL_METHODS so only toDescribeResponse is overridden; any
other converter call
+ // (none on this path) keeps its real behavior.
+ try (MockedStatic<StreamsGroupTopologyDescriptionConverter> converter =
+ mockStatic(StreamsGroupTopologyDescriptionConverter.class,
CALLS_REAL_METHODS)) {
+ converter.when(() ->
StreamsGroupTopologyDescriptionConverter.toDescribeResponse(any()))
+ .thenThrow(new RuntimeException("conversion failed"));
+
+ service.streamsGroupDescribe(
+ requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE),
List.of("foo"), true
+ ).get(5, TimeUnit.SECONDS);
+ }
+
+
verify(metrics).recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_ERROR_SENSOR_NAME);
+ verify(metrics,
never()).recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_SUCCESS_SENSOR_NAME);
+ }
+
private static StreamsGroupDescribeResponseData.DescribedGroup
describedGroupWithTopology(
String groupId, int topologyEpoch
) {
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsTest.java
index df031f07935..cdb0e832626 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsTest.java
@@ -158,6 +158,14 @@ public class GroupCoordinatorMetricsTest {
metrics.metricName("streams-group-topology-description-delete-success-count",
GroupCoordinatorMetrics.METRICS_GROUP),
metrics.metricName("streams-group-topology-description-delete-error-rate",
GroupCoordinatorMetrics.METRICS_GROUP),
metrics.metricName("streams-group-topology-description-delete-error-count",
GroupCoordinatorMetrics.METRICS_GROUP),
+
metrics.metricName("streams-group-topology-description-set-success-rate",
GroupCoordinatorMetrics.METRICS_GROUP),
+
metrics.metricName("streams-group-topology-description-set-success-count",
GroupCoordinatorMetrics.METRICS_GROUP),
+
metrics.metricName("streams-group-topology-description-set-error-rate",
GroupCoordinatorMetrics.METRICS_GROUP),
+
metrics.metricName("streams-group-topology-description-set-error-count",
GroupCoordinatorMetrics.METRICS_GROUP),
+
metrics.metricName("streams-group-topology-description-get-success-rate",
GroupCoordinatorMetrics.METRICS_GROUP),
+
metrics.metricName("streams-group-topology-description-get-success-count",
GroupCoordinatorMetrics.METRICS_GROUP),
+
metrics.metricName("streams-group-topology-description-get-error-rate",
GroupCoordinatorMetrics.METRICS_GROUP),
+
metrics.metricName("streams-group-topology-description-get-error-count",
GroupCoordinatorMetrics.METRICS_GROUP),
metrics.metricName(
"streams-group-count",
GroupCoordinatorMetrics.METRICS_GROUP,
@@ -407,6 +415,31 @@ public class GroupCoordinatorMetricsTest {
), 50);
}
+ @Test
+ public void testStreamsGroupTopologyDescriptionSetAndGetSensors() {
+ MetricsRegistry registry = new MetricsRegistry();
+ Time time = new MockTime();
+ Metrics metrics = new Metrics(time);
+ GroupCoordinatorMetrics coordinatorMetrics = new
GroupCoordinatorMetrics(registry, metrics);
+
+
coordinatorMetrics.recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_SET_SUCCESS_SENSOR_NAME);
+
coordinatorMetrics.recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_SET_ERROR_SENSOR_NAME);
+
coordinatorMetrics.recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_SET_ERROR_SENSOR_NAME);
+
coordinatorMetrics.recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_SUCCESS_SENSOR_NAME);
+
coordinatorMetrics.recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_SUCCESS_SENSOR_NAME);
+
coordinatorMetrics.recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_SUCCESS_SENSOR_NAME);
+
coordinatorMetrics.recordSensor(GroupCoordinatorMetrics.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_GET_ERROR_SENSOR_NAME);
+
+ assertMetricValue(metrics, metrics.metricName(
+ "streams-group-topology-description-set-success-count",
GroupCoordinatorMetrics.METRICS_GROUP), 1);
+ assertMetricValue(metrics, metrics.metricName(
+ "streams-group-topology-description-set-error-count",
GroupCoordinatorMetrics.METRICS_GROUP), 2);
+ assertMetricValue(metrics, metrics.metricName(
+ "streams-group-topology-description-get-success-count",
GroupCoordinatorMetrics.METRICS_GROUP), 3);
+ assertMetricValue(metrics, metrics.metricName(
+ "streams-group-topology-description-get-error-count",
GroupCoordinatorMetrics.METRICS_GROUP), 1);
+ }
+
private void assertMetricValue(Metrics metrics, MetricName metricName,
double val) {
assertEquals(val, metrics.metric(metricName).metricValue());
}
