private void collectSum(MetricKey metricKey, double value, MetricsEmitter metricsEmitter, Instant timestamp) {
        if (!metricsEmitter.shouldEmitMetric(metricKey)) {
            return;
        }

        if (metricsEmitter.shouldEmitDeltaMetrics()) {
            InstantAndValue<Double> instantAndValue = ledger.delta(metricKey, timestamp, value);

            metricsEmitter.emitMetric(
                SinglePointMetric.deltaSum(metricKey, instantAndValue.getValue(), true, timestamp,
                    instantAndValue.getIntervalStart(), excludeLabels)
            );
        } else {
            metricsEmitter.emitMetric(
                SinglePointMetric.sum(metricKey, value, true, timestamp, ledger.instantAdded(metricKey), excludeLabels)
            );
        }
    }
This is called whenever temporality changes, resets the value tracker for metrics.
clients/src/main/java/org/apache/kafka/common/telemetry/internals/KafkaMetricsCollector.java