Hey guys, I noticed that Flink doesn’t acknowledge processed records from Kinesis. Does anyone know what could be missing? I’m simply taking those events and writing them to local parquet files. Every time I cancel and start the same job, I see those same records again.

Have you enabled checkpointing?

yes

That I don't know 😄 but checkpointing is the first/only thing I would flag on this

I even turned on RETAIN_ON_CANCELLATION but I see a different folder created in the checkpoints directory for every job once I cancel and start a new one

No, Kinesis stream is not aware of records consumers read from it. It uses iterator to select position to read from. Are you starting your job from savepoint after cancellation?

no

Checkpoint exists for Flink, Kinesis is not aware of it. When you restore from checkpoint/savepoint connector reads iterators persisted in source state and reads records starting from positions on the time of checkpointing,

but Kinesis has it’s own checkpoints stored on DynamoDB

yes

I think I got it. Once I cancel a job I should resume from the savepoint, and if I do that again, I can remove the previous savepoints?

but Kinesis has it’s own checkpoints stored on DynamoDB

I think you talking about Kinesis Client Library: https://docs.aws.amazon.com/streams/latest/dev/shared-throughput-kcl-consumers.html#shared-throughput-kcl-consumers-concepts

yes. I couldn’t figure out if the consumer for flink is based on it or not

I think I got it. Once I cancel a job I should resume from the savepoint, and if I do that again, I can remove the previous savepoints?

correct, you only need latest savepoint unless you need to reprocess older records

thank you. That was very helpful! I assumed Flink is making Kinesis know which records it consumed. So the best practice is to consume from TRIM_HORIZON and use savepoints?

Using TRIM_HORIZON or LATEST depends on your usecase: whether you need to process data from before application was started or not. But if you need to be able to start app from the position you left of - you need to use savepoints.

thanks a lot man

you’re welcome

last question if it’s okay, being implemented with AWS SDK, the consumer lack the ability to aggregate and de-aggregate multiple records on its own. I understand that the connector can pull a single record at a time (from the implementation perspective) so it means that I will hit Kinesis limits faster with the same amount of shards. Is there a workaround for Flink around that?

Source supports de-aggregation (it actually has a dependency on KCL just to use the de-aggregation code).

FlinkKinesisProducer

is based on KPL and support aggregation.

KinesisStreamsSink

is native AWS SDK implementation and does not support aggregation

what about

FlinkKinesisConsumer

?

FlinkKinesisConsumer is a Kinesis source and, as Danny pointed out, supports de-aggregation.

thank you Danny and Aleksandr!