class StreamsKafkaClient {

    public MetadataResponse fetchMetadata() {
        final ClientRequest clientRequest = kafkaClient.newClientRequest(
            getAnyReadyBrokerId(),
            MetadataRequest.Builder.allTopics(),
            Time.SYSTEM.milliseconds(),
            true);
        final ClientResponse clientResponse = sendRequestSync(clientRequest);

        if (!clientResponse.hasResponse()) {
            throw new StreamsException("Empty response for client request.");
        }
        if (!(clientResponse.responseBody() instanceof MetadataResponse)) {
            throw new StreamsException("Inconsistent response type for internal topic metadata request. " +
                "Expected MetadataResponse but received " + clientResponse.responseBody().getClass().getName());
        }
<<<<<<< HEAD
        final MetadataResponse metadataResponse = (MetadataResponse) clientResponse.responseBody();
        return metadataResponse;
    }

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

}