apoorvmittal10 commented on code in PR #15234:
URL: https://github.com/apache/kafka/pull/15234#discussion_r1461574809
##########
server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java:
##########
@@ -433,4 +462,35 @@ public Map<String, Pattern> matchPattern() {
return matchPattern;
}
}
+
+ private final class ExpirationTimerTask extends TimerTask {
+
+ private static final long CACHE_ERROR_LOG_INTERVAL_MS = 5 * 60 * 1000;
// 5 minutes
+
+ private final Uuid uuid;
+
+ private ExpirationTimerTask(Uuid uuid, long delayMs) {
+ super(delayMs);
+ this.uuid = uuid;
+ }
+
+ @Override
+ public void run() {
+ log.trace("Expiration timer task run for client instance id: {},
after delay ms: {}", uuid, delayMs);
+ if (!clientInstanceCache.remove(uuid)) {
+ /*
+ This can only happen if the client instance is removed from
the cache by the LRU
+ eviction policy before the expiration timer task is executed.
Log a warning as broker
+ cache is not able to hold all the client instances. Log only
once every CACHE_ERROR_LOG_INTERVAL_MS
+ to avoid flooding the logs.
+ */
+ long lastErrorMs = lastCacheErrorLogMs.get();
+ if (time.milliseconds() - lastErrorMs >
CACHE_ERROR_LOG_INTERVAL_MS &&
+ lastCacheErrorLogMs.compareAndSet(lastErrorMs,
time.milliseconds())) {
Review Comment:
I agree, and if I am not wrong then that's what happen with above code. Am I
missing something?
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
