public static void addThreadBlockedTimeMetric(final String threadId,
                                                  final StreamThreadTotalBlockedTime blockedTime,
                                                  final StreamsMetricsImpl streamsMetrics) {
        streamsMetrics.addThreadLevelMutableMetric(
            BLOCKED_TIME,
            BLOCKED_TIME_DESCRIPTION,
            threadId,
            (config, now) -> blockedTime.compute()
        );
    }
@Test
    public void shouldAddTotalBlockedTimeMetric() {
        // Given:
        final double startTime = 123.45;
        final StreamThreadTotalBlockedTime blockedTime = mock(StreamThreadTotalBlockedTime.class);
        when(blockedTime.compute()).thenReturn(startTime);

        // When:
        ThreadMetrics.addThreadBlockedTimeMetric(
            "burger",
            blockedTime,
            streamsMetrics
        );

        // Then:
        final ArgumentCaptor<Gauge<Double>> captor = gaugeCaptor();
        verify(streamsMetrics).addThreadLevelMutableMetric(
            eq("blocked-time-ns-total"),
            eq("The total time the thread spent blocked on kafka in nanoseconds"),
            eq("burger"),
            captor.capture()
        );
        assertThat(captor.getValue().value(null, 678L), is(startTime));
    }