public void init(final StreamsMetricsImpl streamsMetrics,
                     final TaskId taskId) {
        Objects.requireNonNull(streamsMetrics, "Streams metrics must not be null");
        Objects.requireNonNull(streamsMetrics, "task ID must not be null");
        if (this.taskId != null && !this.taskId.equals(taskId)) {
            throw new IllegalStateException("Metrics recorder is re-initialised with different task: previous task is " +
                this.taskId + " whereas current task is " + taskId + ". This is a bug in Kafka Streams. " +
                "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues");
        }
        if (this.streamsMetrics != null && this.streamsMetrics != streamsMetrics) {
            throw new IllegalStateException("Metrics recorder is re-initialised with different Streams metrics. "
                + "This is a bug in Kafka Streams. " +
                "Please open a bug report under https://issues.apache.org/jira/projects/KAFKA/issues");
        }
        final RocksDBMetricContext metricContext = new RocksDBMetricContext(taskId.toString(), metricsScope, storeName);
        initSensors(streamsMetrics, metricContext);
        initGauges(streamsMetrics, metricContext);
        this.taskId = taskId;
        this.streamsMetrics = streamsMetrics;
    }
@Test
    public void shouldThrowIfMetricRecorderIsReInitialisedWithDifferentTask() {
        assertThrows(
            IllegalStateException.class,
            () -> recorder.init(streamsMetrics, TASK_ID2)
        );
    }