0

我已经配置了一个带有 transactionIdPrefix 的 Kafka ProducerFactory，以便使用@Transactional（请参阅Spring 文档关于仅生产者事务）启用事务同步。

我在集成测试中运行 EmbeddedKafka，以查看它的行为方式。

日志显示以下内容：

DEBUG 8384 --- [ad | producer-1] o.a.k.clients.producer.internals.Sender  :
    [Producer clientId=producer-1, transactionalId=tx-0-0]
    Sending transactional request (type=FindCoordinatorRequest, coordinatorKey=tx-0-0, coordinatorType=TRANSACTION) to node 127.0.0.1:61445 (id: -1 rack: null)

DEBUG 8384 --- [ad | producer-1] o.a.k.c.p.internals.TransactionManager   :
    [Producer clientId=producer-1, transactionalId=tx-0-0]
    Enqueuing transactional request (type=FindCoordinatorRequest, coordinatorKey=tx-0-0, coordinatorType=TRANSACTION)

Timeout expired while initializing transactional state in 60000ms.

这是在 DefaultKafkaProducerFactory 执行时抛出的newProducer.initTransactions()。

我的配置如下：

集成测试

@EmbeddedKafka(brokerProperties = { "transaction.state.log.replication.min.isr=1", "transaction.state.log.replication.factor=1" })

生产者配置

@Bean
  public ProducerFactory<String, String> transactionalProducerFactory() {
    Map<String, Object> configuration = new HashMap<>();

    configuration.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());
    configuration.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configuration.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    String transactionIdPrefix = "tx-0-";
    configuration.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
    DefaultKafkaProducerFactory<String, String> factory = new DefaultKafkaProducerFactory<>(configuration);
    factory.setTransactionIdPrefix(transactionIdPrefix);

    return factory;
  }

@Bean
  public KafkaTemplate<String, String> transactionalKafka() {
    return new KafkaTemplate<>(transactionalProducerFactory());
  }

Spring-Kafka 版本：2.2.7.RELEASE

我不知道如何继续前进，我认为我遵循了文档中的每一步，并且在事务初始化期间 Kafka 客户端和代理之间的通信应该没问题。谁能帮我解决这个问题？

4

1 回答 1

1

多亏了嵌入式 kafka 服务器日志，我可以解决这个问题。

属性transaction.state.log.min.isr默认为 2，我必须用它覆盖它transaction.state.log.min.isr = 1以修复服务器错误。之后我的集成测试通过了。

于 2021-08-23T17:34:53.933 回答