Is it possible to achieve at least once semantics using kafka in flink without using chekpoints?

1
votes

I would like to write a simple Flink application that reads from a Kafka queue and processes the message and stores the output to an external system, with at least once semantics and without using checkpoints. I would like to avoid checkpoints because if the Kafka offsets are checkpointed, then all intermediate state will have to be checkpointed as well. In other words, I want the application to be as stateless as possible.

The way I envision at least once to work is the following:

  1. a source reads from kafka
  2. processing happens
  3. the output is stored to the external system
  4. the message is acknowledged to kafka

Note that:

  • If 2. or 3. fail, and the app restarts, the same message will be processed again (good)
  • If 2. and 3. succeed, 4. fails and the app restarts, we will will have stored the result twice (acceptable)

Based on https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/kafka.html#kafka-consumers-offset-committing-behaviour-configuration, the only way to get at least once (or the stronger exactly once) guarantees is by using checkpoints.

It seems that the core of the issue is that 4. needs to communicate back to 1. to ack to Kafka, which cannot happen in standard Flink, but should be possible using stateful functions.

To put it all together, the question is: Is it possible to achieve at least once semantics using kafka in flink without using chekpoints?

1
apache-kafkaapache-flink

1 Answers

1
votes

According to the documentation you already linked it says:

"Checkpointing disabled: if checkpointing is disabled, the Flink Kafka Consumer relies on the automatic periodic offset committing capability of the internally used Kafka clients. Therefore, to disable or enable offset committing, simply set the enable.auto.commit / auto.commit.interval.ms keys to appropriate values in the provided Properties configuration."

As your goal is to disable checkpointing, you could set

enable.auto.commit=true
auto.commit.interval.ms=??? // use a time high enough such that your steps 2. and 3. are covered.