Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/TaskMetrics.java (Java)

Code before:
                                          final String descriptionOfAvg,
                                          final String descriptionOfMax,
                                          final RecordingLevel recordingLevel,
                                          final StreamsMetricsImpl streamsMetrics,
                                          final Sensor... parentSensors) {
        final Sensor sensor = streamsMetrics.taskLevelSensor(threadId, taskId, metricName, recordingLevel, parentSensors);
        final Map<String, String> tagMap = streamsMetrics.taskLevelTagMap(threadId, taskId);
        addAvgAndMaxToSensor(
            sensor,
            TASK_LEVEL_GROUP,
            tagMap,
            metricName,
            descriptionOfAvg,
            descriptionOfMax
        );
        return sensor;
    }

    private static Sensor invocationRateAndCountAndAvgAndMaxLatencySensor(final String threadId,
                                                                          final String taskId,
                                                                          final String metricName,
                                                                          final String descriptionOfRate,
                                                                          final String descriptionOfCount,
                                                                          final String descriptionOfAvg,
                                                                          final String descriptionOfMax,
                                                                          final RecordingLevel recordingLevel,
                                                                          final StreamsMetricsImpl streamsMetrics,
                                                                          final Sensor... parentSensors) {
        final Sensor sensor = streamsMetrics.taskLevelSensor(threadId, taskId, metricName, recordingLevel, parentSensors);
        final Map<String, String> tagMap = streamsMetrics.taskLevelTagMap(threadId, taskId);
        addAvgAndMaxToSensor(
            sensor,
            TASK_LEVEL_GROUP,
            tagMap,
            metricName + LATENCY_SUFFIX,
            descriptionOfAvg,
            descriptionOfMax
        );
        addInvocationRateAndCountToSensor(
            sensor,
            TASK_LEVEL_GROUP,
            tagMap,
            metricName,
            descriptionOfRate,
            descriptionOfCount
        );
        return sensor;
    }
}


Review comment (security):
This is a piggy-backed metric fix: we should use cumulativeSum than cumulativeCount for dropped records, even though today with most callees as `sensor.record()` it is effectively the same as it only increment by 1, it is still vulnerable in case we record a non-one value in the future.

Code after:
            ENFORCED_PROCESSING_RATE_DESCRIPTION,
            ENFORCED_PROCESSING_TOTAL_DESCRIPTION,
            RecordingLevel.DEBUG,
            streamsMetrics,
            parentSensors
        );
    }

    public static Sensor recordLatenessSensor(final String threadId,
                                              final String taskId,
                                              final StreamsMetricsImpl streamsMetrics) {
        return avgAndMaxSensor(
            threadId,
            taskId,
            RECORD_LATENESS,
            RECORD_LATENESS_AVG_DESCRIPTION,
            RECORD_LATENESS_MAX_DESCRIPTION,
            RecordingLevel.DEBUG,
            streamsMetrics
        );
    }

    public static Sensor droppedRecordsSensor(final String threadId,
                                              final String taskId,
                                              final StreamsMetricsImpl streamsMetrics) {
        return invocationRateAndTotalSensor(
            threadId,
            taskId,
            DROPPED_RECORDS,
            DROPPED_RECORDS_RATE_DESCRIPTION,
            DROPPED_RECORDS_TOTAL_DESCRIPTION,
            RecordingLevel.INFO,
            streamsMetrics
        );
    }

    private static Sensor invocationRateAndCountSensor(final String threadId,
                                                       final String taskId,
                                                       final String operation,
                                                       final String descriptionOfRate,
                                                       final String descriptionOfCount,
                                                       final RecordingLevel recordingLevel,
                                                       final StreamsMetricsImpl streamsMetrics,
                                                       final Sensor... parentSensors) {
        final Sensor sensor = streamsMetrics.taskLevelSensor(threadId, taskId, operation, recordingLevel, parentSensors);
        addInvocationRateAndCountToSensor(
            sensor,
            TASK_LEVEL_GROUP,
            streamsMetrics.taskLevelTagMap(threadId, taskId),
            operation,
            descriptionOfRate,