class StreamsKafkaClient {

    public static StreamsKafkaClient create(final Config streamsConfig) {
        final Time time = new SystemTime();

        final Map<String, String> metricTags = new LinkedHashMap<>();
        final String clientId = streamsConfig.getString(StreamsConfig.CLIENT_ID_CONFIG);
        metricTags.put("client-id", clientId);

        final Metadata metadata = new Metadata(streamsConfig.getLong(StreamsConfig.RETRY_BACKOFF_MS_CONFIG),
                                               streamsConfig.getLong(StreamsConfig.METADATA_MAX_AGE_CONFIG),
                                               false);
        final List<InetSocketAddress> addresses = ClientUtils.parseAndValidateAddresses(streamsConfig.getList(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG));
        metadata.update(Cluster.bootstrap(addresses), Collections.<String>emptySet(), time.milliseconds());

        final MetricConfig metricConfig = new MetricConfig().samples(streamsConfig.getInt(CommonClientConfigs.METRICS_NUM_SAMPLES_CONFIG))
                .timeWindow(streamsConfig.getLong(CommonClientConfigs.METRICS_SAMPLE_WINDOW_MS_CONFIG), TimeUnit.MILLISECONDS)
                .tags(metricTags);
        final List<MetricsReporter> reporters = streamsConfig.getConfiguredInstances(
            ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG,
            MetricsReporter.class);
        // TODO: This should come from the KafkaStream
        reporters.add(new JmxReporter("kafka.admin.client"));
        final Metrics metrics = new Metrics(metricConfig, reporters, time);

        final ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(streamsConfig);
        final LogContext logContext = createLogContext(clientId);

        final Selector selector = new Selector(
                streamsConfig.getLong(StreamsConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG),
                metrics,
                time,
                "kafka-client",
                channelBuilder,
                logContext);

        final KafkaClient kafkaClient = new NetworkClient(
                selector,
                metadata,
                clientId,
                MAX_INFLIGHT_REQUESTS, // a fixed large enough value will suffice
                streamsConfig.getLong(StreamsConfig.RECONNECT_BACKOFF_MS_CONFIG),
                streamsConfig.getLong(StreamsConfig.RECONNECT_BACKOFF_MAX_MS_CONFIG),
                streamsConfig.getInt(StreamsConfig.SEND_BUFFER_CONFIG),
                streamsConfig.getInt(StreamsConfig.RECEIVE_BUFFER_CONFIG),
                streamsConfig.getInt(StreamsConfig.REQUEST_TIMEOUT_MS_CONFIG),
                time,
                true,
                new ApiVersions(),
                logContext);
        return new StreamsKafkaClient(streamsConfig, kafkaClient, reporters, logContext);
    }

      StreamsKafkaClient(final Config streamsConfig,
                       final KafkaClient kafkaClient,
                       final List<MetricsReporter> reporters,
                       final LogContext log);

}