You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "apoorvmittal10 (via GitHub)" <gi...@apache.org> on 2024/10/11 23:41:32 UTC
[PR] KAFKA-17778: Add listener to remove client instances on connection disconnect [kafka]
apoorvmittal10 opened a new pull request, #17474:
URL: https://github.com/apache/kafka/pull/17474
The PR uses the capability added in PR: https://github.com/apache/kafka/pull/17473, which allows to attach listener for client instances cleanup on client disconnection.
The ClientsMetricManager keeps the client instance cache at a limit of 16384. The active connections can be low, but connections can be created and destroyed in short span. Which hits the cache limit and results in cache miss for new connections.
The client instance cache keeps the instances for 3 * push interval ms. Hence when the cache gets full it creates new instances for every request while evicting others from cache. This gives some bumps to the GC for the broker and eventually CPU. Though with evicting early it will still be garbage collected but the long running active connections will not be removed from the cache.
This PR adds capability to evict client instances from cache when the connection from client is dropped.
Note: There exists a cleanup task which evicts the instances on timebound expiry. I haven't removed that as of now, we should have that for now and should deprecate slowly in future releases.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17474:
URL: https://github.com/apache/kafka/pull/17474#discussion_r1801147606
##########
server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java:
##########
@@ -114,13 +117,15 @@ public ClientMetricsManager(ClientMetricsReceiverPlugin receiverPlugin, int clie
this.subscriptionMap = new ConcurrentHashMap<>();
this.subscriptionUpdateVersion = new AtomicInteger(0);
this.clientInstanceCache = new SynchronizedCache<>(new LRUCache<>(CACHE_MAX_SIZE));
+ this.clientConnectionIdMap = new ConcurrentHashMap<>();
this.expirationTimer = new SystemTimerReaper(CLIENT_METRICS_REAPER_THREAD_NAME, new SystemTimer("client-metrics"));
this.clientTelemetryMaxBytes = clientTelemetryMaxBytes;
this.time = time;
this.cacheExpiryMs = cacheExpiryMs;
this.lastCacheErrorLogMs = new AtomicLong(0);
this.metrics = metrics;
this.clientMetricsStats = new ClientMetricsStats();
+ this.connectionDisconnectListener = new ClientConnectionDisconnectListener(clientInstanceCache, clientConnectionIdMap, clientMetricsStats);
Review Comment:
Agree, moved the class to non-static and removed these.
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #17474:
URL: https://github.com/apache/kafka/pull/17474#issuecomment-2410485313
>So, I'm wondering—will terminating still work if this PR removes the client instance on disconnection?
Though terminating is good enough to not accept further push telemetry calls from same client but if we remove the instance from cache on `terminating` flag then it will open us for attacks with rogue clients which can send telemetry push request with terminating flag as true and we will skip push time based throttling for this special push request. Hence it would be better to keep these 2 semantics separately.
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #17474:
URL: https://github.com/apache/kafka/pull/17474#issuecomment-2408639782
Thanks for reviewing and feedback. Here are my thoughts.
> I assume the 'short span' is caused by an unexpected connection issue rather than a malicious client, as preventing malicious clients is a different issue. The root cause of hitting the cache limit is that ClientMetricsManager creates many meaningless clientInstanceIds for the 'short span' because their GetTelemetrySubscriptionsRequest doesn't include a valid UUID.
Though not neccessarily. There happens to be workloads where application creates short lived producer/consumers which does specific read/write and closes. The TTL for client metrics is 3x of push interval which makes the instances to live longer in cache. For a push interval of say 5 minutes (default push interval), the instances can live for 15 minutes in cache. Hence if just 50 connections are created and closed per sec then cache can get full in 6 minutes.
So there could be 3 questions:
1. Why short lived clients are created? This depends on application but this is seen in numerous cases where short lived client connections are created.
2. Is 16384 limit enough? Though the limit seems to be a good enough but it will depend on the running infrastructure and what broker can support.
3. Is 5 minutes default push interval too large? It could be lower or higher depends how much cost an operator wants to have to store the client metrics as the clients can be numerous hence cardinality will be high. So a higher push interval makes sense, but it can be dynamically changed is some client seems to be problematic and operator decided to fetch metrics aggressively to determinw the issue.
So, I think a clear way to clean up resources is missing in Kafka where decision can be made on client disconnection. Rather waiting for a request from client which specifies the client is being disconnected.
> Perhaps we could update the protocol to require that GetTelemetrySubscriptionsRequest must include a client instance ID? In KIP-1082, we agreed that clients are capable of generating UUIDs. With this approach, ClientMetricsManager would no longer create meaningless clientInstanceIds for short spans, as they would always use the same UUID (acting as a kind of incarnation).
This might not solve the problem specified in 1 above. As when new client is created we should have a unique client instance id to determine uniquene instance. As the client metrics are tied to specific instance which is being created.
In general, the requirement for connection disconnet listener can also address use cases where closing request from client is never reached to broker. Say for Share Groups where the records fetched by a client instance and not yet acknowledged waits for the lock timeout to be re-delivered, can use this functionality to release such records where disconnect request is missed.
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #17474:
URL: https://github.com/apache/kafka/pull/17474#issuecomment-2408775196
@apoorvmittal10 Thanks for sharing. The disconnection listener seems like a good approach, but I’d like to have more discussion for my own understanding.
> This might not solve the problem specified in 1 above. As when new client is created we should have a unique client instance id to determine uniquene instance. As the client metrics are tied to specific instance which is being created.
I assume there are three scenarios for resending GET_TELEMETRY_SUBSCRIPTION:
1. The client is live but encounters a network issue, so it retries the request.
2. The client is restarted before completing the GET_TELEMETRY_SUBSCRIPTION.
3. The client is restarted after completing the GET_TELEMETRY_SUBSCRIPTION.
The connection listener works well for handling cases 1 and 2, but case 3 remains a problem, correct? In case 3, the connection listener isn’t triggered to remove the client from the cache, yet the user application will eventually create another client instance.
If our goal is to avoid the cache filling up with ‘ghost’ clients, perhaps we could introduce prioritization in the cache. The priority of a client instance could be based on either the number of PUSH_TELEMETRY_REQUEST calls or its lifetime.
WDYT?
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17474:
URL: https://github.com/apache/kafka/pull/17474#discussion_r1801149423
##########
server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java:
##########
@@ -459,6 +469,48 @@ Timer expirationTimer() {
return expirationTimer;
}
+ // Visible for testing
+ Map<String, Uuid> clientConnectionIdMap() {
+ return clientConnectionIdMap;
+ }
+
+ private final class ClientConnectionDisconnectListener implements ConnectionDisconnectListener {
+
+ private final Cache<Uuid, ClientMetricsInstance> clientInstanceCache;
+ private final Map<String, Uuid> clientConnectionIdMap;
+ private final ClientMetricsStats clientMetricsStats;
+
+ ClientConnectionDisconnectListener(
+ Cache<Uuid, ClientMetricsInstance> clientInstanceCache,
+ Map<String, Uuid> clientConnectionIdMap,
+ ClientMetricsStats clientMetricsStats
+ ) {
+ this.clientInstanceCache = clientInstanceCache;
+ this.clientConnectionIdMap = clientConnectionIdMap;
+ this.clientMetricsStats = clientMetricsStats;
+ }
+
+ @Override
+ public void onDisconnect(String connectionId) {
+ log.trace("Removing client connection id [{}] from the client instance cache", connectionId);
+
+ Uuid clientInstanceId = clientConnectionIdMap.remove(connectionId);
+ if (clientInstanceId == null) {
+ log.trace("Client connection id [{}] is not found in the client instance cache", connectionId);
+ return;
+ }
+
+ // Unregister the client instance metrics from the broker metrics.
+ clientMetricsStats.unregisterClientInstanceMetrics(clientInstanceId);
+
+ ClientMetricsInstance clientInstance = clientInstanceCache.get(clientInstanceId);
+ if (clientInstance != null) {
+ clientInstance.cancelExpirationTimerTask();
+ clientInstanceCache.remove(clientInstanceId);
Review Comment:
If a broker is restarted then anyways the cache will be clean. I am not sure if I understood the question right. I must be missing somehting to understand.
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #17474:
URL: https://github.com/apache/kafka/pull/17474#discussion_r1800244484
##########
server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java:
##########
@@ -114,13 +117,15 @@ public ClientMetricsManager(ClientMetricsReceiverPlugin receiverPlugin, int clie
this.subscriptionMap = new ConcurrentHashMap<>();
this.subscriptionUpdateVersion = new AtomicInteger(0);
this.clientInstanceCache = new SynchronizedCache<>(new LRUCache<>(CACHE_MAX_SIZE));
+ this.clientConnectionIdMap = new ConcurrentHashMap<>();
this.expirationTimer = new SystemTimerReaper(CLIENT_METRICS_REAPER_THREAD_NAME, new SystemTimer("client-metrics"));
this.clientTelemetryMaxBytes = clientTelemetryMaxBytes;
this.time = time;
this.cacheExpiryMs = cacheExpiryMs;
this.lastCacheErrorLogMs = new AtomicLong(0);
this.metrics = metrics;
this.clientMetricsStats = new ClientMetricsStats();
+ this.connectionDisconnectListener = new ClientConnectionDisconnectListener(clientInstanceCache, clientConnectionIdMap, clientMetricsStats);
Review Comment:
ClientConnectionDisconnectListener is a private class and can use all members in the parent class. Do we need to pass in clientInstanceCache, clientConnectionIdMap, clientMetricsStats?
##########
server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java:
##########
@@ -459,6 +469,48 @@ Timer expirationTimer() {
return expirationTimer;
}
+ // Visible for testing
+ Map<String, Uuid> clientConnectionIdMap() {
+ return clientConnectionIdMap;
+ }
+
+ private final class ClientConnectionDisconnectListener implements ConnectionDisconnectListener {
+
+ private final Cache<Uuid, ClientMetricsInstance> clientInstanceCache;
+ private final Map<String, Uuid> clientConnectionIdMap;
+ private final ClientMetricsStats clientMetricsStats;
+
+ ClientConnectionDisconnectListener(
+ Cache<Uuid, ClientMetricsInstance> clientInstanceCache,
+ Map<String, Uuid> clientConnectionIdMap,
+ ClientMetricsStats clientMetricsStats
+ ) {
+ this.clientInstanceCache = clientInstanceCache;
+ this.clientConnectionIdMap = clientConnectionIdMap;
+ this.clientMetricsStats = clientMetricsStats;
+ }
+
+ @Override
+ public void onDisconnect(String connectionId) {
+ log.trace("Removing client connection id [{}] from the client instance cache", connectionId);
+
+ Uuid clientInstanceId = clientConnectionIdMap.remove(connectionId);
+ if (clientInstanceId == null) {
+ log.trace("Client connection id [{}] is not found in the client instance cache", connectionId);
+ return;
+ }
+
+ // Unregister the client instance metrics from the broker metrics.
+ clientMetricsStats.unregisterClientInstanceMetrics(clientInstanceId);
+
+ ClientMetricsInstance clientInstance = clientInstanceCache.get(clientInstanceId);
+ if (clientInstance != null) {
+ clientInstance.cancelExpirationTimerTask();
+ clientInstanceCache.remove(clientInstanceId);
Review Comment:
This means that every time we restart a broker, we could prematurely remove a live clientInstanceId. Do we gain more than we lose?
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #17474:
URL: https://github.com/apache/kafka/pull/17474#issuecomment-2410292857
>I feel this situation is similar to the consumer membership issue.
To some extent yes.
>What do you think about using RPC-level cleanup by sending a DeleteTelemetrySubscriptions request?
I think if we add a definite way of determining when client connection is terminated at broker then we won't require additional explicit delete RPC from clients. However depends on the behaviour we want in other places i.e. for client telemetry our goal is to do minimum compute when a telemetry request is received hence keep the cache. Might be somewhat different for consumers as there we want to save on rebalance, hence we might want if same client reconnects with same membership data.
>So, what happens if a client sends a PushTelemetry request after re-establishing an idle connection?
The request will be fulfilled and a client instance entry in cache will be created.
The following happens: the push telemetry requires 2 things: client instance id and subscription id. If both matches then request can be fulfilled. If client instance is not present in the cache with instance id then one is created, though subscription id checksum is computed with crc32 against matching subscriptions. Hence if same instance id along matching subscription id is used then request will be processed.
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao merged PR #17474:
URL: https://github.com/apache/kafka/pull/17474
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #17474:
URL: https://github.com/apache/kafka/pull/17474#issuecomment-2410389870
With respect to a DeleteTelemetrySubscriptions request, there is already the special "terminating" PushTelemetry request, so it seems unnecessary to me to have both.
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #17474:
URL: https://github.com/apache/kafka/pull/17474#issuecomment-2410218821
@chia7712 Does this sound good?
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #17474:
URL: https://github.com/apache/kafka/pull/17474#issuecomment-2410535536
> Though terminating is good enough to not accept further push telemetry calls from same client but if we remove the instance from cache on terminating flag
That's what we discussed on [KAFKA-17760](https://issues.apache.org/jira/browse/KAFKA-17760), but my new question is regarding the PR description: 'This PR adds the capability to evict client instances from the cache when the connection from the client is dropped.'
A normal client that is closing will send a request to set terminating=true and then close the connection, which results in the client instance being evicted from the cache. This seems like a kind of two-phase close: the first phase protects against rogue clients by setting terminating=true, and the second phase ensures the cache is protected from overloading. Is that correct?
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #17474:
URL: https://github.com/apache/kafka/pull/17474#issuecomment-2408242643
Build will pass once this PR is rebased over dependent PR: https://github.com/apache/kafka/pull/17473
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #17474:
URL: https://github.com/apache/kafka/pull/17474#issuecomment-2409130891
>If our goal is to avoid the cache filling up with ‘ghost’ clients
Not neccesarily ghost clients but also the clients which are spawned for really short span and stopped. For example the issue fixed in the PR: https://github.com/apache/kafka/pull/17431 should never be encountered if a Kafka Consumer is not closed just after instantiation. But we have seen applications which does have such workloads i.e. clients are created for a very short span as well. Hence those instances are legit but shall occupy a cache entry till time based eviction happens.
>perhaps we could introduce prioritization in the cache. The priority of a client instance could be based on either the number of PUSH_TELEMETRY_REQUEST calls or its lifetime
This is a good idea. But do we want to have additional compute and tracking for number of requests as well in the cache? I think that will be extra and not required when our goal is to get rid of instances which will never send the telemetry data again i.e. their connections are not disconnected.
With this PR, we are not solving the scenario when a broker is handling the long running connections for 16384 clients, hence cache can again get full. Because that requires separate handling, if a broker can handle such large number of long running active clients then the value of cache should be bumped. I can do a follow up PR to make the cache config exposed in server.properties.
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17474:
URL: https://github.com/apache/kafka/pull/17474#discussion_r1803728797
##########
server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java:
##########
@@ -459,6 +469,48 @@ Timer expirationTimer() {
return expirationTimer;
}
+ // Visible for testing
+ Map<String, Uuid> clientConnectionIdMap() {
+ return clientConnectionIdMap;
+ }
+
+ private final class ClientConnectionDisconnectListener implements ConnectionDisconnectListener {
+
+ private final Cache<Uuid, ClientMetricsInstance> clientInstanceCache;
+ private final Map<String, Uuid> clientConnectionIdMap;
+ private final ClientMetricsStats clientMetricsStats;
+
+ ClientConnectionDisconnectListener(
+ Cache<Uuid, ClientMetricsInstance> clientInstanceCache,
+ Map<String, Uuid> clientConnectionIdMap,
+ ClientMetricsStats clientMetricsStats
+ ) {
+ this.clientInstanceCache = clientInstanceCache;
+ this.clientConnectionIdMap = clientConnectionIdMap;
+ this.clientMetricsStats = clientMetricsStats;
+ }
+
+ @Override
+ public void onDisconnect(String connectionId) {
+ log.trace("Removing client connection id [{}] from the client instance cache", connectionId);
+
+ Uuid clientInstanceId = clientConnectionIdMap.remove(connectionId);
+ if (clientInstanceId == null) {
+ log.trace("Client connection id [{}] is not found in the client instance cache", connectionId);
+ return;
+ }
+
+ // Unregister the client instance metrics from the broker metrics.
+ clientMetricsStats.unregisterClientInstanceMetrics(clientInstanceId);
+
+ ClientMetricsInstance clientInstance = clientInstanceCache.get(clientInstanceId);
+ if (clientInstance != null) {
+ clientInstance.cancelExpirationTimerTask();
+ clientInstanceCache.remove(clientInstanceId);
Review Comment:
>Does that mean we shouldn't evict from the cache for clients marked as terminating? If we do that, do we really get any benefit since most of the clients will go through the terminating process?
What I meant was that we should keep terminating flag behaviour separately from cache eviction. We shouldn't evict from cache based on terminating flag but should only evict when connection is dropped.
Say, there is a rogue client which sends telemetry push requests always with terminating as true and on a same open connection. Then because the instance will be in cache as connection is still open then we can deny the request. Hence to protect broker and down stream systems, we should always act on client disconnection.
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #17474:
URL: https://github.com/apache/kafka/pull/17474#issuecomment-2419139241
@chia7712 @junrao I have rebased this PR now and build passed. Can you please take a look.
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #17474:
URL: https://github.com/apache/kafka/pull/17474#issuecomment-2410220855
@apoorvmittal10 Thanks for the feedback. I have a few questions about this PR—thanks for your patience.
> Not neccesarily ghost clients but also the clients which are spawned for really short span and stopped. For example the issue fixed in the PR: https://github.com/apache/kafka/pull/17431 should never be encountered if a Kafka Consumer is not closed just after instantiation. But we have seen applications which does have such workloads i.e. clients are created for a very short span as well. Hence those instances are legit but shall occupy a cache entry till time based eviction happens.
I feel this situation is similar to the consumer membership issue. What do you think about using RPC-level cleanup by sending a `DeleteTelemetrySubscriptions` request? On the other hand, should we also remove consumer members using this connection listener? The connection-level cleanup has the advantage of handling cases where the client can't send a request to leave. However, on the client side, idle connections can be validly removed, and the default idle timeout is 5 minutes, which is shorter than the TTL for telemetry. So, what happens if a client sends a `PushTelemetry` request after re-establishing an idle connection?
> But do we want to have additional compute and tracking for number of requests as well in the cache? I think that will be extra and not required when our goal is to get rid of instances which will never send the telemetry data again i.e. their connections are not disconnected.
yes, you are right.
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #17474:
URL: https://github.com/apache/kafka/pull/17474#issuecomment-2410542237
> A normal client that is closing will send a request to set terminating=true and then close the connection, which results in the client instance being evicted from the cache. This seems like a kind of two-phase close: the first phase protects against rogue clients by setting terminating=true, and the second phase ensures the cache is protected from overloading. Is that correct?
Yes, you are right. That's the behaviour.
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #17474:
URL: https://github.com/apache/kafka/pull/17474#issuecomment-2410573174
@apoorvmittal10 Sorry that I have a last question:
> Note: There exists a cleanup task which evicts the instances on timebound expiry. I haven't removed that as of now, we should have that for now and should deprecate slowly in future releases.
If we remove the cleanup task in the future, how would we handle connections that initially send telemetry-related requests but then remain alive by sending other non-telemetry requests? I assume they should still expire and be removed by a cleanup task.
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #17474:
URL: https://github.com/apache/kafka/pull/17474#issuecomment-2410580413
> If we remove the cleanup task in the future, how would we handle connections that initially send telemetry-related requests but then remain alive by sending other non-telemetry requests? I assume they should still expire and be removed by a cleanup task.
Hmmm good point. Idle connections cleanup still require the background time based clean. I am updating the description by this context.
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #17474:
URL: https://github.com/apache/kafka/pull/17474#discussion_r1803537126
##########
server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java:
##########
@@ -459,6 +469,48 @@ Timer expirationTimer() {
return expirationTimer;
}
+ // Visible for testing
+ Map<String, Uuid> clientConnectionIdMap() {
+ return clientConnectionIdMap;
+ }
+
+ private final class ClientConnectionDisconnectListener implements ConnectionDisconnectListener {
+
+ private final Cache<Uuid, ClientMetricsInstance> clientInstanceCache;
+ private final Map<String, Uuid> clientConnectionIdMap;
+ private final ClientMetricsStats clientMetricsStats;
+
+ ClientConnectionDisconnectListener(
+ Cache<Uuid, ClientMetricsInstance> clientInstanceCache,
+ Map<String, Uuid> clientConnectionIdMap,
+ ClientMetricsStats clientMetricsStats
+ ) {
+ this.clientInstanceCache = clientInstanceCache;
+ this.clientConnectionIdMap = clientConnectionIdMap;
+ this.clientMetricsStats = clientMetricsStats;
+ }
+
+ @Override
+ public void onDisconnect(String connectionId) {
+ log.trace("Removing client connection id [{}] from the client instance cache", connectionId);
+
+ Uuid clientInstanceId = clientConnectionIdMap.remove(connectionId);
+ if (clientInstanceId == null) {
+ log.trace("Client connection id [{}] is not found in the client instance cache", connectionId);
+ return;
+ }
+
+ // Unregister the client instance metrics from the broker metrics.
+ clientMetricsStats.unregisterClientInstanceMetrics(clientInstanceId);
+
+ ClientMetricsInstance clientInstance = clientInstanceCache.get(clientInstanceId);
+ if (clientInstance != null) {
+ clientInstance.cancelExpirationTimerTask();
+ clientInstanceCache.remove(clientInstanceId);
Review Comment:
You are right. Restarting a broker doesn't cause additional issues.
> Though terminating is good enough to not accept further push telemetry calls from same client but if we remove the instance from cache on terminating flag then it will open us for attacks with rogue clients which can send telemetry push request with terminating flag as true and we will skip push time based throttling for this special push request. Hence it would be better to keep these 2 semantics separately.
Does that mean we shouldn't evict from the cache for clients marked as terminating? If we do that, do we really get any benefit since most of the clients will go through the terminating process?
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #17474:
URL: https://github.com/apache/kafka/pull/17474#discussion_r1803867963
##########
server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java:
##########
@@ -459,6 +469,48 @@ Timer expirationTimer() {
return expirationTimer;
}
+ // Visible for testing
+ Map<String, Uuid> clientConnectionIdMap() {
+ return clientConnectionIdMap;
+ }
+
+ private final class ClientConnectionDisconnectListener implements ConnectionDisconnectListener {
+
+ private final Cache<Uuid, ClientMetricsInstance> clientInstanceCache;
+ private final Map<String, Uuid> clientConnectionIdMap;
+ private final ClientMetricsStats clientMetricsStats;
+
+ ClientConnectionDisconnectListener(
+ Cache<Uuid, ClientMetricsInstance> clientInstanceCache,
+ Map<String, Uuid> clientConnectionIdMap,
+ ClientMetricsStats clientMetricsStats
+ ) {
+ this.clientInstanceCache = clientInstanceCache;
+ this.clientConnectionIdMap = clientConnectionIdMap;
+ this.clientMetricsStats = clientMetricsStats;
+ }
+
+ @Override
+ public void onDisconnect(String connectionId) {
+ log.trace("Removing client connection id [{}] from the client instance cache", connectionId);
+
+ Uuid clientInstanceId = clientConnectionIdMap.remove(connectionId);
+ if (clientInstanceId == null) {
+ log.trace("Client connection id [{}] is not found in the client instance cache", connectionId);
+ return;
+ }
+
+ // Unregister the client instance metrics from the broker metrics.
+ clientMetricsStats.unregisterClientInstanceMetrics(clientInstanceId);
+
+ ClientMetricsInstance clientInstance = clientInstanceCache.get(clientInstanceId);
+ if (clientInstance != null) {
+ clientInstance.cancelExpirationTimerTask();
+ clientInstanceCache.remove(clientInstanceId);
Review Comment:
So, we will always evict the client from the cache on connection close. But if a client is terminating, the broker will still remember the client uuid somewhere else to protect against rogue client. Is that correct?
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17474:
URL: https://github.com/apache/kafka/pull/17474#discussion_r1803880654
##########
server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java:
##########
@@ -459,6 +469,48 @@ Timer expirationTimer() {
return expirationTimer;
}
+ // Visible for testing
+ Map<String, Uuid> clientConnectionIdMap() {
+ return clientConnectionIdMap;
+ }
+
+ private final class ClientConnectionDisconnectListener implements ConnectionDisconnectListener {
+
+ private final Cache<Uuid, ClientMetricsInstance> clientInstanceCache;
+ private final Map<String, Uuid> clientConnectionIdMap;
+ private final ClientMetricsStats clientMetricsStats;
+
+ ClientConnectionDisconnectListener(
+ Cache<Uuid, ClientMetricsInstance> clientInstanceCache,
+ Map<String, Uuid> clientConnectionIdMap,
+ ClientMetricsStats clientMetricsStats
+ ) {
+ this.clientInstanceCache = clientInstanceCache;
+ this.clientConnectionIdMap = clientConnectionIdMap;
+ this.clientMetricsStats = clientMetricsStats;
+ }
+
+ @Override
+ public void onDisconnect(String connectionId) {
+ log.trace("Removing client connection id [{}] from the client instance cache", connectionId);
+
+ Uuid clientInstanceId = clientConnectionIdMap.remove(connectionId);
+ if (clientInstanceId == null) {
+ log.trace("Client connection id [{}] is not found in the client instance cache", connectionId);
+ return;
+ }
+
+ // Unregister the client instance metrics from the broker metrics.
+ clientMetricsStats.unregisterClientInstanceMetrics(clientInstanceId);
+
+ ClientMetricsInstance clientInstance = clientInstanceCache.get(clientInstanceId);
+ if (clientInstance != null) {
+ clientInstance.cancelExpirationTimerTask();
+ clientInstanceCache.remove(clientInstanceId);
Review Comment:
Yes we will evict the instance on connection close. But once the instance is evicted then the broker can't remember the UUID. The 2 scenarios I am talking about are
1. On existing connection a rogue client sends all telemetry push metrics with flag as terminating: true: a) As the connection is not closed hence instance is not evicted, though client sends the rpc as terminating: true but we have the instance cached which errors out the requests.
2. When connection is closed: The scenario is same as earlier, before this PR. When a new client is spawned after closed connection then anyways new instance id is generated which has it's own respective checks.
By not evicting client instances on terminating:true we are handling the 1st scenario mentioned above.
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #17474:
URL: https://github.com/apache/kafka/pull/17474#issuecomment-2408616573
> The ClientsMetricManager keeps the client instance cache at a limit of 16384. The active connections can be low, but connections can be created and destroyed in short span. Which hits the cache limit and results in cache miss for new connections.
I assume the 'short span' is caused by an unexpected connection issue rather than a malicious client, as preventing malicious clients is a different issue. The root cause of hitting the cache limit is that ClientMetricsManager creates many meaningless clientInstanceIds for the 'short span' because their GetTelemetrySubscriptionsRequest doesn't include a valid UUID.
Perhaps we could update the protocol to require that GetTelemetrySubscriptionsRequest must include a client instance ID? In KIP-1082, we agreed that clients are capable of generating UUIDs. With this approach, ClientMetricsManager would no longer create meaningless clientInstanceIds for short spans, as they would always use the same UUID (acting as a kind of incarnation).
@apoorvmittal10 WDYT?
--
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-17778: Add listener to remove client instances on connection disconnect [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #17474:
URL: https://github.com/apache/kafka/pull/17474#issuecomment-2410471529
> With respect to a DeleteTelemetrySubscriptions request, there is already the special "terminating" PushTelemetry request, so it seems unnecessary to me to have both.
That's a good point. The terminating flag is used to prevent clients from sending telemetry requests unless they're evicted from the cache. So, I'm wondering—will terminating still work if this PR removes the client instance on disconnection?
--
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