AWS MSK(kafka)生产者事务提交超时

Question

我一直在尝试使用 python

confluent-kafka

库将消息发送到 AWS MSK。 我想确保每条消息的单一传递，这就是我使用基于事务的生产者的原因。我目前每笔交易发送 50 万条消息。

交易的发送部分工作正常并为我们提供了所需的吞吐量，但是，当我提交交易时，一些交易随机超时。

在正常流程中，当问题没有发生时，提交事务部分不需要时间（几秒）。但是，我在提交事务上添加了 10 分钟的超时，但其中一些仍然超时。

这是我正在使用的代码：-

connection_config={
"bootstrap.servers": server-url,
"security.protocol": "SASL_SSL",
"sasl.username": "test",
"sasl.password": "test",
"sasl.mechanism": "SCRAM-SHA-512",
"enable.idempotence": "True",
"transaction.timeout.ms": 1200000,
"acks": "all",
"queue.buffering.max.messages": 200,
"retries": 50
}
p = Producer(connection_config)
p.init_transactions()
p.begin_transaction()
logging.info("Connection successful, writing messages..")
  for index, record in enumerate(data):
      try:
          p.produce(topic_name, json.dumps(record).encode('utf-8'), callback=receipt)
          p.poll(0)
      except BufferError as e:
          p.flush()
          p.produce(topic_name, json.dumps(record).encode('utf-8'), callback=receipt)
logging.info("Flushing remaining messages to kafka ")
p.flush()
logging.info(f"Sending complete for producer,commiting transaction")
p.commit_transaction(int(producer_timeout))

这是我用于 MSK（kafka）的配置：-

auto.create.topics.enable=true
default.replication.factor=2
min.insync.replicas=2
num.io.threads=8
num.network.threads=5
num.partitions=50
num.replica.fetchers=2
replica.lag.time.max.ms=30000
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
socket.send.buffer.bytes=102400
unclean.leader.election.enable=true
zookeeper.session.timeout.ms=18000
offsets.topic.replication.factor=2
transaction.state.log.replication.factor=2
transaction.max.timeout.ms=1200000
num.network.threads=10

超时错误：-

cimpl.KafkaException: KafkaError{code=_TIMED_OUT,val=-185,str="Transactional API operation (commit_transaction) timed out"}

我试过查看服务器日志，但找不到与发生这种情况的原因相关的任何信息。有人可以帮助调试这个问题吗？非常感谢。

我曾尝试减少每笔交易的消息数量，并且可以看到，如果消息较少，失败率会提高，但是我在网上发现的数量是每笔交易的消息越多，效果越好。