Кафка: конфигурация семантики ровно один раз с использованием Apache Beam

Question

Я пытаюсь настроить один раз семантику в Kafka (Apache Beam).Вот изменения, которые я собираюсь представить:

Производитель:

1. enable.idenpotence = true

2. transactional.id = uniqueTransactionalId

Потребитель:

1. set enable.auto.commit = false

   // добавлено следующее для потребителя:

2. .commitOffsetsInFinalize()
3. .withReadCommitted()

Добавлено следующее к KafkaIO#write builder:

1. .withEOS(numShards, sinkGroupId)

Кто-нибудь знает, что еще нужно изменить, чтобы достичь ровно семантики в Apache Beam KafkaIO?

Вышеприведенная конфигурация выглядит хорошо, или я что-то неправильно понял?

Нужно ли указывать свойство transactional.id, если я не использую API транзакций (потому что у меня нет явного производителя в apache beam)?

Answer 1

Ну, похоже, я наконец-то нашел правильные настройки, которые соответствуют моим требованиям.Вот что я закончил:

1) KafkaIO.Read:

• обновить свойства потребителя с помощью enable.auto.commit = false
• .withReadCommitted()
• .commitOffsetsInFinalize()

2) KafkaIO#write:

• .withEOS(numShards, sinkGroupId)

  Это также включит идемпотентность и установит transactional.id под капотом для производителя.

Так что в результате с такими настройками у нас будет семантика * хотя бы раз при чтении и семантика при записи.