KAFKA-15778 & KAFKA-15779: Implement metrics manager (KIP-714) by apoorvmittal10 · Pull Request #14699 · apache/kafka

apoorvmittal10 · 2023-11-03T18:43:17Z

The PR provide implementation for client metrics manager along with other classes. Manager is responsible to support 3 operations:

UpdateSubscription - From kafka-configs.sh and reload from metadata cache.
Process Get Telemetry Request - From KafkaApis.scala
Process Push Telemetry Request - From KafkaApis.scala

Manager maintains an in-memory cache to keep track of client instances against their instance id.

Committer Checklist (excluded from commit message)

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

…ance state (KIP-714)

apoorvmittal10 · 2023-11-03T18:50:59Z

+        return INSTANCE;
+    }
+    // Max cache size (16k active client connections per broker)
+    private static final int CM_CACHE_MAX_SIZE = 16384;


The cache should hold the information of all connected client hence it should support the number of clients that can connect to single broker. Kept it high enough for now but would require suggestions for best approach. @AndrewJSchofield mentioned that it might be the max.connections broker config but the configs uppoer bound is Int.MAX_VALUE hence we need to think about better upper bound here.

Is this enough? A broker could handle 100s of Ks of connections.

Then should it be in low 100K i.e. 2^17 = 131072? I am not sure what the right config should be here, I know the cloud providers like MSK typically provides 3000 active connections per broker where this limit is higher in case of Confluent but what typically are the active number of client connecting to broker should be.

The challenge is that the connections can vary depending on the type of broker host. Larger instance typically can accommodate more connections. Do we need to make this an LRU cache? If a client is terminated or idle, we should remove them from the cache. Otherwise, we probably should just rely on the existing max.connections to control the client connections?

Do we need to make this an LRU cache? If a client is terminated or idle, we should remove them from the cache

I have started with LRU cache and planning to improve this with cache which timebounds the connection. The KIP says: client instance specific state is maintained in broker memory up to MAX(60*1000, PushIntervalMs * 3) milliseconds. I ll add improvement on the cache to respect that: https://issues.apache.org/jira/browse/KAFKA-15813

apoorvmittal10 · 2023-11-03T18:54:46Z

+    }
+    // Max cache size (16k active client connections per broker)
+    private static final int CM_CACHE_MAX_SIZE = 16384;
+    private final Cache<Uuid, ClientMetricsInstance> clientInstanceCache;


The KIP-714 says: This client instance specific state is maintained in broker memory up to MAX(60*1000, PushIntervalMs * 3) milliseconds and is used to enforce the push interval rate-limiting. There is no persistence of client instance metrics state across broker restarts or between brokers.

I have started with LRUCache with oldest entry eviction but will implement something similar to Selector.IdleExpiryManager which cleansup old connections.

apoorvmittal10 · 2023-11-07T20:54:09Z

@junrao @hachikuji @AndrewJSchofield @mjsax Please if I can get feedback on the PR.

junrao

@apoorvmittal10 : Thanks for the PR. Made a pass of non-testing files. Left a few comments.

junrao · 2023-11-08T22:35:57Z

 */
 package kafka.server;

+import java.util.Collections;


Should this class be in the same package as other client metric related classes like DefaultClientTelemetryPayload?

The reason I kept this at kafka.server package as I see all managers (in scala) processing API calls from KafkaApis.scala resides kafka.server package.

junrao · 2023-11-08T23:52:35Z

    @Override
    public void close() throws IOException {
-        // TODO: Implement the close logic to close the client metrics manager.
+        // Do nothing for now.


Should we clear up the internal data?

I have cleaned up the subscription map, there is no method currently exposed to clear cache but I ll add that as improvement while improving the cache code.

junrao · 2023-11-09T19:55:06Z

+        // broker's point of view. But the broker does not have this information rather the port could be
+        // the broker's port where the client connection is established. We might want to consider removing
+        // the client source port from the KIP or use broker port if that can be helpful.
+        // TODO: fix port


We populate RequestContext in SocketServer. We could obtain the client port through transportLayer.socketChannel().socket().getPort in KafkaChannel.

Thanks a lot @junrao this is helpful. I will make the changes in subsequent PR to address this. I have created following jira for same: https://issues.apache.org/jira/browse/KAFKA-15811

junrao · 2023-11-09T21:13:47Z

+    }
+
+    private static ByteBuffer decompressMetricsData(CompressionType compressionType, byte[] metrics) {
+        // TODO: Add support for decompression of metrics data


So, this will be added in a future PR?

Yes, I have created jira in parent KIP-714 task to address this: https://issues.apache.org/jira/browse/KAFKA-15807. I am planning to get end-to-end metrics flow without compression first.

apoorvmittal10 · 2023-11-13T20:23:54Z

@apoorvmittal10 : Thanks for the PR. Made a pass of non-testing files. Left a few comments.

Thanks a lot for the review @junrao . I have addressed the comments and have a question related to throttleTimeMs for errors in the comments. Please if you can re-review.

junrao

@apoorvmittal10 : Thanks for the updated PR. Made a pass of all files. A few more comments.

junrao · 2023-11-14T00:00:25Z

+    public PushTelemetryResponse createResponse(int throttleTimeMs, Errors errors) {
+        PushTelemetryResponseData responseData = new PushTelemetryResponseData();
+        responseData.setErrorCode(errors.code());
+        responseData.setThrottleTimeMs(throttleTimeMs);


The following is my understanding. There are two types of throttling.

Generic request throttling based on % of CPU a client uses on the broker. This applies to any request, including PushTelemetry. In this case, the error code is none and the throttleTimeMs field is set. The client will mute the channel for throttleTimeMs before sending future requests.

PushTelemetry specific throttling because PushTelemetry is sent too frequently. In this case, we should set THROTTLING_QUOTA_EXCEEDED as the error code and avoid setting the throttleTimeMs field since we don't want to the client to mute the channel for all requests.

junrao · 2023-11-14T22:05:40Z

+    }
+
+    public boolean terminating() {
+        return terminating;


After the instance is terminated, when do we remove the instance form the in-memory state?

The reason why I didn't remove the instance from in-memory cache as the subsequent requests from terminated client should be rejected. Having said that, the terminated client should also not remain in-memory forever hence the terminated client should be removed from cache as per the eviction policy of cache i.e. MAX(60*1000, PushIntervalMs * 3) milliseconds. This shall be inherently handled by the cache improvement task where we have some time based eviction policy.

apoorvmittal10 · 2023-11-16T18:33:36Z

Thanks for reviewing @junrao, I have updated the PR.

junrao · 2023-11-17T18:19:10Z

+    @Test
+    public void testMaybeUpdateRequestEpochValid() {
+        // First request should be accepted.
+        assertTrue(clientInstance.maybeUpdateGetRequestEpoch(System.currentTimeMillis()));


Could we use MockTime instead of System time?

As the class/methods I am testing do not have Time reference rather just tries to updates the epoch supplied hence I didn't see any value in using MockTime. However I did change the code and tests for ClientMetricsManager where these methods are invoked from, I started using Time in ClientMetricsManager and corresponding MockTime in ClientMetricsManagerTest which eliminated the use of Thread.sleep.

Please let me know if I am missing anything here.

apoorvmittal10 · 2023-11-17T20:52:31Z

Thanks @junrao for leaving the comments, I have tried to address them.

junrao

@apoorvmittal10 : Thanks for the updated PR. Left a few more comments.

apoorvmittal10 · 2023-11-25T18:50:22Z

@junrao @AndrewJSchofield Thanks for the LGTM and approving the PR. Below is the test run status, none of the failing tests are related to the changes in the PR but I have tried to debug further. Among 20, 17 have already been reported as flaky test in jira, for remaining 3 I tried to locally reproduce but couldn't hence created flaky test for same. Can we merge the PR?

New failing - 20

Already existing jira for flaky test: Build / JDK 11 and Scala 2.13 / testResetSinkConnectorOffsetsZombieSinkTasks – org.apache.kafka.connect.integration.OffsetsApiIntegrationTest
1m 2s: https://issues.apache.org/jira/browse/KAFKA-15524

Already existing jira for flaky test: Build / JDK 11 and Scala 2.13 / testRackAwareRangeAssignor(String).quorum=kraft – integration.kafka.server.FetchFromFollowerIntegrationTest
8s: https://issues.apache.org/jira/browse/KAFKA-15020

Already existing jira for flaky test: Build / JDK 11 and Scala 2.13 / testRackAwareRangeAssignor(String).quorum=kraft – integration.kafka.server.FetchFromFollowerIntegrationTest
6s: https://issues.apache.org/jira/browse/KAFKA-15020

Already existing jira for flaky test: Build / JDK 11 and Scala 2.13 / testTaskRequestWithOldStartMsGetsUpdated() – org.apache.kafka.trogdor.coordinator.CoordinatorTest
2m 0s: https://issues.apache.org/jira/browse/KAFKA-15760

Already existing jira for flaky test: Build / JDK 21 and Scala 2.13 / testReplicateSourceDefault() – org.apache.kafka.connect.mirror.integration.IdentityReplicationIntegrationTest
2m 0s: https://issues.apache.org/jira/browse/KAFKA-15292

Already existing jira for flaky test: Build / JDK 21 and Scala 2.13 / testConsumptionWithBrokerFailures() – kafka.api.ConsumerBounceTest
31s: https://issues.apache.org/jira/browse/KAFKA-15146

Already existing jira for flaky test: Build / JDK 21 and Scala 2.13 / testDescribeClusterRequestIncludingClusterAuthorizedOperations(String).quorum=kraft – kafka.server.DescribeClusterRequestTest
4s: https://issues.apache.org/jira/browse/KAFKA-15759

Already existing jira for flaky test: Build / JDK 21 and Scala 2.13 / testSendOffsetsWithGroupId(String).quorum=zk – org.apache.kafka.tiered.storage.integration.TransactionsWithTieredStoreTest
1m 15s: https://issues.apache.org/jira/browse/KAFKA-15772

Created Jira, cannot reproduce locally (also exists other flaky tests from same class, have already ben reported by others): Build / JDK 21 and Scala 2.13 / shouldQuerySpecificStalePartitionStores() – org.apache.kafka.streams.integration.StoreQueryIntegrationTest
4s: Another test in same class flaky: https://issues.apache.org/jira/browse/KAFKA-9897, created jira: https://issues.apache.org/jira/browse/KAFKA-15896

Already existing jira for flaky test: Build / JDK 17 and Scala 2.13 / testReplicateSourceDefault() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationBaseTest
2m 23s: https://issues.apache.org/jira/browse/KAFKA-15699

Already existing jira for flaky test: Build / JDK 17 and Scala 2.13 / testMultiWorkerRestartOnlyConnector – org.apache.kafka.connect.integration.ConnectorRestartApiIntegrationTest
2m 33s: https://issues.apache.org/jira/browse/KAFKA-15675

Already existing jira for flaky test: Build / JDK 17 and Scala 2.13 / testDescribeTokenForOtherUserPasses(String).quorum=kraft – kafka.api.DelegationTokenEndToEndAuthorizationWithOwnerTest
9s: https://issues.apache.org/jira/browse/KAFKA-15411

Created flaky test jira, cannot reproduce locally: Build / JDK 17 and Scala 2.13 / testWrongIncarnationId() – kafka.server.ControllerRegistrationManagerTest
30s: https://issues.apache.org/jira/browse/KAFKA-15897

Created flaky test jira, cannot reproduce locally: Build / JDK 17 and Scala 2.13 / testFenceMultipleBrokers() – org.apache.kafka.controller.QuorumControllerTest
43s: https://issues.apache.org/jira/browse/KAFKA-15898

Already existing jira for flaky test: Build / JDK 8 and Scala 2.12 / testSyncTopicConfigs() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationBaseTest
2m 33s: https://issues.apache.org/jira/browse/KAFKA-14971

Already existing jira for flaky test: Build / JDK 8 and Scala 2.12 / [2] quorum=kraft, isIdempotenceEnabled=false – kafka.api.DelegationTokenEndToEndAuthorizationWithOwnerTest
10s: https://issues.apache.org/jira/browse/KAFKA-15411

Already existing jira for flaky test: Build / JDK 8 and Scala 2.12 / testDescribeClusterRequestExcludingClusterAuthorizedOperations(String).quorum=kraft – kafka.server.DescribeClusterRequestTest
4s: https://issues.apache.org/jira/browse/KAFKA-15419

Already existing jira for flaky test: Build / JDK 8 and Scala 2.12 / shouldWriteLatestOffsetsToCheckpointOnShutdown[at_least_once, processing threads = true] – org.apache.kafka.streams.integration.EosIntegrationTest
1m 25s: https://issues.apache.org/jira/browse/KAFKA-15690

Already existing jira for flaky test: Build / JDK 8 and Scala 2.12 / shouldWriteLatestOffsetsToCheckpointOnShutdown[exactly_once, processing threads = true] – org.apache.kafka.streams.integration.EosIntegrationTest
1m 21s: https://issues.apache.org/jira/browse/KAFKA-15690

Already existing jira for flaky test: Build / JDK 8 and Scala 2.12 / shouldWriteLatestOffsetsToCheckpointOnShutdown[exactly_once_v2, processing threads = true] – org.apache.kafka.streams.integration.EosIntegrationTest: https://issues.apache.org/jira/browse/KAFKA-15690

junrao · 2023-11-27T17:47:08Z

@apoorvmittal10 : Thanks for triaging the tests. In the mailing list, it seems that we are still leaning towards requiring green builds before merging a PR.

junrao · 2023-11-28T17:30:44Z

Just merged #14632. Triggering another test run to make sure there are no new issues.

apoorvmittal10 · 2023-11-28T18:32:21Z

Just merged #14632. Triggering another test run to make sure there are no new issues.

Thanks @junrao.

junrao · 2023-11-28T21:52:39Z

@apoorvmittal10 : It seems that the build for JDK 17 and Scala 2.13 didn't complete.

apoorvmittal10 · 2023-11-29T02:03:45Z

@apoorvmittal10 : It seems that the build for JDK 17 and Scala 2.13 didn't complete.

Thanks @junrao. Strange to see the failure in. :clients:test. I can see below unrelated failure in logs, retriggered build.

> Task :clients:test
org.apache.kafka.common.network.SelectorTest.testConnectionsByClientMetric() failed, log available in /home/jenkins/jenkins-agent/712657a4/workspace/Kafka_kafka-pr_PR-14699/clients/build/reports/testOutput/org.apache.kafka.common.network.SelectorTest.testConnectionsByClientMetric().test.stdout

Gradle Test Run :clients:test > Gradle Test Executor 13 > SelectorTest > testConnectionsByClientMetric() FAILED
    java.util.concurrent.TimeoutException: testConnectionsByClientMetric() timed out after 240 seconds
        at org.junit.jupiter.engine.extension.TimeoutExceptionFactory.create(TimeoutExceptionFactory.java:29)
        at org.junit.jupiter.engine.extension.SameThreadTimeoutInvocation.proceed(SameThreadTimeoutInvocation.java:58)
        at org.junit.jupiter.engine.extension.TimeoutExtension.intercept(TimeoutExtension.java:156)
        at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestableMethod(TimeoutExtension.java:147)
        at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestMethod(TimeoutExtension.java:86)
        at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker$ReflectiveInterceptorCall.lambda$ofVoidMethod$0(InterceptingExecutableInvoker.java:103)
        at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker.lambda$invoke$0(InterceptingExecutableInvoker.java:93)
        at org.junit.jupiter.engine.execution.InvocationInterceptorChain$InterceptedInvocation.proceed(InvocationInterceptorChain.java:106)
        at org.junit.jupiter.engine.execution.InvocationInterceptorChain.proceed(InvocationInterceptorChain.java:64)
        at org.junit.jupiter.engine.execution.InvocationInterceptorChain.chainAndInvoke(InvocationInterceptorChain.java:45)
        at org.junit.jupiter.engine.execution.InvocationInterceptorChain.invoke(InvocationInterceptorChain.java:37)
        at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker.invoke(InterceptingExecutableInvoker.java:92)
        at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker.invoke(InterceptingExecutableInvoker.java:86)
        at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.lambda$invokeTestMethod$7(TestMethodTestDescriptor.java:218)
        at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
        at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.invokeTestMethod(TestMethodTestDescriptor.java:214)

apoorvmittal10 · 2023-11-29T15:24:00Z

Hi @junrao, I have triaged the test cases as per the latest run.

Existing: Build / JDK 17 and Scala 2.13 / testMultiWorkerRestartOnlyConnector – org.apache.kafka.connect.integration.ConnectorRestartApiIntegrationTest
2m 21s: https://issues.apache.org/jira/browse/KAFKA-15675

Existing: Build / JDK 17 and Scala 2.13 / testRackAwareRangeAssignor(String).quorum=zk – integration.kafka.server.FetchFromFollowerIntegrationTest
39s: https://issues.apache.org/jira/browse/KAFKA-15020

Existing: Build / JDK 17 and Scala 2.13 / testCoordinatorFailover(String, String).quorum=zk.groupProtocol=generic – kafka.api.SslConsumerTest
8s: https://issues.apache.org/jira/browse/KAFKA-15920

Existing: Build / JDK 17 and Scala 2.13 / testDescribeClusterRequestExcludingClusterAuthorizedOperations(String).quorum=kraft – kafka.server.DescribeClusterRequestTest
3s: https://issues.apache.org/jira/browse/KAFKA-15419

Existing: Build / JDK 17 and Scala 2.13 / testFenceMultipleBrokers() – org.apache.kafka.controller.QuorumControllerTest
51s: https://issues.apache.org/jira/browse/KAFKA-15898

Existing: Build / JDK 21 and Scala 2.13 / testSyncTopicConfigs() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationBaseTest
1m 42s: https://issues.apache.org/jira/browse/KAFKA-15523

Existing: Build / JDK 21 and Scala 2.13 / testTaskRequestWithOldStartMsGetsUpdated() – org.apache.kafka.trogdor.coordinator.CoordinatorTest
2m 0s: https://issues.apache.org/jira/browse/KAFKA-15760

Existing: Build / JDK 11 and Scala 2.13 / testSyncTopicConfigs() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationTransactionsTest
3m 44s: https://issues.apache.org/jira/browse/KAFKA-15523

Triaged: Build / JDK 11 and Scala 2.13 / testRestartReplication() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationTransactionsTest
3m 21s: https://issues.apache.org/jira/browse/KAFKA-15933

Triaged: Build / JDK 11 and Scala 2.13 / testMultiConsumerStickyAssignor(String, String).quorum=kraft+kip848.groupProtocol=generic – kafka.api.PlaintextConsumerTest
2m 12s: https://issues.apache.org/jira/browse/KAFKA-15934

Existing: Build / JDK 8 and Scala 2.12 / testSyncTopicConfigs() – org.apache.kafka.connect.mirror.integration.IdentityReplicationIntegrationTest
4m 25s: https://issues.apache.org/jira/browse/KAFKA-14971

Triaged: Build / JDK 8 and Scala 2.12 / testRestartReplication() – org.apache.kafka.connect.mirror.integration.IdentityReplicationIntegrationTest
2m 8s: https://issues.apache.org/jira/browse/KAFKA-15935

Existing: Build / JDK 8 and Scala 2.12 / testNoCheckpointsIfNoRecordsAreMirrored() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationBaseTest
2m 59s: https://issues.apache.org/jira/browse/KAFKA-15699

Existing: Build / JDK 8 and Scala 2.12 / testSyncTopicConfigs() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationTransactionsTest
3m 37s: https://issues.apache.org/jira/browse/KAFKA-15523

Existing: Build / JDK 8 and Scala 2.12 / testMultiWorkerRestartOnlyConnector – org.apache.kafka.connect.integration.ConnectorRestartApiIntegrationTest
2m 28s: https://issues.apache.org/jira/browse/KAFKA-15761

Existing: Build / JDK 8 and Scala 2.12 / testMultiWorkerRestartOnlyConnector – org.apache.kafka.connect.integration.ConnectorRestartApiIntegrationTest
2m 39s: https://issues.apache.org/jira/browse/KAFKA-15761

Triaged: Build / JDK 8 and Scala 2.12 / testGetActiveTopics – org.apache.kafka.connect.integration.ConnectorTopicsIntegrationTest
1m 50s: https://issues.apache.org/jira/browse/KAFKA-15936

Triaged: Build / JDK 8 and Scala 2.12 / testTopicTrackingResetIsDisabled – org.apache.kafka.connect.integration.ConnectorTopicsIntegrationTest
2m 37s: https://issues.apache.org/jira/browse/KAFKA-15936

Triaged: Build / JDK 8 and Scala 2.12 / testTopicTrackingIsDisabled – org.apache.kafka.connect.integration.ConnectorTopicsIntegrationTest
3m 4s: https://issues.apache.org/jira/browse/KAFKA-15936

Existing: Build / JDK 8 and Scala 2.12 / testConnectorReconfiguration – org.apache.kafka.connect.integration.ExactlyOnceSourceIntegrationTest
1m 42s: https://issues.apache.org/jira/browse/KAFKA-14901

Existing: Build / JDK 8 and Scala 2.12 / testSeparateOffsetsTopic – org.apache.kafka.connect.integration.ExactlyOnceSourceIntegrationTest
3m 35s: https://issues.apache.org/jira/browse/KAFKA-14089

Triaged: Build / JDK 8 and Scala 2.12 / testTopicDeletion(String).quorum=kraft – kafka.admin.RemoteTopicCrudTest
1m 30s: https://issues.apache.org/jira/browse/KAFKA-15937

Triaged: Build / JDK 8 and Scala 2.12 / testCreateRemoteTopicWithValidRetentionTime(String).quorum=kraft – kafka.admin.RemoteTopicCrudTest
2m 17s: https://issues.apache.org/jira/browse/KAFKA-15938

Triaged: Build / JDK 8 and Scala 2.12 / testInvalidAlterConfigs(String).quorum=kraft – kafka.api.AdminClientWithPoliciesIntegrationTest
2m 2s: https://issues.apache.org/jira/browse/KAFKA-15939

Existing: Build / JDK 8 and Scala 2.12 / testDescribeClusterRequestExcludingClusterAuthorizedOperations(String).quorum=kraft – kafka.server.DescribeClusterRequestTest
3s: https://issues.apache.org/jira/browse/KAFKA-15419

Triaged: Build / JDK 8 and Scala 2.12 / testLogCleanerConfig(String).quorum=kraft – kafka.server.DynamicBrokerReconfigurationTest
41s: https://issues.apache.org/jira/browse/KAFKA-15940

Existing: Build / JDK 8 and Scala 2.12 / shouldHaveSamePositionBoundActiveAndStandBy – org.apache.kafka.streams.integration.ConsistencyVectorIntegrationTest
22s: https://issues.apache.org/jira/browse/KAFKA-15770

Existing: Build / JDK 8 and Scala 2.12 / shouldWriteLatestOffsetsToCheckpointOnShutdown[exactly_once_v2, processing threads = true] – org.apache.kafka.streams.integration.EosIntegrationTest
2m 11s: https://issues.apache.org/jira/browse/KAFKA-15690

Existing: Build / JDK 8 and Scala 2.12 / shouldUpgradeFromEosAlphaToEosV2[false] – org.apache.kafka.streams.integration.EosV2UpgradeIntegrationTest
1m 47s: https://issues.apache.org/jira/browse/KAFKA-15095

Existing: Build / JDK 8 and Scala 2.12 / shouldUpgradeFromEosAlphaToEosV2[true] – org.apache.kafka.streams.integration.EosV2UpgradeIntegrationTest
4m 31s: https://issues.apache.org/jira/browse/KAFKA-15095

Triaged: Build / JDK 8 and Scala 2.12 / shouldRestoreNullRecord() – org.apache.kafka.streams.integration.RestoreIntegrationTest
1m 5s: https://issues.apache.org/jira/browse/KAFKA-15941

Existing: Build / JDK 8 and Scala 2.12 / testTopicDeletion(String).quorum=kraft – org.apache.kafka.tools.TopicCommandIntegrationTest
4m 11s: https://issues.apache.org/jira/browse/KAFKA-15140

Existing: Build / JDK 8 and Scala 2.12 / testDescribeReportOverriddenConfigs(String).quorum=kraft – org.apache.kafka.tools.TopicCommandIntegrationTest: https://issues.apache.org/jira/browse/KAFKA-15140

apoorvmittal10 · 2023-11-29T17:02:48Z

@junrao The tests failures are not related to the changes.

junrao

Thanks for the analysis, @apoorvmittal10. The PR LGTM.

…e#14699) The PR provide implementation for client metrics manager along with other classes. Manager is responsible to support 3 operations: UpdateSubscription - From kafka-configs.sh and reload from metadata cache. Process Get Telemetry Request - From KafkaApis.scala Process Push Telemetry Request - From KafkaApis.scala Manager maintains an in-memory cache to keep track of client instances against their instance id. Reviewers: Andrew Schofield <aschofield@confluent.io>, Jun Rao <junrao@gmail.com>

chia7712 · 2026-08-06T06:09:16Z

+            return request.getErrorResponse(0, exception);
+        } finally {
+            // Update the client instance with the latest push request parameters.
+            clientInstance.terminating(request.data().terminating());


We noticed an edge case and have opened #23094 to discuss it. It would be great to get your review :)

KAFKA-15778 & KAFKA-15779: Implement metrics manager with client inst…

d2499d0

…ance state (KIP-714)

apoorvmittal10 commented Nov 3, 2023

View reviewed changes

Comment thread core/src/main/java/kafka/metrics/ClientMetricsConfigs.java

apoorvmittal10 commented Nov 3, 2023

View reviewed changes

apoorvmittal10 added 2 commits November 3, 2023 22:12

Adding kafka config with telemerty config

26db98e

Merge remote-tracking branch 'upstream/trunk' into kip-714-ak-manager

31a1e23

apoorvmittal10 commented Nov 3, 2023

View reviewed changes

Comment thread core/src/main/java/kafka/metrics/ClientMetricsConfigs.java

apoorvmittal10 added 2 commits November 7, 2023 16:16

Updated request time tracking

fb4251f

Update subscribed metrics config to include all

d4f70e1

Adding compression type error code for retry

84038aa

junrao reviewed Nov 9, 2023

View reviewed changes

Addressing review comments

8f11a0e

apoorvmittal10 requested a review from junrao November 13, 2023 20:22

junrao reviewed Nov 14, 2023

View reviewed changes

apoorvmittal10 mentioned this pull request Nov 15, 2023

KAFKA-15830: Add telemetry API handling (KIP-714) #14767

Merged

3 tasks

Addressing review comments

c5e2098

apoorvmittal10 requested review from AndrewJSchofield and junrao November 16, 2023 18:32

mjsax added core Kafka Broker kip Requires or implements a KIP labels Nov 16, 2023

junrao reviewed Nov 17, 2023

View reviewed changes

Moved client manager initialization, addressed comments

177edb4

apoorvmittal10 requested a review from junrao November 17, 2023 20:51

junrao reviewed Nov 17, 2023

View reviewed changes

Updated files as per review comments

83963a0

apoorvmittal10 requested a review from junrao November 19, 2023 19:28

apoorvmittal10 closed this Nov 23, 2023

apoorvmittal10 reopened this Nov 23, 2023

apoorvmittal10 mentioned this pull request Nov 23, 2023

MINOR: Always send cumulative failed dirs in HB request #14770

Merged

3 tasks

apoorvmittal10 added 2 commits November 23, 2023 11:22

Merge remote-tracking branch 'upstream/trunk' into kip-714-ak-manager

f9efcfb

Merge remote-tracking branch 'upstream/trunk' into kip-714-ak-manager

d4ea0a5

AndrewJSchofield reviewed Nov 24, 2023

View reviewed changes

Comment thread clients/src/main/java/org/apache/kafka/common/protocol/Errors.java Outdated

Adding . in the error message

b92d5fa

apoorvmittal10 requested a review from AndrewJSchofield November 24, 2023 17:50

Merge remote-tracking branch 'upstream/trunk' into kip-714-ak-manager

46ad39f

AndrewJSchofield approved these changes Nov 24, 2023

View reviewed changes

junrao closed this Nov 28, 2023

junrao reopened this Nov 28, 2023

apoorvmittal10 closed this Nov 29, 2023

apoorvmittal10 reopened this Nov 29, 2023

Merge remote-tracking branch 'upstream/trunk' into kip-714-ak-manager

deef326

junrao approved these changes Nov 29, 2023

View reviewed changes

junrao merged commit f1819f4 into apache:trunk Nov 29, 2023

apoorvmittal10 deleted the kip-714-ak-manager branch January 5, 2024 12:47

chia7712 reviewed Aug 6, 2026

View reviewed changes

Uh oh!

Conversation

apoorvmittal10 commented Nov 3, 2023

Committer Checklist (excluded from commit message)

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

apoorvmittal10 Nov 10, 2023 • 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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

apoorvmittal10 commented Nov 7, 2023

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

apoorvmittal10 commented Nov 13, 2023

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

apoorvmittal10 commented Nov 16, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

apoorvmittal10 commented Nov 17, 2023

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

apoorvmittal10 Nov 10, 2023 •

edited

Loading