[jira] [Created] (FLINK-20753) Duplicates With Exactly-once Kafka -> Kakfa Producer

[Shang Yuanchun (Jira)]

Nazar Volynets created FLINK-20753:
--------------------------------------

             Summary: Duplicates With Exactly-once Kafka -> Kakfa Producer
                 Key: FLINK-20753
                 URL: https://issues.apache.org/jira/browse/FLINK-20753
             Project: Flink
          Issue Type: Bug
          Components: Connectors / Kafka, Runtime / Checkpointing
    Affects Versions: 1.12.0
         Environment: Java 11
Flink stated within IDE
            Reporter: Nazar Volynets


*Introduction*

Based on as follows statements from Flink's docs:

1. [https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/kafka.html]
{quote}Flink provides an [Apache Kafka|https://kafka.apache.org/] connector for reading data from and writing data to Kafka topics with exactly-once guarantees.
{quote}
2. [https://ci.apache.org/projects/flink/flink-docs-release-1.11/learn-flink/fault_tolerance.html#exactly-once-end-to-end]
{quote}To achieve exactly once end-to-end, so that every event from the sources affects the sinks exactly once, the following must be true:
 # your sources must be replayable, and
 # your sinks must be transactional (or idempotent){quote}
3. [https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/kafka.html#caveats]
{quote}{{Semantic.EXACTLY_ONCE}} mode relies on the ability to commit transactions that were started before taking a checkpoint, after recovering from the said checkpoint. If the time between Flink application crash and completed restart is larger than Kafka's transaction timeout there will be data loss (Kafka will automatically abort transactions that exceeded timeout time)
{quote}
4. [https://issues.apache.org/jira/browse/FLINK-7210]
There is references/mentions about two-phase commit mechanic used in old Flink Kafka connector. So it is expected that latest one version of connector has the same functionality.

it is indirectly expectation of EXACTLY_ONCE Kafka->Kafka end-to-end delivery guarantees.

Moreover it is emphasised to tune Kafka cluster transaction timeout (make it from 15 mins to 1 hour) to omit data loss.

Moving forward, all these three statements are met by `Kafka Source` -> `Kafka Sink` app:
 * regarding first-one -> you are reading from & to Kafka
 * about second-one -> `Kafka Source` is replayable & `Kafka Sink` is transactional
 * last one -> `Kafka Sink` is transactional & consequently in case of EXACTLY_ONCE this operator has a state; so it expected that transaction will be rolled back.

But in fact there is no possibility to achieve EXACTLY_ONCE for simple Flink `Kafka Source` -> `Kafka Sink` application. Duplicates still exists as result EXACTLY_ONCE semantics is violated.

*Details*

+STRs:+
 # Create simple Flink's `Kafka Source` -> `Kafka Sink` app
 ## Stream execution env:
 ### Parallelism -> 1
 ### Enable checkpointing -> 10000 ms (do it so big intentionally)
 ### State backend -> RocksDB
 ### Checkpointing mode -> EXACTLY_ONCE
 ### Min pause between checkpoints -> 500 ms
 ### Max concurrent checkpoints -> 1
 ## Flink Kafka consumer
 ### Nothing valuable
 ## Flink Kafka producer
 ### Props:
 #### ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true"
 #### ProducerConfig.ACKS_CONFIG, "all"
 #### ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1"
 ### EXACTLY_ONCE Semantic
 # Deploy `Kafka Source` Cluster
 ## Cretae `topic-1` with 3 patitions
 # Deploy `Kafka Sink` Cluster
 ## Cretae `topic-1` with 3 patitions
 # Spin up some Kafka client to generate data into `Kafka Source`:`topic-1` (e.g. Confluent `kafka-console-producer`)
 # Spin up +transactional+ Kafka consumer to drain data from `Kafka Sink`:`topic-1` (e.g. Confluent `kafka-console-consumer`)
 # Use Flink's app described in step #1 to ship data from `Kafka Source` -> `Kafka Sink` Kafka cluster.
 # Wait until Flink app will create a first checkpoint. 
 # Brutally kill Flink's app (SIGKILL)
 # Wait 10 secs
 # Start Flink app again.
 # Check on duplications in +transactional+ Kafka consumer (described in step #5)

+Actual+

Duplication are exist in +transactional+ Kafka consumer output.

+Expected+
 * Kafka transaction should be rolled back by Flink Kafka producer with EXACTLY_ONCE Semantic
 * Flink should automatically replay the data from `Kafka Source` based on offsets persisted in latest checkpoint

*Example*

+App+
{code:java|title=build.gradle (dependencies)}
...
ext {
  ...
  javaVersion = '11'
  flinkVersion = '1.12.0'
  scalaBinaryVersion = '2.11'
  ...
}

dependencies {
  ...
  implementation "org.apache.flink:flink-streaming-java_${scalaBinaryVersion}:${flinkVersion}"
  implementation "org.apache.flink:flink-clients_${scalaBinaryVersion}:${flinkVersion}"
  implementation "org.apache.flink:flink-statebackend-rocksdb_${scalaBinaryVersion}:${flinkVersion}"
  ...
}
{code}
{code:java|title=App}
public static void main(String[] args) {
  ...
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

  env.setParallelism(1); // to make things simple

  env.enableCheckpointing(10000); // intentionally specified 10 secs to have a room to stop app between checkpoints

  env.setStateBackend(new RocksDBStateBackend("file:///xxx/config/checkpoints/rocksdb", true));

  env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
  env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
  env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
  env.getCheckpointConfig().setPreferCheckpointForRecovery(true);

  FlinkKafkaConsumer<Record> consumer = createConsumer();
  FlinkKafkaProducer<Record> producer = createProducer();

  env
    .addSource(consumer)
    .uid("kafka-consumer")
    .addSink(producer)
    .uid("kafka-producer")
  ;

  env.execute();
}

public static FlinkKafkaConsumer<Record> createConsumer() {
  ...
  Properties props = new Properties();
  props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-source-1:9091");
  ... // nothing special
  props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");

  FlinkKafkaConsumer<Record> consumer = new FlinkKafkaConsumer<>("topic-1", new RecordKafkaDerSchema(), props);

  ... // RecordKafkaDerSchema --> custom schema is used to copy not only message body but message key too
  ... // SimpleStringSchema --> can be used instead to reproduce issue

  consumer.setStartFromGroupOffsets();
  consumer.setCommitOffsetsOnCheckpoints(true);

  return consumer;
}

public static FlinkKafkaProducer<Record> createProducer() {
  ...
  Properties props = new Properties();
  ...
  props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-target-1:9094");
  props.setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1");
  props.setProperty(ProducerConfig.ACKS_CONFIG, "all");
  props.setProperty(ProducerConfig.BATCH_SIZE_CONFIG, "16384");
  props.setProperty(ProducerConfig.LINGER_MS_CONFIG, "1000");
  props.setProperty(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, "2000");
  props.setProperty(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, "9000");
  props.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
  props.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "xxx"); // ignored due to expected behaviour - https://issues.apache.org/jira/browse/FLINK-17691
  props.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "" + (15 * 60 * 1000)); // decreased from 1 hour to 15 mins; app is going to be shutdown less than 15 mins
  ...
  FlinkKafkaProducer<Record> producer = new FlinkKafkaProducer<>("topic-1", new RecordKafkaSerSchema(true), props, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);

  ... // RecordKafkaSerSchema --> custom schema is used to copy not only message body but message key too
  ... // SimpleStringSchema --> can be used instead to reproduce issue

  return producer;
}
{code}
{code:bash|title=kafka-source-1 Producer}
bash -c 'echo Producing data... && \
 for ((i=0; ;++i)); do echo "t1-k-$$i:t1-v-$$i"; sleep 2; done | kafka-console-producer --request-required-acks 1 --broker-list kafka-source-1:9091 --topic topic-1 --property parse.key=true --property key.separator=":"'
{code}
{code:bash|title=kafka-target-1 Consumer - 0 partition}
bash -c 'echo Consuming data for topic-1... && \
      kafka-console-consumer --bootstrap-server kafka-target-1:9094 --topic topic-1 --partition 0 --from-beginning --property print.key=true --property key.separator=":" --value-deserializer org.apache.kafka.common.serialization.StringDeserializer' --isolation-level read_committed
{code}
{code:bash|title=kafka-target-1 Consumer - 1 partition}
bash -c 'echo Consuming data for topic-1... && \
      kafka-console-consumer --bootstrap-server kafka-target-1:9094 --topic topic-1 --partition 1 --from-beginning --property print.key=true --property key.separator=":" --value-deserializer org.apache.kafka.common.serialization.StringDeserializer' --isolation-level read_committed
{code}
{code:bash|title=kafka-target-1 Consumer - 2 partition}
bash -c 'echo Consuming data for topic-1... && \
      kafka-console-consumer --bootstrap-server kafka-target-1:9094 --topic topic-1 --partition 2 --from-beginning --property print.key=true --property key.separator=":" --value-deserializer org.apache.kafka.common.serialization.StringDeserializer' --isolation-level read_committed
{code}
+Output+
{code:java|title=kafka-target-1 Consumer - 0 partition}
...
t1-k-40:t1-v-40
t1-k-43:t1-v-43
t1-k-44:t1-v-44
t1-k-47:t1-v-47
t1-k-48:t1-v-48
t1-k-49:t1-v-49
t1-k-48:t1-v-48 // DUPLICATION!!! --> EXACTLY ONCE is violated
t1-k-49:t1-v-49 // DUPLICATION!!! --> EXACTLY ONCE is violated
t1-k-54:t1-v-54
t1-k-61:t1-v-61
t1-k-62:t1-v-62
t1-k-66:t1-v-66
t1-k-71:t1-v-71
t1-k-73:t1-v-73
...
{code}
{code:java|title=kafka-target-1 Consumer - 1 partition}
...
t1-k-35:t1-v-35
t1-k-46:t1-v-46
t1-k-50:t1-v-50
t1-k-51:t1-v-51
t1-k-53:t1-v-53
t1-k-56:t1-v-56
t1-k-57:t1-v-57
t1-k-59:t1-v-59
t1-k-60:t1-v-60
t1-k-63:t1-v-63
t1-k-65:t1-v-65
t1-k-69:t1-v-69
t1-k-74:t1-v-74
...
{code}
{code:java|title=kafka-target-1 Consumer - 2 partition}
...
t1-k-39:t1-v-39
t1-k-41:t1-v-41
t1-k-42:t1-v-42
t1-k-45:t1-v-45
t1-k-52:t1-v-52
t1-k-55:t1-v-55
t1-k-58:t1-v-58
t1-k-64:t1-v-64
t1-k-67:t1-v-67
t1-k-68:t1-v-68
t1-k-70:t1-v-70
t1-k-72:t1-v-72
t1-k-75:t1-v-75
t1-k-77:t1-v-77
...
{code}
+Summary+

As we can see from `kafka-target-1 Consumer - 0 partition` EXACTLY ONCE delivery has been violated.

P.S.: If I have missed something. Please let me know what & how achieve EXACTLY ONCE delivery in native way (via Flink configuration) for this particular simple application described above.

P.S.: If it is not possible to do in native way (only manual/custom implementation) then please let me know.

P.S.: Similar issue discussions:
* [https://stackoverflow.com/questions/57308590/exactly-once-semantics-in-flink-kafka-producer]



--
This message was sent by Atlassian Jira
(v8.3.4#803005)