1

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

我们的集群由使用默认 MSK 配置的 2 个代理（因为我们有 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 request (type=FindCoordinatorRequest, coordinatorKey=y,
    coordinatorType=TRANSACTION) to node z (id: -2 rack: null)
NetworkClient - Sending FIND_COORDINATOR {coordinator_key=y,coordinator_type=1} with
    correlation id 424 to node -2
NetworkClient - Completed receive from node -2 for FIND_COORDINATOR with
    correlation id 424, received {throttle_time_ms=0,error_code=15,error_message=null,
    coordinator={node_id=-1,host=,port=-1}}
TransactionManager LogContext.java:129 - Received transactional response
    FindCoordinatorResponse(throttleTimeMs=0, errorMessage='null', 
    error=COORDINATOR_NOT_AVAILABLE, node=:-1 (id: -1 rack: null)) for request
    (type=FindCoordinatorRequest, coordinatorKey=xxx, coordinatorType=TRANSACTION)

据我所知，经纪人可用，并且该物业中的每个主机bootstrap.servers都可用。如果我连接到它们中的每一个并在没有事务的情况下发布，那么它就可以工作。

知道我们缺少什么吗？

4

1 回答 1

4

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

事实证明，默认 AWS MSK 属性和代理数量存在问题。如果您创建的 Kafka 集群少于 3 个代理，则需要调整以下设置。

应将以下设置（我认为）设置为经纪人的数量：

财产 卡夫卡
默认		 AWS
默认		 描述
default.replication.factor 1 3 自动创建主题的默认复制因子。
min.insync.replicas 1 2 必须确认写入才能被视为成功的最小副本数
offsets.topic.replication.factor 3 3 共享主题偏移量的内部主题。
transaction.state.log.replication.factor 3 3 事务主题的复制因子。

这是关于经纪人属性的 Kafka 文档。

因为我们有 2 个经纪人，所以我们最终得到：

default.replication.factor=2
min.insync.replicas=2
offsets.topic.replication.factor=2
transaction.state.log.replication.factor=2

这似乎解决了这个问题。恕我直言，这是 AWS MSK 和默认配置的真正问题。他们需要自动生成默认配置并根据集群中代理的数量对其进行调整。

于 2021-01-15T22:49:36.070 回答