[Quick Question - Duplicate/Missing records] I have a streaming pipeline as shown in the screenshot. The job uses

EXACTLY_ONCE

checkpointing mode (k8s Flink operator). When I change the job’s state to

suspended

to manually and change back to

running

. It seems that the raw data sink, which does not involve

TumblingEventTimeWindows

+

ProcessWindowFunction

, do not write duplicate records, but the processed data seem to either miss some records or write duplicate records. The two sinks are sharing the same sink. How is this happening?

yes your guess is true, Flink guarantees at-least once semantics by replaying data upon potential failure when restoring from checkpoint, exactly once semantics is achieved via 2PC (2 phase commit)sinks which needs the downstream to handle 2PC as well or have idempotent destinations.

@Ahmed Hamdy Thanks for the reply. To clarify, is there a common way to guarantee exactly once writing of records to S3 bucket using FileSink?

I don't think you can look at the UI of Flink to determine if duplicates have been written. Flink will write Pending and In-Progress files to your buckets too. This is also explained on https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/filesystem/#file-sink

Right, I actually did check the records written in each raw and processed data buckets. The processed data bucket, when restored from the last-state, always missed some records, while the raw data bucket didn’t miss any records. Let me check if

inprogress

and

pending

files have any impacts. This matters because I am using detecting files uploaded to S3 bucket and relay these parquet files to another pipeline.

Yes please do. It could be that we need to clarify the documentation because there are e2e tests that validate this behavior

One quick question though. I ran at least 50~60 tests over the weekend, suspending Flink manually and restoring using last-state. However, raw data bucket never missed any records. It didn’t have duplicates either. If

inprogress

files have an impact here, it only makes sense to me if both raw data bucket and processed data bucket have some missing and duplicate records. How should I understand this behaviour?

Most likely that it’s the operator before that hasn’t finished processing yet.

Right. The operator before the sink step of processed data is WindowProcessFunction, which simply takes in records and outputs after cleaning some values. In this case, if the state is restored from the last successful checkpoint, shouldn’t Flink be replaying the events in the

WindowProcessFunction

, and should we not see duplicate records in the Sink bucket, not missing records?

@Hong Teoh Pinging you since this involves Kinesis and you probably know more about this then me 😅

Wow.! Thanks for the detailed investigation. This does sound like a skipped watermark event causing the buffered window to not close. This agrees with the data that the

Raw Sink

does not have this problem, whereas the

Processed Data sink

which has a window before the sink has this issue. I can think of two places where this error might be (am happy to replicate this issue + help investigate this further, since it seems like quite a nasty failure mode): 1. Kinesis Consumer’s watermarking generation has bug ◦ We can help root cause if this is the issue / mitigate this by using the new watermark generation rather than the one built into the Kinesis Consumer 2. Window Operator has bug when restoring from state ◦ This seems unlikely. I haven’t looked, but I would be surprised if we don’t have tests validating this behavior! Couple of clarifications: • Which Flink version are you using? • To test (1), we can try switching out the Watermarking Strategy on consumer

WatermarkStrategy<> s = WatermarkStrategy
    .<>forMonotonousTimestamps()
    .withTimestampAssigner((event,timestamp) -> {
        return timestamp;
    })
    .withIdleness(Duration.ofSeconds(15));
    
env.addSource(consumer)
    .assignTimestampsAndWatermarks(s)

@Hong Teoh Thanks for the reply! 1. I am using Flink 1.15.1 2. Here is a code snippet that I have tested. Below is the code snippet that I tested with. I am aware of the importance of

withIdleness

in this case, and it is set to 1ms.

env.addSource(
            new FlinkKinesisConsumer<>(
                streamName,
                new EventDeserializationSchema(),
                inputProperties
            )
        )
        .assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<Event>forBoundedOutOfOrderness(
                                Duration.ofSeconds(0))
                        .withIdleness(Duration.ofMillis(1))) // This is to avoid the watermark to be stuck
        .uid("event_watermarking_strategy_" + APP_VERSION)
        .name("event_watermarking_strategy_" + APP_VERSION)
        .keyBy(EventUtils::getDynamicPartitionKey)
        .uid(name)
        .name(name)
        .window(TumblingEventTimeWindows.of(Time.seconds(10)))
        .trigger(EventTimeTrigger.create())
        .process(new PredictionEventProcessWindowFunction(
                inputProperties.get("ServiceUri").toString()))
        .uid("prediction_event_process_window_function_" + APP_VERSION)
        .name("prediction_event_process_window_function_" + APP_VERSION)
        .forBulkFormat(
            new Path(path),
                AvroParquetWriters.forReflectRecord(Event.class))
        .withBucketAssigner(new KeyAndDateBucketAssigner(
                "'year='yyyy'/month='MM'/day='dd'/hour='HH/"))
        .withRollingPolicy(OnCheckpointRollingPolicy.build())
        .withOutputFileConfig(DateAndIdOutputFileConfig.configBuilder()
                .withDateFormat("yyyy-MM-dd-HH-mm-ss")
                .withPartSuffix(".parquet")
                .build())
        .build()
        .uid("sink_to_datalake_processed_zone_" + APP_VERSION)
        .name("sink_to_datalake_processed_zone_" + APP_VERSION)
        .setParallelism(2)
        .uid("set_parallelism_to_1_after_sink_to_datalake_processed_zone_" + APP_VERSION)
        .name("set_parallelism_to_1_after_sink_to_datalake_processed_zone_" + APP_VERSION);;

Ok interesting… so the issue happens with the code snippet above?

@Hong Teoh I could certainly test with

forMonotonousTimestamps

to rule out that the Kinesis data stream does not have a bug. Could you please clarify what the expected output should be? It should just be monotonously increasing timestamps, correct?

Hmm, I don’t think the issue is with the Kinesis Data Stream, but more with the Kinesis Consumer (Flink code). When I talked about the “watermark generation in Kinesis consumer having a bug” I meant if we use

FlinkKinesisConsumer<> consumer = new FlinkKinesisConsumer<>(...);
consumer.setPeriodicWatermarkAssigner(...)

This was the old method of creating watermarks where the Source itself had to generate the watermarks, but more recently it has been abstracted away and sources should use

assignTimestampsAndWatermarks()

(which is what you are doing!) Hmm…

forMonotonousTimestamps

vs

BoundedOutOfOrderness

should not be the root cause, but it would be worth testing if easy. Yes,

forMonotonousTimestamps

just assumes that the records are in order and will advance the watermarks to time T when record with event time T comes in

forBoundedOutOfOrderness

means it will emit watermarks to time

T - duration

when record with event time T comes in… Given you are using Duration(0) it is basically the same as

forMonotonousTimestamps()

😆

@Hong Teoh Makes sense! Thanks a lot for your kind explanation!! I am 100% sure that the EventTime of the events that were streamed in the Kinesis data stream are monotonously increasing. My test went like this. I sent 1 record every 30 seconds, and thus the timestamp must be increasing. After the second record is processed in the raw sink, the first record was processed in the processed sink. At this point, I shut down Flink manually by setting

job.state: suspended

. Then, I went to the S3 buckets of the raw sink and processed sink paths, and verified that the raw sink has 2 parquet files, each containing 1 record. On the other hand, the processed sink path contained a single parquet file with 1 record. While the Flink job was down, I sent 2 more records. Next, I restored the job by setting

job.state: running

. Then, I sent 1 more record to the kinesis data stream. The raw sink correctly received 3 records in total. 2 records that were sent while Flink was suspended, and 1 record that was sent after the state was restored. On the other hand, the processed sink step only received 1 record, which is the latest record sent after the state was restored. So, I did manually verify the monotonously increasing event time. But, I can still reproduce this issue with a very simple example.

I see, so the processed sink missed 3 records, 1 from before it was suspended, and 2 that were sent while the job was down. When you checked the “buffer” in the checkpoint, do you know which operator it belonged to? (S3 sink / window?) Can we try and eliminate the sink as the cause? It shouldn’t be because it exists in both top and bottom branch. But maybe we can use

.print()

to output the events to std out /

map()

with

log

Here is the content of the actual checkpoint file. After these lines, I see the actual record

ÁW˝Torg.apache.flink.api.common.typeutils.base.StringSerializer$StringSerializerSnapshotcorg.apache.flink.streaming.api.windowing.windows.TimeWindow$Serializer$TimeWindowSerializerSnapshot _timer_state/event_window-timersVALUE_SERIALIZER@org.apache.flink.streaming.api.operators.TimerSerializerSnapshot
ÁW˝Torg.apache.flink.api.common.typeutils.base.StringSerializer$StringSerializerSnapshotcorg.apache.flink.streaming.api.windowing.windows.TimeWindow$Serializer$TimeWindowSerializerSnapshothij$1ab4dec8-411f-44c8-9e97-c39617c85d9a(k
…fl°<s3://ggg-staging-flink-state-store/eventlogger-pipeline-v0-1/checkpoints/00000000000000000000000000000000/chk-1362/ab2e447b-1305-44e4-897f-d79fe0c0b44eıTorg.apache.flink.api.common.typeutils.base.StringSerializer$StringSerializerSnapshotwindow-contentsKEYED_STATE_TYPELISTVALUE_SERIALIZERAorg.apache.flink.api.common.typeutils.base.ListSerializerSnapshot>

oo… ok this is window state….

I haven’t set any external state backend, so it should be just an in-memory state backend.

Hm… ok so its not a RocksDB “not firing timers” issue…

For sure. Another context is that there are about 30 shards, and the parallelism I had was 48. I reduced to 24 thinking that maybe idle task slots are causing an issue, but I was able to replicate the same problem

Hey @Kyle Ahn, I tried replicating your issue … but I couldn’t. This is my setup. • Kinesis source with 4 shards • Kinesis -> TumblingWindow.of(1) -> Print sink • Tried with job parallelism of 1, and retried experiment at parallelism of 4 • Flink 1.17-SNAPSHOT 😅 sorry I just used my running cluster • Procedure followed: ◦ Write 4 records into kinesis stream. Id “111” ◦ Take a savepoint. Verified the savepoint has the records with Id “111", not flushed because window has not closed ◦ Stop the job ◦ Write 3 records with ID “555” ◦ Start the job from savepoint. ◦ Write 4 records with id “999" ◦ >>> In ouput, I see records with “111” and “555" ◦ Write more records with id “999” ◦ >>> In output, I see records with “999" printed out I see all the records as expected, indicating the window state is reused as expected. This is my code:

var stream = env.addSource(
                new FlinkKinesisConsumer<>(
                    "<streamname>",
                    new SimpleStringSchema(),
                    inputStreamConfig
                )
            ).assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps())
            .keyBy(String::hashCode)
            .window(TumblingEventTimeWindows.of(Time.minutes(1)))
            .process(new ProcessWindowFunction<String, Object, Integer, TimeWindow>() {
                @Override
                public void process(Integer integer, ProcessWindowFunction<String, Object, Integer, TimeWindow>.Context context, Iterable<String> iterable, Collector<Object> collector) throws Exception {
                    iterable.forEach(collector::collect);
                }
            })
            .print();

Okay, let me try recoding a video with the exact reproducible steps on 1.15.1 today or tomorrow! This may help!

👍

Yeah, try without RocksDB state backend.