KAFKA-17248 - KIP 1076 implementation #17021

bbejeck · 2024-08-27T22:25:47Z

Implementation of KIP-1076 to allow for adding client application metrics to the KIP-714 framework

Integration and unit tests forthcoming

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

apoorvmittal10

Thanks for the PR @bbejeck. I have taken a pass and some comments.

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java

clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java

.../java/org/apache/kafka/streams/internals/metrics/StreamsThreadDelegatingMetricsReporter.java

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java

bbejeck · 2024-09-05T21:36:40Z

Thanks for the comments @apoorvmittal10 - I've addressed your comments.

mjsax

Not sure if I understand the end-to-end wiring already, so need to make another pass.

Btw: should we also verify state store metrics (or are they automatically verified as part of task metrics)?

clients/src/main/java/org/apache/kafka/clients/admin/Admin.java

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java

.../java/org/apache/kafka/streams/internals/metrics/StreamsThreadDelegatingMetricsReporter.java

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java

bbejeck · 2024-09-09T20:52:25Z

should we also verify state store metrics (or are they automatically verified as part of task metrics)?

They are covered as part of task metrics, but I expanded the test to explicitly verify state store metrics are correct as well

mjsax

Thanks for the PR update. Made another pass.

clients/src/main/java/org/apache/kafka/clients/admin/Admin.java

.../java/org/apache/kafka/streams/internals/metrics/StreamsClientMetricsDelegatingReporter.java

.../java/org/apache/kafka/streams/internals/metrics/StreamsThreadMetricsDelegatingReporter.java

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java

bbejeck · 2024-09-13T17:21:38Z

@mjsax thanks for the second review, I've addressed your comments

.../java/org/apache/kafka/streams/internals/metrics/StreamsClientMetricsDelegatingReporter.java

bbejeck · 2024-09-30T17:39:58Z

@mjsax, @apoorvmittal10, @AndrewJSchofield comments addressed

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java

AndrewJSchofield

Just one remaining comment.

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

bbejeck · 2024-10-01T17:46:41Z

@AndrewJSchofield, thanks for the review; I've addressed your comment.

AndrewJSchofield

lgtm

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

clients/src/main/java/org/apache/kafka/clients/admin/Admin.java

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java

streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java

… dependency, update import control, fix test

all clients in the ClientsInstanceIds in consumerInstanceIds map. This PR makes sure only the main consumer client instance id is selected.

bbejeck · 2024-11-02T16:59:36Z

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java

+            final List<String> expectedMetrics = streams.metrics().values().stream().map(Metric::metricName)
+                    .filter(metricName -> metricName.tags().containsKey("thread-id")).map(mn -> {
+                        final String name = mn.name().replace('-', '.');
+                        final String group = mn.group().replace("-metrics", "").replace('-', '.');
+                        return "org.apache.kafka." + group + "." + name;
+                    }).sorted().collect(Collectors.toList());


Needed to convert the metric names into the expected KIP-714 naming format

bbejeck · 2024-11-02T17:01:26Z

@mjsax rebased on trunk - ready for final review

mjsax · 2024-11-04T19:51:08Z

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java

+
+            final Uuid mainConsumerInstanceId = clientInstanceIds.consumerInstanceIds().entrySet().stream()
+                    .filter(entry -> !entry.getKey().endsWith("-restore-consumer")
+                            && !entry.getKey().endsWith("GlobalStreamThread"))


We don't use -global-consumer in the key for the global consumer?

This sounds like a bug to me -- checking the code, it seems we use getName() + "-consumer" inside StreamThread while we use getName() only for the global case inside KafkaStreams.

Out of scope for this PR, but I think we should fix it in a follow up PR (given that it is technically a breaking change, we should think about it twice though... But as we head into 4.0, it might be ok -- however, we should call it out in the upgrade docs and we need a Jira for it)

mjsax · 2024-11-04T20:00:13Z

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java

+                    30_000,
+                    "Never received subscribed metrics");
+            final List<String> actualInstanceMetrics = TelemetryPlugin.SUBSCRIBED_METRICS.get(adminInstanceId);
+            final List<String> expectedInstanceMetrics = Arrays.asList("org.apache.kafka.stream.alive.stream.threads", "org.apache.kafka.stream.failed.stream.threads");


Unrelated, but "alive stream thread" seems to be missing in the docs: https://kafka.apache.org/documentation/#kafka_streams_client_monitoring

Would you be interested to do a follow up PR about it?

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java

mjsax

Only some nits. Overall LGTM. Feel free to merge.

bbejeck · 2024-11-05T16:30:14Z

Merged #17021 into trunk

Implementation of KIP-1076 to allow for adding client application metrics to the KIP-714 framework Reviewers: Apoorv Mittal <[email protected]>, Andrew Schofield <[email protected]>, Matthias Sax <[email protected]>

bbejeck marked this pull request as draft August 27, 2024 22:26

bbejeck added streams kip Requires or implements a KIP KIP-1076 Metrics for client applications KIP-714 extension labels Aug 27, 2024

bbejeck force-pushed the kip_1076_implementation branch from 4e8ffa7 to d390f0a Compare August 28, 2024 22:11

bbejeck changed the title ~~[WIP NO MERGE] KAFKA-17248 - KIP 1076 implementation~~ KAFKA-17248 - KIP 1076 implementation Sep 2, 2024

bbejeck force-pushed the kip_1076_implementation branch from 61270fb to b566d34 Compare September 3, 2024 20:27

bbejeck marked this pull request as ready for review September 3, 2024 20:33

bbejeck force-pushed the kip_1076_implementation branch from d6f9919 to b96b967 Compare September 4, 2024 16:28

apoorvmittal10 self-requested a review September 4, 2024 20:23

apoorvmittal10 requested changes Sep 5, 2024

View reviewed changes

mjsax reviewed Sep 8, 2024

View reviewed changes

mjsax reviewed Sep 12, 2024

View reviewed changes

bbejeck force-pushed the kip_1076_implementation branch from 020d7c4 to 965fd23 Compare September 13, 2024 17:00

apoorvmittal10 requested changes Sep 25, 2024

View reviewed changes

.../java/org/apache/kafka/streams/internals/metrics/StreamsClientMetricsDelegatingReporter.java Outdated Show resolved Hide resolved

.../java/org/apache/kafka/streams/internals/metrics/StreamsClientMetricsDelegatingReporter.java Outdated Show resolved Hide resolved

bbejeck force-pushed the kip_1076_implementation branch from 965fd23 to 8d64d7d Compare September 25, 2024 19:38

github-actions bot added producer consumer clients labels Sep 25, 2024

bbejeck force-pushed the kip_1076_implementation branch from b3d420d to 235ccb6 Compare September 28, 2024 15:36

github-actions bot added the core Kafka Broker label Sep 28, 2024

bbejeck commented Sep 30, 2024

View reviewed changes

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java Outdated Show resolved Hide resolved

AndrewJSchofield requested changes Oct 1, 2024

View reviewed changes

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java Outdated Show resolved Hide resolved

AndrewJSchofield approved these changes Oct 2, 2024

View reviewed changes

mjsax reviewed Oct 2, 2024

View reviewed changes

bbejeck added 13 commits November 2, 2024 11:35

Get initial end-to-end integration test running

1c5884c

Respond to comments

427c65e

More descriptive name

78cef95

Reslove conflict from this commit Add open telemetry protobuf as test…

f61fb20

… dependency, update import control, fix test

Address comments, fix a potential bug and side cleanup

d5c5a7b

Store clientInstanceId before returning

3f64e7d

Address review comments

6704891

Update end-to-end test

ec48119

Update end-to-end test to validate metrics for INFO, DEBUG and TRACE

52d389f

commit f30f2d0 puts

03de445

all clients in the ClientsInstanceIds in consumerInstanceIds map. This PR makes sure only the main consumer client instance id is selected.

Standardize on lower-case if names ever change

6a40bba

Standardize on lower-case if names ever change

26ae0b5

Rebased trunk and updated test

21667fe

bbejeck force-pushed the kip_1076_implementation branch from 623cb7b to 21667fe Compare November 2, 2024 16:53

bbejeck commented Nov 2, 2024

View reviewed changes

mjsax reviewed Nov 4, 2024

View reviewed changes

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java Outdated Show resolved Hide resolved

mjsax reviewed Nov 4, 2024

View reviewed changes

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java Outdated Show resolved Hide resolved

mjsax reviewed Nov 4, 2024

View reviewed changes

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java Outdated Show resolved Hide resolved

mjsax reviewed Nov 4, 2024

View reviewed changes

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java Outdated Show resolved Hide resolved

mjsax reviewed Nov 4, 2024

View reviewed changes

...src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java Outdated Show resolved Hide resolved

mjsax approved these changes Nov 4, 2024

View reviewed changes

switch back to combined mode - getting not enough resources error

8b5682d

bbejeck merged commit 4ed0a95 into apache:trunk Nov 5, 2024
8 checks passed

bbejeck deleted the kip_1076_implementation branch November 5, 2024 16:41

KAFKA-17248 - KIP 1076 implementation #17021

KAFKA-17248 - KIP 1076 implementation #17021

Uh oh!

Conversation

bbejeck commented Aug 27, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

apoorvmittal10 left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

bbejeck commented Sep 5, 2024

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

bbejeck commented Sep 9, 2024

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

bbejeck commented Sep 13, 2024

Uh oh!

Uh oh!

Uh oh!

bbejeck commented Sep 30, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

bbejeck commented Oct 1, 2024

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

bbejeck Nov 2, 2024

Choose a reason for hiding this comment

Uh oh!

bbejeck commented Aug 27, 2024 •

edited

Loading

apoorvmittal10 left a comment •

edited

Loading

bbejeck commented Sep 30, 2024 •

edited

Loading