当有多个生产者时，如何在 kafka 中一次性交付答案

【问题标题】：How to exactly-once delivery in kafka When there are multiple producers当有多个生产者时，如何在 kafka 中一次性交付
【发布时间】：2021-06-02 02:33:34
【问题描述】：

我正在使用 syslog-ng（最新版本 3.30）和 kafka(2.2)

syslog-ng 和 kafka 在 3 个节点上运行
客户端将日志发送到多个 syslog-ng 节点
syslog-ng 节点将这些日志发送到 kafka 代理
对于 kafka syslog-ng 是生产者

我面临的问题是多个充当生产者的 syslog-ng 正在向主题发送相同的数据。在 kafka 主题中，如果一个客户端正在向多个 syslog-ng 发送类似 ABC 的消息，那么 syslog-ng 的所有节点都会收到相同的消息 ABC，并且生产者（多个 syslog-ng 节点）会向 kafka 发送 ABC 消息，因此会有 3 条 ABC 消息


@define kafka-implementation kafka-c

#############################################
source s_network_udp {
     network(
     transport("udp")
     port(514)
);
};

#############################################

rewrite test {
set(
"test",
        value("tags")
);
};

#############################################
destination pfsense_kafka
{
kafka (
bootstrap-servers("xx.xx.xx.xx:9092")
topic("logs")
properties_file("/etc/syslog-ng/syslog_producer.properties")
 message("$(format-json time=$ISODATE tags host=$HOST message='${MSGHDR}${MSG}')")
);
};

log{ source(s_network_udp); rewrite(test); destination(pfsense_kafka);};

syslog_producer.properties

acks=all
#transactional.id=pfsenseProducer
#transaction.state.log.replication.factor=1
enable.idempotence=true
#retries=10000000
message.send.max.retries=10000000
max.in.flight.requests.per.connection=1

我已经阅读了关于幂等生产者和设置 transaction.id 将提供一次交付，但对我没有任何作用

当我启用enable.idempotence=true,message.send.max.retries=10000000,max.in.flight.requests.per.connection=1 我收到了 syslog-ng 发送的消息，但是当我启用“transaction.id”时，我没有收到任何消息

【问题讨论】：

标签： apache-kafka syslog-ng

【解决方案1】：

让我们从幂等生产者开始。通过设置enable.idempotence=true，你就设置好了。 transaction.id 不是必需的，实际上你不应该设置它。这是个好消息。

坏消息是，幂等生产者确实为您提供了一次交付。但这意味着，如果您提供消息 A，则该消息将传递给 kafka 一次，而不是更少或更多。

如果您有两个生产者（例如两个 syslog-ng）并且它们都通过生产者向 kafka 发送消息，则这些消息（即使内容相同）不是同一条消息（消息 A 和 B，而是内容为 C)。结果是两者都发送消息 A 和 B，并且 kafka 主题看起来像 C、C。这完全不是你想要的。但是，

准确交付一个 C 需要应用程序级逻辑。

【讨论】：