public static Sensor producedSensor(final String threadId,
                                        final String taskId,
                                        final String processorNodeId,
                                        final String topic,
                                        final StreamsMetricsImpl streamsMetrics) {
        final Sensor sensor = streamsMetrics.topicLevelSensor(
            threadId,
            taskId,
            processorNodeId,
            topic,
            PRODUCED,
            RecordingLevel.INFO);
        addTotalCountAndSumMetricsToSensor(
            sensor,
            TOPIC_LEVEL_GROUP,
            streamsMetrics.topicLevelTagMap(threadId, taskId, processorNodeId, topic),
            RECORDS_PRODUCED,
            BYTES_PRODUCED,
            RECORDS_PRODUCED_TOTAL_DESCRIPTION,
            BYTES_PRODUCED_TOTAL_DESCRIPTION
        );
        return sensor;
    }
@Test
    public void shouldGetRecordsAndBytesProducedSensor() {
        final String recordsMetricNamePrefix = "records-produced";
        final String bytesMetricNamePrefix = "bytes-produced";
        final String descriptionOfRecordsTotal = "The total number of records produced to this topic";
        final String descriptionOfBytesTotal = "The total number of bytes produced to this topic";

        when(streamsMetrics.topicLevelSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, TOPIC, "produced", RecordingLevel.INFO))
            .thenReturn(expectedSensor);
        when(streamsMetrics.topicLevelSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, TOPIC, "produced", RecordingLevel.INFO))
            .thenReturn(expectedSensor);
        when(streamsMetrics.topicLevelTagMap(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, TOPIC)).thenReturn(tagMap);

        verifySensor(() -> TopicMetrics.producedSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, TOPIC, streamsMetrics));

        STREAMS_METRICS_STATIC_MOCK.verify(
            () -> StreamsMetricsImpl.addTotalCountAndSumMetricsToSensor(
                expectedSensor,
                TOPIC_LEVEL_GROUP,
                tagMap,
                recordsMetricNamePrefix,
                bytesMetricNamePrefix,
                descriptionOfRecordsTotal,
                descriptionOfBytesTotal
            )
        );
    }