相关疑难解决方法(0)

最近，我们开始对 Kafka 连接器与 MSK（Amazon 托管的 Kafka 服务）进行一些测试。发布记录似乎工作正常，但在启用事务时则不然。

我们的集群由 2 个使用默认 MSK 配置的代理（因为我们有 2 个区域）组成。我们正在使用以下属性创建 Java Kafka 生产者：

bootstrap.servers=x.us-east-1.amazonaws.com:9094,y.us-east-1.amazonaws.com:9094
client.id=kafkautil
max.block.ms=5000
request.timeout.ms=5000
security.protocol=SSL
transactional.id=transactions 

但是，当生产者以transactional.id启用事务的设置启动时，该initTransactions()方法会挂起：

producer = new KafkaProducer<Object, Object>(kafkaProperties);
if (kafkaProperties.containsKey(ProducerConfig.TRANSACTIONAL_ID_CONFIG)) {
    // this hangs
    producer.initTransactions();
}

查看日志输出，我们看到以下流，并且似乎从未超时。

TransactionManager - Enqueuing transactional request (type=FindCoordinatorRequest,
    coordinatorKey=y, coordinatorType=TRANSACTION)
TransactionManager - Request (type=FindCoordinatorRequest, coordinatorKey=y,
    coordinatorType=TRANSACTION) dequeued for sending
NetworkClient - Found least loaded node z:9094 (id: -2 rack: null) connected with no
    in-flight requests
Sender - Sending transactional …