@Override
        public Optional<Uuid> clientInstanceId(Duration timeout) {
            final long timeoutMs = timeout.toMillis();
            if (timeoutMs < 0) {
                throw new IllegalArgumentException("The timeout cannot be negative for fetching client instance id.");
            }

            lock.writeLock().lock();
            try {
                if (subscription == null) {
                    // If we have a non-negative timeout and no-subscription, let's wait for one to be retrieved.
                    log.debug("Waiting for telemetry subscription containing the client instance ID with timeoutMillis = {} ms.", timeoutMs);
                    try {
                        if (!subscriptionLoaded.await(timeoutMs, TimeUnit.MILLISECONDS)) {
                            log.debug("Wait for telemetry subscription elapsed; may not have actually loaded it");
                        }
                    } catch (InterruptedException e) {
                        throw new InterruptException(e);
                    }
                }

                if (subscription == null) {
                    log.debug("Client instance ID could not be retrieved with timeout {}", timeout);
                    return Optional.empty();
                }

                Uuid clientInstanceId = subscription.clientInstanceId();
                if (clientInstanceId == null) {
                    log.info("Client instance ID was null in telemetry subscription while in state {}", state);
                    return Optional.empty();
                }

                return Optional.of(clientInstanceId);
            } finally {
                lock.writeLock().unlock();
            }
        }
Method is invoked whenever a metric is removed
clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporter.java