@Override
    public String toString() {
        return toString("");
    }
@Test
    public void testMetrics() {
        when(stateManager.taskId()).thenReturn(taskId);
        when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
        task = createStatelessTask(createConfig("100"));

        assertNotNull(getMetric(
            "enforced-processing",
            "%s-rate",
            task.id().toString()
        ));
        assertNotNull(getMetric(
            "enforced-processing",
            "%s-total",
            task.id().toString()
        ));

        assertNotNull(getMetric(
            "record-lateness",
            "%s-avg",
            task.id().toString()
        ));
        assertNotNull(getMetric(
            "record-lateness",
            "%s-max",
            task.id().toString()
        ));

        assertNotNull(getMetric(
            "active-process",
            "%s-ratio",
            task.id().toString()
        ));

        assertNotNull(getMetric(
            "active-buffer",
            "%s-count",
            task.id().toString()
        ));

        testMetricsForBuiltInMetricsVersionLatest();

        final JmxReporter reporter = new JmxReporter();
        final MetricsContext metricsContext = new KafkaMetricsContext("kafka.streams");
        reporter.contextChange(metricsContext);

        metrics.addReporter(reporter);
        final String threadIdTag = THREAD_ID_TAG;
        assertTrue(reporter.containsMbean(String.format(
            "kafka.streams:type=stream-task-metrics,%s=%s,task-id=%s",
            threadIdTag,
            threadId,
            task.id()
        )));
    }