[GitHub] flink pull request #6387: [FLINK-9846] [table] Add a Kafka table sink factor...

[pnowojski]

Github user pnowojski commented on a diff in the pull request:

    https://github.com/apache/flink/pull/6387#discussion_r204346197
  
    --- Diff: 
flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSink.java
 ---
    @@ -84,7 +93,14 @@ public Kafka08JsonTableSink(String topic, Properties 
properties, KafkaPartitione
     
        @Override
        protected FlinkKafkaProducerBase<Row> createKafkaProducer(String topic, 
Properties properties, SerializationSchema<Row> serializationSchema, 
FlinkKafkaPartitioner<Row> partitioner) {
    -           return new FlinkKafkaProducer08<>(topic, serializationSchema, 
properties, partitioner);
    +           final FlinkKafkaProducerBase<Row> kafkaProducer = new 
FlinkKafkaProducer08<>(
    +                   topic,
    +                   serializationSchema,
    +                   properties,
    +                   partitioner);
    +           // always enable flush on checkpoint to achieve at-least-once 
if query runs with checkpointing enabled.
    --- End diff --
    
    ditto


---