private Optional<Builder<?>> createPushRequest(ClientTelemetrySubscription localSubscription) {
            if (localSubscription == null) {
                log.warn("Telemetry state is {} but subscription is null; not sending telemetry", state);
                if (!maybeSetState(ClientTelemetryState.SUBSCRIPTION_NEEDED)) {
                    log.warn("Unable to transition state after failed create push telemetry from state {}", state);
                }
                return Optional.empty();
            }

            log.debug("Creating telemetry push request with client instance id {}", localSubscription.clientInstanceId());
            /*
             Don't send a push request if we don't have the collector initialized. Re-attempt
             the push on the next interval.
            */
            if (kafkaMetricsCollector == null) {
                log.warn("Cannot make telemetry request as collector is not initialized");
                // Update last accessed time for push request to be retried on next interval.
                updateErrorResult(localSubscription.pushIntervalMs, time.milliseconds());
                return Optional.empty();
            }

            boolean terminating;
            lock.writeLock().lock();
            try {
                /*
                 We've already been terminated, or we've already issued our last push, so we
                 should just exit now.
                */
                if (state == ClientTelemetryState.TERMINATED || state == ClientTelemetryState.TERMINATING_PUSH_IN_PROGRESS) {
                    return Optional.empty();
                }

                /*
                 Check the *actual* state (while locked) to make sure we're still in the state
                 we expect to be in.
                */
                terminating = state == ClientTelemetryState.TERMINATING_PUSH_NEEDED;
                if (!maybeSetState(terminating ? ClientTelemetryState.TERMINATING_PUSH_IN_PROGRESS : ClientTelemetryState.PUSH_IN_PROGRESS)) {
                    return Optional.empty();
                }
            } finally {
                lock.writeLock().unlock();
            }

            return createPushRequest(localSubscription, terminating);
        }
Method is invoked whenever a metric is removed
clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporter.java