class StreamsKafkaClient {

    public void checkBrokerCompatibility(final boolean eosEnabled) throws StreamsException {
        final ClientRequest clientRequest = kafkaClient.newClientRequest(
            getAnyReadyBrokerId(),
            new ApiVersionsRequest.Builder(),
            Time.SYSTEM.milliseconds(),
            true);

        final ClientResponse clientResponse = sendRequestSync(clientRequest);
        if (!clientResponse.hasResponse()) {
            throw new StreamsException("Empty response for client request.");
        }
        if (!(clientResponse.responseBody() instanceof ApiVersionsResponse)) {
            throw new StreamsException("Inconsistent response type for API versions request. " +
                "Expected ApiVersionsResponse but received " + clientResponse.responseBody().getClass().getName());
        }

        final ApiVersionsResponse apiVersionsResponse =  (ApiVersionsResponse) clientResponse.responseBody();

        if (apiVersionsResponse.apiVersion(ApiKeys.CREATE_TOPICS.id) == null) {
            throw new StreamsException("Kafka Streams requires broker version 0.10.1.x or higher.");
        }

        if (eosEnabled && !brokerSupportsTransactions(apiVersionsResponse)) {
            throw new StreamsException("Setting " + PROCESSING_GUARANTEE_CONFIG + "=" + EXACTLY_ONCE + " requires broker version 0.11.0.x or higher.");
        }
    }

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

}