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.

apoorvmittal10 · 2024-09-05T09:24:57Z

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

@@ -621,6 +631,9 @@ private KafkaAdminClient(AdminClientConfig config,
            retryBackoffMaxMs,
            CommonClientConfigs.RETRY_BACKOFF_JITTER);
        this.clientTelemetryEnabled = config.getBoolean(AdminClientConfig.ENABLE_METRICS_PUSH_CONFIG);
+        List<MetricsReporter> reporters = CommonClientConfigs.metricsReporters(this.clientId, config);
+        this.clientTelemetryReporter = clientTelemetryReporter;
+        this.clientTelemetryReporter.ifPresent(reporters::add);


Should we add close of the client telemetry reporter on admin client close?

Should reporters not be closed when admin client calls metrics.close() (I would assume it does correctly close it's metrics object?)

Following up, because I don't see code changes about it yet.

Added metrics.close()to the KafkaAdmin.close() - metrics.close() does close all reporters

Hmmm, but the close will not send the last telemetry push as ClientTelemetryReporter need initiateClose method call to set state as terminating_push_needed which sends the last telemetry metrics to broker. However if we decide not to send last telemetry push in admin client that's fine but we should write it in comments somewhere.

@apoorvmittal10 good catch, I've updated the close method to call initiateClose, the use in the admin client should follow the same semantics of the other clients.

apoorvmittal10 · 2024-09-05T09:27:09Z

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

+            clientTelemetryReporter = CommonClientConfigs.telemetryReporter(clientId, config);
+            clientTelemetryReporter.ifPresent(telemetryReporter -> telemetryReporter.contextChange(metricsContext));


Though it's good to have the integartion of KIP-714 with admin client but I left it last time because of adhoc connections by Admin client, this will establish telemetry connection to send metrics, do we require them? Secondly, does admin client has any metric of it's own in Kafka? At the time of KIP-714, I didn't see any relevant metric for admin client. Last month we added https://issues.apache.org/jira/browse/KAFKA-17239 to record node latency which was related to https://issues.apache.org/jira/browse/KAFKA-17230. But again should we have KIP-714 pipeline for this 1 admin metric, I am not sure.

Hence should we register additional metrics here? Seems more of an additional overhead for adhoc requests. Else if we want external applications to register additional metrics in Admin client anyways then we might want to consider only enabling client telemetry reporter if additional metrics are registered. Wdyt?

That's a good point, I'll take a look at doing that.

do we require them?

Yes, as we want to use Admin to report KS instance metrics.

Secondly, does admin client has any metric of it's own in Kafka

I would assume so? Why would we create a new Metrics object if we don't use it? (Even if I have to admit, the docs don't say anything about admin metrics... 🤔 -- maybe @cmccabe can help on this question?)

we might want to consider only enabling client telemetry reporter if additional metrics are registered. Wdyt?

I guess we could do this -- but it's kinda odd to have the Admin config "enable push metrics" but not enable it if set to true...? -- If we don't want to enable it on the admin client, we should rather remove (or for now only deprecate) the existing config, and add a new one via KIP-1076 which would enable "application push metrics" which would only create a reporter if application metrics are registered -- otherwise, we end up with some weird behavior which can easily confuse users.

Btw: I was just looking into the consumer/producer code, and it seem there we just create the reporter and add to reporters list before new Metrics is called -- this way, metricsContext will be "added" automatically to the reporter. Should simplify the code a little bit?

I'm leaning toward not doing this, as I agree with @mjsax. Plus, there's additional complexity in the case where metrics are unregistered. Should that trigger the mirror action of disabling metrics pushing?

Btw: I was just looking into the consumer/producer code, and it seem there we just create the reporter and add to reporters list before new Metrics is called -- this way, metricsContext will be "added" automatically to the reporter. Should simplify the code a little bit?

I'm not sure I follow this comment. Are you referring to the reporters we create in Kafka Streams to act as proxies for passing the metrics to the clients and suggesting we create them in the consumer/producer clients themselves?
Those Reporter instances are specific to Kafka Streams only. Also, the Metrics API provides an addReporter method, so it seems it was designed to add a reporter after instantiating the Metrics object.

So the only metric I think admin client emits is node latency metric, which also got recently added as regression. Though I understand that KS wants to send application metrics as part of admin client but we will be adding an additional overhead with short lived admin client operations. I am not very aligned regarding with KIP-714 integration in admin client. @AndrewJSchofield wdyt?

@apoorvmittal10 I spoke offline with @mjsax and what we've come up with is that we'll disable metrics in the AdminClient by default (I'll update KIP-1076 with this detail). KafkaStreams will explicitly enable the metrics push for the AdminClient, avoiding any additional overhead for non-KafkaStreams usage. WDYT?

That sounds fair to me. For my knowledge, does admin client in streams application has a long running connection where metrics will be emitted? My concern is with if connections are adhoc and we do create too many admin client short lived connections then it might impact the connection cache i.e. connection cache keeps the client instance for maximum of push interval * 3 ms hence if too many admin client connections created in short span then it might hurt the performance.

Sounds good to me too.

Kafka Streams uses the admin client to also send "delete record" request for repartition topics. For this case, the connection should be long lived, as we send a "delete record" request after each commit by default (it's actually configurable).

For the (more rare) case that there is no repartition topic, it could be a slightly different story, but IIRC, @bbejeck did some tests already, verifying that the connection is not closed by long lived, even for this case. So I think we are good?

We could run more test before the 4.0 release to double check, and if we really find an issue, we could still update the KIP, to let KS use different connection timeout default, to make sure that the connection will be long lived.

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.

mjsax · 2024-09-12T00:02:22Z

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

+     * - `Sum`: Monotonic total count meter (Counter). Suitable for metrics like total number of X, e.g., total bytes sent.
+     * - `Gauge`: Non-monotonic current value meter (UpDownCounter). Suitable for metrics like current value of Y, e.g., current queue count.
+     * Metrics not matching these types are silently ignored.
+     * Executing this method for a previously registered metric is a benign operation and results in updating that metrics entry.


benign

I just learned a new word :)

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

mjsax · 2024-09-12T00:41:16Z

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

+
+    @Test
+    @DisplayName("Calling unregisterMetric on metrics not registered should not cause an error")
+    public void shouldNotThrowExceptionWhenRemovingNonExistingMetrics() throws InterruptedException {


Do we need to test this with a heavy-weight integration test? Or could we unit-test this in a more light weight fashion?

I don't think so. I used this test with a "live" consumer because the call to remove metrics goes into the ClientTelemetryReporter, and I want to confirm no exceptions at this level. Mocking would miss this interaction.

mjsax · 2024-09-12T00:42:22Z

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

+
+    @ParameterizedTest
+    @MethodSource("singleAndMultiTaskParameters")
+    @DisplayName("Streams metrics should get passed to Consumer")


Similar to above? Should we add this to StreamThreadTest as unit-test?

Possibly, but the KafkaStreams constructor creates the admin client and the Reporter instance to pass streams client metrics at this point as well. Given the integration test is needed anyway to verify metrics track with task assignment changes due to a rebalance, it isn't a burden to have this in an integration test.

mjsax · 2024-09-12T00:44:33Z

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

+    }
+
+    @Test
+    @DisplayName("Streams metrics should not be visible in consumer metrics")


Sounds unit-testable via StreamThreadTest ?

I'll leave it as is for the same reason as above. This test asserts that passing stream metrics to the consumer, which in turn adds them the ClientTelemetryReporter does not bleed over to the JMX reporter.

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

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

apoorvmittal10 · 2024-09-25T15:36:42Z

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

+    }
+
+    private boolean isStreamsClientMetric(final KafkaMetric metric) {
+        final boolean shouldInclude = metric.metricName().group().equals("stream-metrics");


What about stream-thread-metrics, stream-task-metrics, stream-processor-node-metrics, stream-state-metrics and stream-record-cache-metrics? Are they not needed or they fall in same stream-metrics group?

@apoorvmittal10, those groups are not part of client-level grouping and get reported separately

apoorvmittal10 · 2024-09-25T15:37:05Z

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

+import java.util.Map;
+import java.util.Objects;
+
+public class StreamsClientMetricsDelegatingReporter implements MetricsReporter {


And are we going to expect this reporter being registered additionally in metrics.reporter server config to send Streams metrics?

Ahhh, I see it's added as default metrics reporter on metrics instance.

Correct metric filtering to only include metrics with thread-id Remove unused method

…cs infra totally for adding removing metrics.

…ove setting consumer to null in reporter close

…rics, change visibility on test methods to public

Added a new class `StreamsThreadMetricsDelegatingReporter` to handle metrics reporting for Kafka Streams threads. Updated method names across multiple files to improve clarity and consistency by changing `registerMetric` to `registerMetricForSubscription` and `unregisterMetric` to `unregisterMetricFromSubscription`. Additionally, included new tests to verify the changes.

Refactored `KafkaStreamsTelemetryIntegrationTest` to use an added a custom telemetry reporter `TestingClientTelemetry` to integrate telemetry with the test environment. Updated checkstyle to permit new telemetry-related imports.

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

@mjsax, @apoorvmittal10, @AndrewJSchofield comments addressed

bbejeck · 2024-09-30T20:37:50Z

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

+                    .map(Map.Entry::getValue)
+                    .findFirst().get();
+            assertNotNull(adminInstanceId);
+            assertNotNull(mainConsumerInstanceId);


I have more to do with this test, but having it complete should not block the PR progress.

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 suggested 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 suggested changes Sep 25, 2024

View reviewed changes

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 added 8 commits September 27, 2024 18:17

Initial checkin

d2ff6c0

It compiles!

8e618f3

Added producer and basic impl in mock clients.

669cfba

Fix producer javadoc comment

fd1c283

Correct metric filtering to only include metrics with thread-id Remove unused method

Fix spotbugs and checkstyle errors

60a958e

Simplify filtering for adding metrics

cd573ee

Clean up the creation of the reporter such that it relys on the metri…

c925e91

…cs infra totally for adding removing metrics.

Added parameterized integration test

3c53d18

bbejeck added 24 commits September 27, 2024 18:17

Added test for passing streams metrics

36ead77

Added integration tests, set logging level in reporter to debug

7ff542f

Added test to confirm stream metrics don't bleed into consumer metrics

20497e2

Added unit test for StreamsThreadDelegatingReporter, rebased trunk

f8ddb69

Add some description to large test method

d2d2e2c

checkstyle fixes

007dc48

Remove unintentional formatting

11efba9

Changes per comments

5e59e74

Close telemetry reporter on admin client close.

94e5fb1

Cleanup, revert closing telemetry reporter pending investigation, rem…

5077f45

…ove setting consumer to null in reporter close

Checkstyle errors

9d394c2

Address review comments: More descriptive javadoc

032d07e

Address review comments: indentation, consumer generic types

77d1e29

Address review comments: update test to account for correct state met…

75e8ce3

…rics, change visibility on test methods to public

Adding Admin delegate for testing sending metrics via admin client

8483dcd

Add private access modifiers to fields

275191e

Refactor and add telemetry support in integration test

f824f8b

Refactored `KafkaStreamsTelemetryIntegrationTest` to use an added a custom telemetry reporter `TestingClientTelemetry` to integrate telemetry with the test environment. Updated checkstyle to permit new telemetry-related imports.

Updates per review comments

56be12e

Add the telemetry reporter to the reporters list after creation

8e10fc5

Fix broken tests in KafkaAdminClientTest

8faac9f

Minor cleanup

17d12db

Disable metrics push in AdminClient by default; update tests

6d8db41

fix tests, start of end-to-end metrics test

235ccb6

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

Call initiate close on ClientTelemetryReporter during admin close

c018afe

bbejeck commented Sep 30, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-17248 - KIP 1076 implementation #17021

KAFKA-17248 - KIP 1076 implementation #17021

bbejeck commented Aug 27, 2024 •

edited

Loading

apoorvmittal10 left a comment •

edited

Loading

apoorvmittal10 Sep 5, 2024

mjsax Sep 12, 2024

bbejeck Sep 13, 2024

apoorvmittal10 Sep 25, 2024

bbejeck Sep 30, 2024 •

edited

Loading

apoorvmittal10 Sep 5, 2024

bbejeck Sep 5, 2024

mjsax Sep 12, 2024

bbejeck Sep 13, 2024

apoorvmittal10 Sep 25, 2024

bbejeck Sep 27, 2024

apoorvmittal10 Sep 27, 2024 •

edited

Loading

AndrewJSchofield Sep 27, 2024

mjsax Sep 27, 2024

bbejeck commented Sep 5, 2024

mjsax left a comment

bbejeck commented Sep 9, 2024

mjsax left a comment

mjsax Sep 12, 2024

apoorvmittal10 Sep 25, 2024

mjsax Sep 12, 2024

bbejeck Sep 13, 2024

mjsax Sep 12, 2024

bbejeck Sep 13, 2024

mjsax Sep 12, 2024

bbejeck Sep 13, 2024 •

edited

Loading

bbejeck commented Sep 13, 2024

apoorvmittal10 Sep 25, 2024

bbejeck Sep 30, 2024

apoorvmittal10 Sep 25, 2024

apoorvmittal10 Sep 25, 2024

bbejeck commented Sep 30, 2024 •

edited

Loading

bbejeck Sep 30, 2024

		clientTelemetryReporter = CommonClientConfigs.telemetryReporter(clientId, config);
		clientTelemetryReporter.ifPresent(telemetryReporter -> telemetryReporter.contextChange(metricsContext));

KAFKA-17248 - KIP 1076 implementation #17021

Are you sure you want to change the base?

KAFKA-17248 - KIP 1076 implementation #17021

Conversation

bbejeck commented Aug 27, 2024 • edited Loading

Committer Checklist (excluded from commit message)

apoorvmittal10 left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck Sep 30, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apoorvmittal10 Sep 27, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck commented Sep 5, 2024

mjsax left a comment

Choose a reason for hiding this comment

bbejeck commented Sep 9, 2024

mjsax left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck Sep 13, 2024 • edited Loading

Choose a reason for hiding this comment

bbejeck commented Sep 13, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck commented Sep 30, 2024 • edited Loading

Choose a reason for hiding this comment

bbejeck commented Aug 27, 2024 •

edited

Loading

apoorvmittal10 left a comment •

edited

Loading

bbejeck Sep 30, 2024 •

edited

Loading

apoorvmittal10 Sep 27, 2024 •

edited

Loading

bbejeck Sep 13, 2024 •

edited

Loading

bbejeck commented Sep 30, 2024 •

edited

Loading