presto-kafka/src/main/java/io/prestosql/plugin/kafka/KafkaSimpleConsumerManager.java/KafkaSimpleConsumerManager/KafkaSimpleConsumerManager
class KafkaSimpleConsumerManager:    @Inject
    public KafkaSimpleConsumerManager(
            KafkaConnectorConfig kafkaConnectorConfig,
            NodeManager nodeManager)
    {
        this.nodeManager = requireNonNull(nodeManager, "nodeManager is null");

        requireNonNull(kafkaConnectorConfig, "kafkaConfig is null");
        this.connectTimeoutMillis = toIntExact(kafkaConnectorConfig.getKafkaConnectTimeout().toMillis());
        this.bufferSizeBytes = toIntExact(kafkaConnectorConfig.getKafkaBufferSize().toBytes());

        this.consumerCache = CacheBuilder.newBuilder().build(CacheLoader.from(this::createConsumer));
    }