1

我想使用 kafka-console-producer cli 工具使用事务客户端将消息发送到主题。我认为我的配置正确，但是当我运行 cli 工具时，出现以下异常：

java.lang.IllegalStateException: Cannot perform a 'send' before completing a call to initTransactions when transactions are enabled.

因此，我认为实例化 kafka.tools.ConsoleProducer 类的 kafka-console-producer 似乎没有调用 initTransactions，或者在调用完成之前发送。

我可以做些什么来让 kafka-console-producer cli 使用事务发送到主题？

这是我用来运行 kafka-console-producuer 的命令行：

export KAFKA_OPTS="-Djava.security.auth.login.config=/tmp/kafka/jaas.conf";echo "test message" | kafka-console-producer --broker-list wildthing-2.gce.cloudera.com:9092 --topic ${topic}  --producer.config /tmp/kafka/client.properties

我正在使用 Kafka 版本 2.2.1。这是结果输出中的生产者配置：

21/04/01 14:58:41 INFO producer.ProducerConfig: ProducerConfig values: 
        acks = all
        batch.size = 16384
        bootstrap.servers = [wildthing-2.gce.cloudera.com:9092]
        buffer.memory = 33554432
        client.dns.lookup = default
        client.id = console-producer
        compression.type = none
        connections.max.idle.ms = 540000
        delivery.timeout.ms = 120000
        enable.idempotence = false
        interceptor.classes = []
        key.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
        linger.ms = 1000
        max.block.ms = 60000
        max.in.flight.requests.per.connection = 5
        max.request.size = 1048576
        metadata.max.age.ms = 300000
        metric.reporters = []
        metrics.num.samples = 2
        metrics.recording.level = INFO
        metrics.sample.window.ms = 30000
        partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
        receive.buffer.bytes = 32768
        reconnect.backoff.max.ms = 1000
        reconnect.backoff.ms = 50
        request.timeout.ms = 1500
        retries = 3
        retry.backoff.ms = 100
        sasl.client.callback.handler.class = null
        sasl.jaas.config = null
        sasl.kerberos.kinit.cmd = /usr/bin/kinit
        sasl.kerberos.min.time.before.relogin = 60000
        sasl.kerberos.service.name = kafka
        sasl.kerberos.ticket.renew.jitter = 0.05
        sasl.kerberos.ticket.renew.window.factor = 0.8
        sasl.login.callback.handler.class = null
        sasl.login.class = null
        sasl.login.refresh.buffer.seconds = 300
        sasl.login.refresh.min.period.seconds = 60
        sasl.login.refresh.window.factor = 0.8
        sasl.login.refresh.window.jitter = 0.05
        sasl.mechanism = GSSAPI
        security.protocol = SASL_PLAINTEXT
        send.buffer.bytes = 102400
        ssl.cipher.suites = null
        ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
        ssl.endpoint.identification.algorithm = null
        ssl.key.password = null
        ssl.keymanager.algorithm = SunX509
        ssl.keystore.location = null
        ssl.keystore.password = null
        ssl.keystore.type = JKS
        ssl.protocol = TLS
        ssl.provider = null
        ssl.secure.random.implementation = null
        ssl.trustmanager.algorithm = PKIX
        ssl.truststore.location = null
        ssl.truststore.password = null
        ssl.truststore.type = JKS
        transaction.timeout.ms = 60000
        transactional.id = 123
        value.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
4

2 回答 2

1

如果您正在寻找一个 cli 工具，kcat应该提供以事务方式生成消息的能力。在编写新的 Java 生产者的情况下，我们在 Nussknacker 中遇到了类似的问题，您可以在其中检查如何管理 Kafka 生产者事务以及如何设置生产者选项。

于 2022-02-17T10:29:11.167 回答
0

@Ben，你是对的。cli kafka-console-producer 调用 scala 类 kafka.tools.ConsoleProducer。据我了解，只有新的 Java 生产者和消费者支持完全一次语义。有关详细信息，请参阅此文档。所以直接回答你的问题 - 不，我认为你不能做一些事情来让开箱即用的 cli kafka-console-producer 以事务方式发送消息。但是，您可以使用新的 Java 生产者 API 编写自己的生产者，并从 shell 脚本中调用它。

于 2021-04-02T01:54:54.530 回答