private StreamsMetricsImpl setupMetrics(final StreamsConfig streamsConfig) {
        final String threadId = Thread.currentThread().getName();

        final MetricConfig metricConfig = new MetricConfig()
            .samples(streamsConfig.getInt(StreamsConfig.METRICS_NUM_SAMPLES_CONFIG))
            .recordLevel(Sensor.RecordingLevel.forName(streamsConfig.getString(StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG)))
            .timeWindow(streamsConfig.getLong(StreamsConfig.METRICS_SAMPLE_WINDOW_MS_CONFIG), TimeUnit.MILLISECONDS);
        metrics = new Metrics(metricConfig, mockWallClockTime);

        final StreamsMetricsImpl streamsMetrics = new StreamsMetricsImpl(
            metrics,
            "test-client",
            streamsConfig.getString(StreamsConfig.BUILT_IN_METRICS_VERSION_CONFIG),
            mockWallClockTime
        );
        TaskMetrics.droppedRecordsSensor(threadId, TASK_ID.toString(), streamsMetrics);

        return streamsMetrics;
    }
Create a new test diver instance.

@param builder builder for the topology to be tested
@param config the configuration for the topology
@param initialWallClockTimeMs the initial value of internally mocked wall-clock time
streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java