static Metrics getMetrics(final StreamsConfig config, final Time time, final String clientId) {
        final MetricConfig metricConfig = new MetricConfig()
            .samples(config.getInt(StreamsConfig.METRICS_NUM_SAMPLES_CONFIG))
            .recordLevel(Sensor.RecordingLevel.forName(config.getString(StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG)))
            .timeWindow(config.getLong(StreamsConfig.METRICS_SAMPLE_WINDOW_MS_CONFIG), TimeUnit.MILLISECONDS);
        final List<MetricsReporter> reporters = CommonClientConfigs.metricsReporters(clientId, config);

        final MetricsContext metricsContext = new KafkaMetricsContext(JMX_PREFIX,
                                                                      config.originalsWithPrefix(CommonClientConfigs.METRICS_CONTEXT_PREFIX));
        return new Metrics(metricConfig, reporters, time, metricsContext);
    }
Create a {@code KafkaStreams} instance.
<p>
Note: even if you never call {@link #start()} on a {@code KafkaStreams} instance,
you still must {@link #close()} it to avoid resource leaks.

@param topology       the topology specifying the computational logic
@param applicationConfigs         configs for Kafka Streams
@param time           {@code Time} implementation; cannot be null
@throws StreamsException if any fatal error occurs
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java