You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "aliehsaeedii (via GitHub)" <gi...@apache.org> on 2026/06/24 14:13:33 UTC
[PR] KAFKA-20717: Add broker-side set/get observability sensors for streams topology description (KIP-1331) [kafka]
aliehsaeedii opened a new pull request, #22664:
URL: https://github.com/apache/kafka/pull/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)
--
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-20717: Add broker-side set/get observability sensors for streams topology description (KIP-1331) [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #22664:
URL: https://github.com/apache/kafka/pull/22664#discussion_r3468140353
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java:
##########
@@ -73,6 +74,7 @@ public class StreamsGroupTopologyDescriptionManager implements AutoCloseable {
private final Logger log;
private final Optional<StreamsGroupTopologyDescriptionPlugin> plugin;
private final StreamsGroupTopologyDescriptionBackoff backoff;
+ private final GroupCoordinatorMetrics metrics;
Review Comment:
Recording the get sensors here in the manager — rather than in `GroupCoordinatorService` like set/delete — reads as an asymmetry at first, but it's actually the right placement. The get disposition (null return -> NOT_STORED/success, async-error/sync-throw/null-future -> error, converted -> success) is only finally resolved inside `maybeAttachOne` / `applyGetTopologyOutcome`; the service never sees that classification. Recording it here keeps the metric next to the single source of truth instead of leaking the outcome back out to the service just to count it, which is why the new `GroupCoordinatorMetrics` dependency belongs on this class. Set is the mirror image — its outcome is the `PluginOutcome.kind()` dispatched in the service's `thenCompose`, so recording there is equally natural. Good design.
We can add a one-line comment here noting the deliberate split so the next reader doesn't mistake it for an inconsistency with set/delete.
--
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-20717: Add broker-side set/get observability sensors for streams topology description (KIP-1331) [kafka]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #22664:
URL: https://github.com/apache/kafka/pull/22664#discussion_r3468263589
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -442,6 +446,46 @@ public GroupCoordinatorMetrics(MetricsRegistry registry, Metrics metrics) {
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 plugin.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 plugin.getTopology calls"),
Review Comment:
The get-error meter description says this metric counts failed *plugin.getTopology* calls, but `StreamsGroupTopologyDescriptionManager.applyGetTopologyOutcome` also records this sensor when response conversion throws. To avoid misleading operators, update the description to reflect that this metric covers any getTopology operation resulting in an ERROR response (including conversion failures / SPI contract violations), or stop recording conversion failures against this sensor.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -442,6 +446,46 @@ public GroupCoordinatorMetrics(MetricsRegistry registry, Metrics metrics) {
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 plugin.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 plugin.getTopology calls"),
+ metrics.metricName("streams-group-topology-description-get-error-count",
+ METRICS_GROUP,
+ "The total number of failed plugin.getTopology calls")));
Review Comment:
Same issue as the get-error rate description: this count description currently claims to measure failed *plugin.getTopology* calls, but conversion errors are also recorded against the get-error sensor. Update the description to match the actual semantics.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -442,6 +446,46 @@ public GroupCoordinatorMetrics(MetricsRegistry registry, Metrics metrics) {
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 plugin.getTopology calls (a call returning null counts as success)"),
Review Comment:
This description says the metric counts successful *plugin.getTopology* calls, but the implementation records `get-error` (not `get-success`) when conversion to the wire response fails even though the plugin call itself succeeded. Consider updating the description to reflect that this metric tracks successful getTopology *operations* (i.e., those that do not surface ERROR), including null returns.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -442,6 +446,46 @@ public GroupCoordinatorMetrics(MetricsRegistry registry, Metrics metrics) {
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 plugin.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)")));
Review Comment:
Same issue as the get-success rate description: the count description currently claims to measure successful *plugin.getTopology* calls, but conversion failures are counted as get-error despite a successful plugin return. Update this string to match the actual semantics (successful getTopology operations).
--
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-20717: Add broker-side set/get observability sensors for streams topology description (KIP-1331) [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #22664:
URL: https://github.com/apache/kafka/pull/22664#discussion_r3469639444
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java:
##########
@@ -70,8 +72,11 @@
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.spy;
Review Comment:
I wonder if there is a way to avoid spy, mockStatic and so on. We have made bad experience with using these "advanced" mockito features (memory problems etc.). If possible, I'd avoid them.
--
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-20717: Add broker-side set/get observability sensors for streams topology description (KIP-1331) [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #22664:
URL: https://github.com/apache/kafka/pull/22664#discussion_r3473268325
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java:
##########
@@ -73,6 +74,7 @@ public class StreamsGroupTopologyDescriptionManager implements AutoCloseable {
private final Logger log;
private final Optional<StreamsGroupTopologyDescriptionPlugin> plugin;
private final StreamsGroupTopologyDescriptionBackoff backoff;
+ private final GroupCoordinatorMetrics metrics;
Review Comment:
Agreed on the rationale — adding a one-line comment would be helpful. 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-20717: Add broker-side set/get observability sensors for streams topology description (KIP-1331) [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #22664:
URL: https://github.com/apache/kafka/pull/22664#discussion_r3469611124
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java:
##########
@@ -73,6 +74,7 @@ public class StreamsGroupTopologyDescriptionManager implements AutoCloseable {
private final Logger log;
private final Optional<StreamsGroupTopologyDescriptionPlugin> plugin;
private final StreamsGroupTopologyDescriptionBackoff backoff;
+ private final GroupCoordinatorMetrics metrics;
Review Comment:
Not sure I buy the second part - we could equally move the set recording to the GroupMetricsManager. But I guess it's fine, since we also record other metrics here.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java:
##########
@@ -533,24 +539,40 @@ private void applyGetTopologyOutcome(
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;
}
try {
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() {
Review Comment:
I'd just inline the one-liners.
--
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