Flink CDC | OOM during initial snapshot

[ayush-san]

I am using flink CDC to stream CDC changes in an iceberg table. When I first run the flink job for a topic with all the data for a table, it gets out of heap memory as flink tries to load all the data during my 15mins checkpointing interval. Right now, the only solution I have is to pass -ytm 8192 -yjm 2048m -p 2 for a table with 10M rows and then reduce it after the flink has consumed all the data.
I am attaching a sample job DAG, for my prod job kafka topic has 2 partitions

Shouldn't my iceberg sink have propagated the back-pressure up the source since it has not written that data yet and it will only do that after the checkpointing interval?

FlinkSink.forRowData(rowDataDataStream)
                .table(icebergTable)
                .tableSchema(FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(icebergTable.schema())))
                .tableLoader(tableLoader)
                .equalityFieldColumns(tableConfig.getEqualityColumns())
                .build();

[openinx]

@ayush-san , I think that's because we've maintained all the keys that come from the same checkpoint in a in-memory HashMap, it mainly used to locate the <file_id, pos> for the rows that was written in the current checkpoint before. In the long run, we need to change this HashMap to a Map that can spill to disk or replace it with an embedded KV lib, so that we can take on a larger number of rows in a single checkpoint. This would be a good document to describe the current design.

FYI @rdblue .

[ayush-san]

@openinx What should be the immediate solution for now? I can think of two ways

• Increase the job and task manager memory and keep the checkpoint the same
• During the initial few hours, reduce the checkpoint to 1 min or 2 mins

[openinx]

@ayush-san I think you may want to try this PR so that we could get ride of OOM when ingest the whole snapshots from the upstream mysql databases: #2680

[github-actions]

This issue has been automatically marked as stale because it has been open for 180 days with no activity. It will be closed in next 14 days if no further activity occurs. To permanently prevent this issue from being considered stale, add the label 'not-stale', but commenting on the issue is preferred when possible.

[github-actions]

This issue has been closed because it has not received any activity in the last 14 days since being marked as 'stale'