public static Sensor activeBufferedRecordsSensor(final String threadId,
                                                     final String taskId,
                                                     final StreamsMetricsImpl streamsMetrics) {
        final String name = ACTIVE_TASK_PREFIX + BUFFER_COUNT;
        final Sensor sensor = streamsMetrics.taskLevelSensor(threadId, taskId, name, Sensor.RecordingLevel.DEBUG);
        addValueMetricToSensor(
            sensor,
            TASK_LEVEL_GROUP,
            streamsMetrics.taskLevelTagMap(threadId, taskId),
            name,
            NUM_BUFFERED_RECORDS_DESCRIPTION
        );
        return sensor;
    }
@Test
    public void shouldGetActiveBufferCountSensor() {
        final String operation = "active-buffer-count";
        when(streamsMetrics.taskLevelSensor(THREAD_ID, TASK_ID, operation, RecordingLevel.DEBUG))
                .thenReturn(expectedSensor);
        final String countDescription = "The count of buffered records that are polled " +
            "from consumer and not yet processed for this active task";
        when(streamsMetrics.taskLevelTagMap(THREAD_ID, TASK_ID)).thenReturn(tagMap);

        try (final MockedStatic<StreamsMetricsImpl> streamsMetricsStaticMock = mockStatic(StreamsMetricsImpl.class)) {
            final Sensor sensor = TaskMetrics.activeBufferedRecordsSensor(THREAD_ID, TASK_ID, streamsMetrics);
            streamsMetricsStaticMock.verify(
                () -> StreamsMetricsImpl.addValueMetricToSensor(
                    expectedSensor,
                    TASK_LEVEL_GROUP,
                    tagMap,
                    operation,
                    countDescription
                )
            );
            assertThat(sensor, is(expectedSensor));
        }
    }