public KafkaConsumer<byte[], byte[]> createConsumer(Map<String, Object> consumerProps) {
        Map<String, Object> props = new HashMap<>(clientConfigs);
        props.putAll(consumerProps);

        props.putIfAbsent(GROUP_ID_CONFIG, UUID.randomUUID().toString());
        props.putIfAbsent(BOOTSTRAP_SERVERS_CONFIG, bootstrapServers());
        props.putIfAbsent(ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.putIfAbsent(AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.putIfAbsent(KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        props.putIfAbsent(VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        if (sslEnabled()) {
            props.putIfAbsent(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, brokerConfig.get(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG));
            props.putIfAbsent(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, brokerConfig.get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG));
            props.putIfAbsent(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
        }
        KafkaConsumer<byte[], byte[]> consumer;
        try {
            consumer = new KafkaConsumer<>(props);
        } catch (Throwable t) {
            throw new ConnectException("Failed to create consumer", t);
        }
        return consumer;
    }
List the latest current offsets for the given {@link Collection} of {@link TopicPartition topic partitions}
@param maxDurationMs the max duration to wait for while fetching metadata from Kafka (in milliseconds)
@param admin the admin client to use for fetching metadata from the Kafka cluster
@param topicPartitions the topic partitions to list end offsets for
@return a {@link Map} containing the latest offset for each requested {@link TopicPartition topic partition}; never null, and never empty
connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedKafkaCluster.java