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

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

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

  1. enable.idenpotence = верно

  2. transactional.id = uniqueTransactionalId

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

  1. установить enable.auto.commit = false

    // добавили в конструктор потребителей:

  2. .commitOffsetsInFinalize()

  3. .withReadCommitted()

В конструктор KafkaIO#write добавлено следующее:

  1. .withEOS(numShards, sinkGroupId)

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

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

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


apache-beam apache-kafka google-cloud-dataflow apache-beam-io
person Anton Litvinenko    schedule 28.02.2019    source источник

Ответы (1)


arrow_upward
3
arrow_downward

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

1) KafkaIO.Read:

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

2) KafkaIO#write:

  • .withEOS(numShards, sinkGroupId)

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

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

person Anton Litvinenko    schedule 22.03.2019