private Optional<Builder<?>> createPushRequest(ClientTelemetrySubscription localSubscription, boolean terminating) {
            byte[] payload;
            try (MetricsEmitter emitter = new ClientTelemetryEmitter(localSubscription.selector(), localSubscription.deltaTemporality())) {
                emitter.init();
                kafkaMetricsCollector.collect(emitter);
                payload = createPayload(emitter.emittedMetrics());
            } catch (Exception e) {
                log.warn("Error constructing client telemetry payload: ", e);
                // Update last accessed time for push request to be retried on next interval.
                updateErrorResult(localSubscription.pushIntervalMs, time.milliseconds());
                return Optional.empty();
            }

            CompressionType compressionType = ClientTelemetryUtils.preferredCompressionType(localSubscription.acceptedCompressionTypes());
            byte[] compressedPayload;
            try {
                compressedPayload = ClientTelemetryUtils.compress(payload, compressionType);
            } catch (IOException e) {
                log.info("Failed to compress telemetry payload for compression: {}, sending uncompressed data", compressionType);
                compressedPayload = payload;
                compressionType = CompressionType.NONE;
            }

            AbstractRequest.Builder<?> requestBuilder = new PushTelemetryRequest.Builder(
                new PushTelemetryRequestData()
                    .setClientInstanceId(localSubscription.clientInstanceId())
                    .setSubscriptionId(localSubscription.subscriptionId())
                    .setTerminating(terminating)
                    .setCompressionType(compressionType.id)
                    .setMetrics(compressedPayload), true);

            return Optional.of(requestBuilder);
        }
Method is invoked whenever a metric is removed
clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporter.java