@Override
        public void run() {
            log.trace("Expiration timer task run for client instance id: {}, after delay ms: {}", clientInstanceId, delayMs);
            clientMetricsStats.unregisterClientInstanceMetrics(clientInstanceId);
            if (!clientInstanceCache.remove(clientInstanceId)) {
                /*
                 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())) {
                    log.warn("Client metrics instance cache cannot find the client instance id: {}. The cache"
                        + " must be at capacity, size: {} ", clientInstanceId, clientInstanceCache.size());
                }
            }
        }
Computes the SubscriptionId as a unique identifier for a client instance's subscription set,
the id is generated by calculating a CRC32C of the configured metrics subscriptions including
the PushIntervalMs, XORed with the ClientInstanceId.
server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java