You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Mohsen Rezaei (Jira)" <ji...@apache.org> on 2026/10/19 20:14:00 UTC
[jira] [Updated] (FLINK-41036) [postgres-cdc] Restoring from a checkpoint taken inside a large transaction runs out of heap
[ https://issues.apache.org/jira/browse/FLINK-41036?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Mohsen Rezaei updated FLINK-41036:
----------------------------------
Description:
When a Postgres CDC source ({{{}PostgresIncrementalSource{}}}) is restored from a checkpoint or savepoint that was taken while a large transaction was in flight, the restore re-reads the whole transaction up to the stored position and keeps one entry per message in memory. The TaskManager runs out of heap ({{{}OutOfMemoryError{}}}). The transaction does not need to touch any captured table: it is enough that its tables are in the publication.
The root cause is that heartbeat events move the checkpointed stream offset to a position in the middle of a transaction, which forces the expensive WAL position search on restore. I reproduced this end to end with an unmodified on {{{}apache/flink-cdc{}}}'s {{master}} and show that the failure disappears when the offset is not moved by heartbeats.
h3. Trigger conditions
# A publication that contains tables the source does not capture (for example a user-managed {{FOR ALL TABLES}} publication, or one shared with other consumers). With the default publication.autocreate.mode=filtered the publication only holds the captured tables and this cannot happen.
# One transaction with a very large number of WAL messages on such tables.
# A checkpoint or savepoint is taken while that transaction is being processed (certain with a 30 s heartbeat and a long transaction).
# The job is restored from it, for example after a TaskManager failure.
h3. Evidence
h4. 1. The checkpointed offset moves into the transaction because of heartbeats
The reader logs the stream split offset at every checkpoint ("Stream split offset on checkpoint N", {{{}IncrementalSourceReader L572{}}}). Table a (captured) receives one row, then a single transaction inserts 8,000,000 rows into table b (in the publication, not captured). {{{}heartbeat.interval.ms=1000{}}}:
{code:java}
checkpoint 6 : Offset{lsn=LSN{0/1524868}, txId=732} before the insert
checkpoint 9 : Offset{lsn=LSN{0/1524A10}, txId=734} last commit on 'a'
checkpoint 12: Offset{lsn=LSN{0/1524A10}, txId=734}
checkpoint 15: Offset{lsn=LSN{0/10741438}, txId=735} inside the big transaction
checkpoint 18: Offset{lsn=LSN{0/3E1157C8}, txId=735} inside the big transaction
{code}
No record for table a was emitted during the transaction, so the only thing that moved the offset is the heartbeat ({{{}IncrementalSourceRecordEmitter L118-120{}}} calls {{updateStreamSplitState}} for heartbeat records). The same run with {{heartbeat.interval.ms=0}} keeps the offset at the last commit for the whole transaction:
{code:java}
checkpoint 8 : Offset{lsn=LSN{0/1524900}, txId=734}
checkpoint 12: Offset{lsn=LSN{0/1524900}, txId=734}
checkpoint 16: Offset{lsn=LSN{0/1524900}, txId=734}
{code}
h4. 2. Restoring from such a checkpoint runs out of memory
The job was killed with {{kill -9}} about 12 s after the commit and restored from the latest retained checkpoint ({{{}chk-19{}}}, heartbeat 1 s) with {{{}-Xmx512m{}}}:
{code:java}
Retrieved latest position from stored offset 'LSN{0/4D6AFAB8}'
Looking for WAL restart position for last commit LSN 'LSN{0/1524A10}' and last change LSN 'LSN{0/4D6AFAB8}'
Searching for WAL resume position
ERROR ErrorHandler - Producer failure
java.lang.OutOfMemoryError: Java heap space
at java.util.HashMap.resize(HashMap.java:702)
at java.util.HashMap.putVal(HashMap.java:661)
at java.util.HashMap.put(HashMap.java:610)
at java.util.HashSet.add(HashSet.java:221)
at io.debezium.connector.postgresql.connection.WalPositionLocator.resumeFromLsn(WalPositionLocator.java:63)
at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.lambda$searchWalPosition$1(PostgresStreamingChangeEventSource.java:410)
at io.debezium.connector.postgresql.connection.pgoutput.PgOutputMessageDecoder.decodeInsert(PgOutputMessageDecoder.java:413)
at io.debezium.connector.postgresql.connection.pgoutput.PgOutputMessageDecoder.processNotEmptyMessage(PgOutputMessageDecoder.java:191)
at io.debezium.connector.postgresql.connection.PostgresReplicationConnection$1.readPending(PostgresReplicationConnection.java:611)
at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.searchWalPosition(PostgresStreamingChangeEventSource.java:407)
at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:200)
at org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresStreamFetchTask$StreamSplitReadTask.execute(PostgresStreamFetchTask.java:223)
{code}
Class histogram of the TaskManager heap ten seconds into the search ({{{}jcmd GC.class_histogram{}}}):
||Class||Instances||Bytes||
|java.util.HashMap$Node|1,920,337|61 MB|
|io.debezium.connector.postgresql.connection.Lsn|2,126,036|51 MB|
That is about 60-80 bytes per message, so the memory needed grows linearly with the distance between the start of the transaction and the stored position (a heap dump of about 750 MB was written with {{-XX:+HeapDumpOnOutOfMemoryError}} and can be added).
h4. 3. Control: same restore without heartbeat finishes the search immediately
Restoring chk-19 of the {{heartbeat.interval.ms=0}} run (stored offset {{{}0/1524900{}}}, equal to the last commit) with the same {{{}-Xmx512m{}}}:
{code:java}
Looking for WAL restart position for last commit LSN 'LSN{0/1524900}' and last change LSN 'LSN{0/1524900}'
Searching for WAL resume position
WAL resume position 'LSN{0/1524A10}' discovered
MEM used=38MiB max=494MiB
{code}
The resume position is found 5 ms after the search starts, because the first message after the stored offset is already past it. Heap stays at about 38 MiB, and the job keeps running and checkpointing.
h4. 4. Why heartbeat.interval.ms=0 is not a usable workaround
In the run without heartbeat the replication slot does not move at all while the large transaction is consumed: at the end of the run {{confirmed_flush_lsn}} was {{0/1524900}} while {{pg_current_wal_lsn()}} was {{{}0/5682ADD8{}}}. With heartbeat enabled confirmed_flush_lsn followed the consumer ({{{}0/4D6AFAB8{}}}). Without heartbeat the slot retains WAL for the whole transaction (and for any other activity that is filtered out), and {{PostgresSourceConfigFactory L90-94}} states that the heartbeat is required for Postgres ("we have to enable heartbeat for PG to make sure {{DebeziumChangeConsumer#handleBatch}} is invoked after job restart"). The default is 30 s ({{{}PostgresSourceOptions L74-77{}}}).
h3. Analysis
# Debezium emits a heartbeat after dispatched events, including events that are filtered out (rate-limited by {{{}heartbeat.interval.ms{}}}), and filtered messages still advance lsn and lsn_proc.
# {{IncrementalSourceRecordEmitter#processElement}} applies the heartbeat offset to the stream split ({{{}L118-120{}}}), so the next checkpoint stores an lsn_proc in the middle of a transaction.
# On restore {{hasCompletelyProcessedPosition()}} is true for that offset, so {{PostgresStreamingChangeEventSource}} runs {{{}searchWalPosition(){}}}. pgoutput re-sends the whole transaction from {{BEGIN}} and Debezium's {{WalPositionLocator}} keeps every LSN in a HashSet ({{{}lsnSeen{}}}) until the stored position shows up. Memory is proportional to the distance from {{BEGIN}} to the stored position. The unbounded set is Debezium behaviour and is reported there separately: [debezium/dbz#2905|https://github.com/debezium/dbz/issues/2905].
The Flink CDC part is that the checkpointed offset can point into the middle of a transaction at all. Without heartbeats the stored offset stays at the last commit (evidence 1 and 3).
h3. Possible directions
* Do not apply a heartbeat offset to the checkpointed stream split while a transaction is in flight; keep the last-commit position until {{{}COMMIT{}}}. The slot can still be advanced through the commit path.
* Skip the position search when everything since the last commit was filtered out.
* Bound the memory of the search in Debezium (see the linked issue).
h3. Reproduction
The harness (FlinkRepro.java, run.sh, restore.sh) is attached. In short:
# docker run {{postgres:16}} with {{{}wal_level=logical{}}}; create tables a and b and a publication dbz_pub FOR TABLE a, b (publication.autocreate.mode=disabled).
# Start FlinkRepro ({{{}-Xmx1g{}}}) with {{{}heartbeat.interval.ms=1000{}}}, checkpoint interval 2 s, startup mode latest.
# Insert one row into a, then insert 8,000,000 rows into b in one transaction.
# About 12 s after the commit, {{kill -9}} the job.
# Restart FlinkRepro with {{-Xmx512m}} and {{execution.savepoint.path}} set to the latest retained chk-N directory.
Result: {{OutOfMemoryError}} in {{{}WalPositionLocator.resumeFromLsn{}}}. With {{heartbeat.interval.ms=0}} (control) the same restore completes in milliseconds.
{code:java}
import java.time.Duration;
import java.util.Properties;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cdc.connectors.base.options.StartupOptions;
import org.apache.flink.cdc.connectors.postgres.source.PostgresSourceBuilder;
import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
/** args: <checkpointDir> <heartbeatMs> [restoreFromCheckpointPath] */
public class FlinkRepro {
public static void main(String[] a) throws Exception {
Configuration conf = new Configuration();
conf.setString("state.checkpoints.dir", "file://" + a[0]);
conf.setString("execution.checkpointing.externalized-checkpoint-retention", "RETAIN_ON_CANCELLATION");
conf.setString("restart-strategy.type", "none");
if (a.length > 2) conf.setString("execution.savepoint.path", "file://" + a[2]);
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
env.setParallelism(1);
env.enableCheckpointing(2000);
Properties dbz = new Properties();
dbz.setProperty("publication.name", "dbz_pub");
dbz.setProperty("publication.autocreate.mode", "disabled");
var src = PostgresSourceBuilder.PostgresIncrementalSource.<String>builder()
.hostname("localhost").port(55432).database("postgres")
.schemaList("public").tableList("public.a")
.username("postgres").password("pw")
.slotName("flink_slot").decodingPluginName("pgoutput")
.deserializer(new JsonDebeziumDeserializationSchema())
.startupOptions(StartupOptions.latest())
.heartbeatInterval(Duration.ofMillis(Long.parseLong(a[1])))
.debeziumProperties(dbz)
.build();
env.fromSource(src, WatermarkStrategy.noWatermarks(), "pg").addSink(new DiscardingSink<>());
Runtime rt = Runtime.getRuntime();
Thread t = new Thread(() -> {
try {
while (true) {
Thread.sleep(5000);
System.out.printf("MEM used=%dMiB max=%dMiB%n", (rt.totalMemory()-rt.freeMemory())>>20, rt.maxMemory()>>20);
}
} catch (Exception e) {}
});
t.setDaemon(true); t.start();
env.execute("repro");
}
}
{code}
h3. Related
* FLINK-40538, FLINK-39265
* Debezium: [debezium/dbz#2905|https://github.com/debezium/dbz/issues/2905] (unbounded {{{}WalPositionLocator.lsnSeen{}}})
[~loserwang1024] [~leonard] [~eskabetxe] since you have worked on the Postgres CDC restore path (FLINK-40538), I would appreciate your thoughts on the proposed fix.
was:
h3. Summary
When a Postgres CDC source ({{{}PostgresIncrementalSource{}}}) is restored from a checkpoint or savepoint that was taken while a large transaction was in flight, the restore re-reads the whole transaction up to the stored position and keeps one entry per message in memory. The TaskManager runs out of heap ({{{}OutOfMemoryError{}}}). The transaction does not need to touch any captured table: it is enough that its tables are in the publication.
The root cause is that heartbeat events move the checkpointed stream offset to a position in the middle of a transaction, which forces the expensive WAL position search on restore. I reproduced this end to end with an unmodified on {{{}apache/flink-cdc{}}}'s {{master}} and show that the failure disappears when the offset is not moved by heartbeats.
h3. Trigger conditions
# A publication that contains tables the source does not capture (for example a user-managed {{FOR ALL TABLES}} publication, or one shared with other consumers). With the default publication.autocreate.mode=filtered the publication only holds the captured tables and this cannot happen.
# One transaction with a very large number of WAL messages on such tables.
# A checkpoint or savepoint is taken while that transaction is being processed (certain with a 30 s heartbeat and a long transaction).
# The job is restored from it, for example after a TaskManager failure.
h3. Evidence
h4. 1. The checkpointed offset moves into the transaction because of heartbeats
The reader logs the stream split offset at every checkpoint ("Stream split offset on checkpoint N", {{{}IncrementalSourceReader L572{}}}). Table a (captured) receives one row, then a single transaction inserts 8,000,000 rows into table b (in the publication, not captured). {{{}heartbeat.interval.ms=1000{}}}:
{code:java}
checkpoint 6 : Offset{lsn=LSN{0/1524868}, txId=732} before the insert
checkpoint 9 : Offset{lsn=LSN{0/1524A10}, txId=734} last commit on 'a'
checkpoint 12: Offset{lsn=LSN{0/1524A10}, txId=734}
checkpoint 15: Offset{lsn=LSN{0/10741438}, txId=735} inside the big transaction
checkpoint 18: Offset{lsn=LSN{0/3E1157C8}, txId=735} inside the big transaction
{code}
No record for table a was emitted during the transaction, so the only thing that moved the offset is the heartbeat ({{{}IncrementalSourceRecordEmitter L118-120{}}} calls {{updateStreamSplitState}} for heartbeat records). The same run with {{heartbeat.interval.ms=0}} keeps the offset at the last commit for the whole transaction:
{code:java}
checkpoint 8 : Offset{lsn=LSN{0/1524900}, txId=734}
checkpoint 12: Offset{lsn=LSN{0/1524900}, txId=734}
checkpoint 16: Offset{lsn=LSN{0/1524900}, txId=734}
{code}
h4. 2. Restoring from such a checkpoint runs out of memory
The job was killed with {{kill -9}} about 12 s after the commit and restored from the latest retained checkpoint ({{{}chk-19{}}}, heartbeat 1 s) with {{{}-Xmx512m{}}}:
{code:java}
Retrieved latest position from stored offset 'LSN{0/4D6AFAB8}'
Looking for WAL restart position for last commit LSN 'LSN{0/1524A10}' and last change LSN 'LSN{0/4D6AFAB8}'
Searching for WAL resume position
ERROR ErrorHandler - Producer failure
java.lang.OutOfMemoryError: Java heap space
at java.util.HashMap.resize(HashMap.java:702)
at java.util.HashMap.putVal(HashMap.java:661)
at java.util.HashMap.put(HashMap.java:610)
at java.util.HashSet.add(HashSet.java:221)
at io.debezium.connector.postgresql.connection.WalPositionLocator.resumeFromLsn(WalPositionLocator.java:63)
at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.lambda$searchWalPosition$1(PostgresStreamingChangeEventSource.java:410)
at io.debezium.connector.postgresql.connection.pgoutput.PgOutputMessageDecoder.decodeInsert(PgOutputMessageDecoder.java:413)
at io.debezium.connector.postgresql.connection.pgoutput.PgOutputMessageDecoder.processNotEmptyMessage(PgOutputMessageDecoder.java:191)
at io.debezium.connector.postgresql.connection.PostgresReplicationConnection$1.readPending(PostgresReplicationConnection.java:611)
at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.searchWalPosition(PostgresStreamingChangeEventSource.java:407)
at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:200)
at org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresStreamFetchTask$StreamSplitReadTask.execute(PostgresStreamFetchTask.java:223)
{code}
Class histogram of the TaskManager heap ten seconds into the search ({{{}jcmd GC.class_histogram{}}}):
||Class||Instances||Bytes||
|java.util.HashMap$Node|1,920,337|61 MB|
|io.debezium.connector.postgresql.connection.Lsn|2,126,036|51 MB|
That is about 60-80 bytes per message, so the memory needed grows linearly with the distance between the start of the transaction and the stored position (a heap dump of about 750 MB was written with {{-XX:+HeapDumpOnOutOfMemoryError}} and can be added).
h4. 3. Control: same restore without heartbeat finishes the search immediately
Restoring chk-19 of the {{heartbeat.interval.ms=0}} run (stored offset {{{}0/1524900{}}}, equal to the last commit) with the same {{{}-Xmx512m{}}}:
{code:java}
Looking for WAL restart position for last commit LSN 'LSN{0/1524900}' and last change LSN 'LSN{0/1524900}'
Searching for WAL resume position
WAL resume position 'LSN{0/1524A10}' discovered
MEM used=38MiB max=494MiB
{code}
The resume position is found 5 ms after the search starts, because the first message after the stored offset is already past it. Heap stays at about 38 MiB, and the job keeps running and checkpointing.
h4. 4. Why heartbeat.interval.ms=0 is not a usable workaround
In the run without heartbeat the replication slot does not move at all while the large transaction is consumed: at the end of the run {{confirmed_flush_lsn}} was {{0/1524900}} while {{pg_current_wal_lsn()}} was {{{}0/5682ADD8{}}}. With heartbeat enabled confirmed_flush_lsn followed the consumer ({{{}0/4D6AFAB8{}}}). Without heartbeat the slot retains WAL for the whole transaction (and for any other activity that is filtered out), and {{PostgresSourceConfigFactory L90-94}} states that the heartbeat is required for Postgres ("we have to enable heartbeat for PG to make sure {{DebeziumChangeConsumer#handleBatch}} is invoked after job restart"). The default is 30 s ({{{}PostgresSourceOptions L74-77{}}}).
h3. Analysis
# Debezium emits a heartbeat after dispatched events, including events that are filtered out (rate-limited by {{{}heartbeat.interval.ms{}}}), and filtered messages still advance lsn and lsn_proc.
# {{IncrementalSourceRecordEmitter#processElement}} applies the heartbeat offset to the stream split ({{{}L118-120{}}}), so the next checkpoint stores an lsn_proc in the middle of a transaction.
# On restore {{hasCompletelyProcessedPosition()}} is true for that offset, so {{PostgresStreamingChangeEventSource}} runs {{{}searchWalPosition(){}}}. pgoutput re-sends the whole transaction from {{BEGIN}} and Debezium's {{WalPositionLocator}} keeps every LSN in a HashSet ({{{}lsnSeen{}}}) until the stored position shows up. Memory is proportional to the distance from {{BEGIN}} to the stored position. The unbounded set is Debezium behaviour and is reported there separately: [debezium/dbz#2905|https://github.com/debezium/dbz/issues/2905].
The Flink CDC part is that the checkpointed offset can point into the middle of a transaction at all. Without heartbeats the stored offset stays at the last commit (evidence 1 and 3).
h3. Possible directions
* Do not apply a heartbeat offset to the checkpointed stream split while a transaction is in flight; keep the last-commit position until {{{}COMMIT{}}}. The slot can still be advanced through the commit path.
* Skip the position search when everything since the last commit was filtered out.
* Bound the memory of the search in Debezium (see the linked issue).
h3. Reproduction
The harness (FlinkRepro.java, run.sh, restore.sh) is attached. In short:
# docker run {{postgres:16}} with {{{}wal_level=logical{}}}; create tables a and b and a publication dbz_pub FOR TABLE a, b (publication.autocreate.mode=disabled).
# Start FlinkRepro ({{{}-Xmx1g{}}}) with {{{}heartbeat.interval.ms=1000{}}}, checkpoint interval 2 s, startup mode latest.
# Insert one row into a, then insert 8,000,000 rows into b in one transaction.
# About 12 s after the commit, {{kill -9}} the job.
# Restart FlinkRepro with {{-Xmx512m}} and {{execution.savepoint.path}} set to the latest retained chk-N directory.
Result: {{OutOfMemoryError}} in {{{}WalPositionLocator.resumeFromLsn{}}}. With {{heartbeat.interval.ms=0}} (control) the same restore completes in milliseconds.
{code:java}
import java.time.Duration;
import java.util.Properties;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cdc.connectors.base.options.StartupOptions;
import org.apache.flink.cdc.connectors.postgres.source.PostgresSourceBuilder;
import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
/** args: <checkpointDir> <heartbeatMs> [restoreFromCheckpointPath] */
public class FlinkRepro {
public static void main(String[] a) throws Exception {
Configuration conf = new Configuration();
conf.setString("state.checkpoints.dir", "file://" + a[0]);
conf.setString("execution.checkpointing.externalized-checkpoint-retention", "RETAIN_ON_CANCELLATION");
conf.setString("restart-strategy.type", "none");
if (a.length > 2) conf.setString("execution.savepoint.path", "file://" + a[2]);
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
env.setParallelism(1);
env.enableCheckpointing(2000);
Properties dbz = new Properties();
dbz.setProperty("publication.name", "dbz_pub");
dbz.setProperty("publication.autocreate.mode", "disabled");
var src = PostgresSourceBuilder.PostgresIncrementalSource.<String>builder()
.hostname("localhost").port(55432).database("postgres")
.schemaList("public").tableList("public.a")
.username("postgres").password("pw")
.slotName("flink_slot").decodingPluginName("pgoutput")
.deserializer(new JsonDebeziumDeserializationSchema())
.startupOptions(StartupOptions.latest())
.heartbeatInterval(Duration.ofMillis(Long.parseLong(a[1])))
.debeziumProperties(dbz)
.build();
env.fromSource(src, WatermarkStrategy.noWatermarks(), "pg").addSink(new DiscardingSink<>());
Runtime rt = Runtime.getRuntime();
Thread t = new Thread(() -> {
try {
while (true) {
Thread.sleep(5000);
System.out.printf("MEM used=%dMiB max=%dMiB%n", (rt.totalMemory()-rt.freeMemory())>>20, rt.maxMemory()>>20);
}
} catch (Exception e) {}
});
t.setDaemon(true); t.start();
env.execute("repro");
}
}
{code}
h3. Related
* FLINK-40538, FLINK-39265
* Debezium: [debezium/dbz#2905|https://github.com/debezium/dbz/issues/2905] (unbounded {{{}WalPositionLocator.lsnSeen{}}})
[~loserwang1024] [~leonard] [~eskabetxe] since you have worked on the Postgres CDC restore path (FLINK-40538), I would appreciate your thoughts on the proposed fix.
> [postgres-cdc] Restoring from a checkpoint taken inside a large transaction runs out of heap
> --------------------------------------------------------------------------------------------
>
> Key: FLINK-41036
> URL: https://issues.apache.org/jira/browse/FLINK-41036
> Project: Flink
> Issue Type: Bug
> Components: Flink CDC
> Environment: * apache/flink-cdc master, commit {{abf18d7a8}} ({{3.8-SNAPSHOT}}), unmodified; Debezium {{1.9.8.Final}}
> * Flink {{{}1.20.3{}}}, local {{{}MiniCluster{}}}, parallelism 1, checkpoint interval 2 s, externalized checkpoints ({{{}RETAIN_ON_CANCELLATION{}}})
> * PostgreSQL {{16.15}} ({{{}postgres:16{}}} container), {{{}pgoutput{}}}, one slot, one user-managed publication
> * JDK 17, {{{}-XX:+UseSerialGC{}}}, {{-Xmx512m}} on restore
> Reporter: Mohsen Rezaei
> Priority: Major
>
> When a Postgres CDC source ({{{}PostgresIncrementalSource{}}}) is restored from a checkpoint or savepoint that was taken while a large transaction was in flight, the restore re-reads the whole transaction up to the stored position and keeps one entry per message in memory. The TaskManager runs out of heap ({{{}OutOfMemoryError{}}}). The transaction does not need to touch any captured table: it is enough that its tables are in the publication.
> The root cause is that heartbeat events move the checkpointed stream offset to a position in the middle of a transaction, which forces the expensive WAL position search on restore. I reproduced this end to end with an unmodified on {{{}apache/flink-cdc{}}}'s {{master}} and show that the failure disappears when the offset is not moved by heartbeats.
> h3. Trigger conditions
> # A publication that contains tables the source does not capture (for example a user-managed {{FOR ALL TABLES}} publication, or one shared with other consumers). With the default publication.autocreate.mode=filtered the publication only holds the captured tables and this cannot happen.
> # One transaction with a very large number of WAL messages on such tables.
> # A checkpoint or savepoint is taken while that transaction is being processed (certain with a 30 s heartbeat and a long transaction).
> # The job is restored from it, for example after a TaskManager failure.
> h3. Evidence
> h4. 1. The checkpointed offset moves into the transaction because of heartbeats
> The reader logs the stream split offset at every checkpoint ("Stream split offset on checkpoint N", {{{}IncrementalSourceReader L572{}}}). Table a (captured) receives one row, then a single transaction inserts 8,000,000 rows into table b (in the publication, not captured). {{{}heartbeat.interval.ms=1000{}}}:
> {code:java}
> checkpoint 6 : Offset{lsn=LSN{0/1524868}, txId=732} before the insert
> checkpoint 9 : Offset{lsn=LSN{0/1524A10}, txId=734} last commit on 'a'
> checkpoint 12: Offset{lsn=LSN{0/1524A10}, txId=734}
> checkpoint 15: Offset{lsn=LSN{0/10741438}, txId=735} inside the big transaction
> checkpoint 18: Offset{lsn=LSN{0/3E1157C8}, txId=735} inside the big transaction
> {code}
> No record for table a was emitted during the transaction, so the only thing that moved the offset is the heartbeat ({{{}IncrementalSourceRecordEmitter L118-120{}}} calls {{updateStreamSplitState}} for heartbeat records). The same run with {{heartbeat.interval.ms=0}} keeps the offset at the last commit for the whole transaction:
> {code:java}
> checkpoint 8 : Offset{lsn=LSN{0/1524900}, txId=734}
> checkpoint 12: Offset{lsn=LSN{0/1524900}, txId=734}
> checkpoint 16: Offset{lsn=LSN{0/1524900}, txId=734}
> {code}
> h4. 2. Restoring from such a checkpoint runs out of memory
> The job was killed with {{kill -9}} about 12 s after the commit and restored from the latest retained checkpoint ({{{}chk-19{}}}, heartbeat 1 s) with {{{}-Xmx512m{}}}:
> {code:java}
> Retrieved latest position from stored offset 'LSN{0/4D6AFAB8}'
> Looking for WAL restart position for last commit LSN 'LSN{0/1524A10}' and last change LSN 'LSN{0/4D6AFAB8}'
> Searching for WAL resume position
> ERROR ErrorHandler - Producer failure
> java.lang.OutOfMemoryError: Java heap space
> at java.util.HashMap.resize(HashMap.java:702)
> at java.util.HashMap.putVal(HashMap.java:661)
> at java.util.HashMap.put(HashMap.java:610)
> at java.util.HashSet.add(HashSet.java:221)
> at io.debezium.connector.postgresql.connection.WalPositionLocator.resumeFromLsn(WalPositionLocator.java:63)
> at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.lambda$searchWalPosition$1(PostgresStreamingChangeEventSource.java:410)
> at io.debezium.connector.postgresql.connection.pgoutput.PgOutputMessageDecoder.decodeInsert(PgOutputMessageDecoder.java:413)
> at io.debezium.connector.postgresql.connection.pgoutput.PgOutputMessageDecoder.processNotEmptyMessage(PgOutputMessageDecoder.java:191)
> at io.debezium.connector.postgresql.connection.PostgresReplicationConnection$1.readPending(PostgresReplicationConnection.java:611)
> at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.searchWalPosition(PostgresStreamingChangeEventSource.java:407)
> at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:200)
> at org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresStreamFetchTask$StreamSplitReadTask.execute(PostgresStreamFetchTask.java:223)
> {code}
> Class histogram of the TaskManager heap ten seconds into the search ({{{}jcmd GC.class_histogram{}}}):
> ||Class||Instances||Bytes||
> |java.util.HashMap$Node|1,920,337|61 MB|
> |io.debezium.connector.postgresql.connection.Lsn|2,126,036|51 MB|
> That is about 60-80 bytes per message, so the memory needed grows linearly with the distance between the start of the transaction and the stored position (a heap dump of about 750 MB was written with {{-XX:+HeapDumpOnOutOfMemoryError}} and can be added).
> h4. 3. Control: same restore without heartbeat finishes the search immediately
> Restoring chk-19 of the {{heartbeat.interval.ms=0}} run (stored offset {{{}0/1524900{}}}, equal to the last commit) with the same {{{}-Xmx512m{}}}:
> {code:java}
> Looking for WAL restart position for last commit LSN 'LSN{0/1524900}' and last change LSN 'LSN{0/1524900}'
> Searching for WAL resume position
> WAL resume position 'LSN{0/1524A10}' discovered
> MEM used=38MiB max=494MiB
> {code}
> The resume position is found 5 ms after the search starts, because the first message after the stored offset is already past it. Heap stays at about 38 MiB, and the job keeps running and checkpointing.
> h4. 4. Why heartbeat.interval.ms=0 is not a usable workaround
> In the run without heartbeat the replication slot does not move at all while the large transaction is consumed: at the end of the run {{confirmed_flush_lsn}} was {{0/1524900}} while {{pg_current_wal_lsn()}} was {{{}0/5682ADD8{}}}. With heartbeat enabled confirmed_flush_lsn followed the consumer ({{{}0/4D6AFAB8{}}}). Without heartbeat the slot retains WAL for the whole transaction (and for any other activity that is filtered out), and {{PostgresSourceConfigFactory L90-94}} states that the heartbeat is required for Postgres ("we have to enable heartbeat for PG to make sure {{DebeziumChangeConsumer#handleBatch}} is invoked after job restart"). The default is 30 s ({{{}PostgresSourceOptions L74-77{}}}).
> h3. Analysis
> # Debezium emits a heartbeat after dispatched events, including events that are filtered out (rate-limited by {{{}heartbeat.interval.ms{}}}), and filtered messages still advance lsn and lsn_proc.
> # {{IncrementalSourceRecordEmitter#processElement}} applies the heartbeat offset to the stream split ({{{}L118-120{}}}), so the next checkpoint stores an lsn_proc in the middle of a transaction.
> # On restore {{hasCompletelyProcessedPosition()}} is true for that offset, so {{PostgresStreamingChangeEventSource}} runs {{{}searchWalPosition(){}}}. pgoutput re-sends the whole transaction from {{BEGIN}} and Debezium's {{WalPositionLocator}} keeps every LSN in a HashSet ({{{}lsnSeen{}}}) until the stored position shows up. Memory is proportional to the distance from {{BEGIN}} to the stored position. The unbounded set is Debezium behaviour and is reported there separately: [debezium/dbz#2905|https://github.com/debezium/dbz/issues/2905].
> The Flink CDC part is that the checkpointed offset can point into the middle of a transaction at all. Without heartbeats the stored offset stays at the last commit (evidence 1 and 3).
> h3. Possible directions
> * Do not apply a heartbeat offset to the checkpointed stream split while a transaction is in flight; keep the last-commit position until {{{}COMMIT{}}}. The slot can still be advanced through the commit path.
> * Skip the position search when everything since the last commit was filtered out.
> * Bound the memory of the search in Debezium (see the linked issue).
> h3. Reproduction
> The harness (FlinkRepro.java, run.sh, restore.sh) is attached. In short:
> # docker run {{postgres:16}} with {{{}wal_level=logical{}}}; create tables a and b and a publication dbz_pub FOR TABLE a, b (publication.autocreate.mode=disabled).
> # Start FlinkRepro ({{{}-Xmx1g{}}}) with {{{}heartbeat.interval.ms=1000{}}}, checkpoint interval 2 s, startup mode latest.
> # Insert one row into a, then insert 8,000,000 rows into b in one transaction.
> # About 12 s after the commit, {{kill -9}} the job.
> # Restart FlinkRepro with {{-Xmx512m}} and {{execution.savepoint.path}} set to the latest retained chk-N directory.
> Result: {{OutOfMemoryError}} in {{{}WalPositionLocator.resumeFromLsn{}}}. With {{heartbeat.interval.ms=0}} (control) the same restore completes in milliseconds.
> {code:java}
> import java.time.Duration;
> import java.util.Properties;
> import org.apache.flink.api.common.eventtime.WatermarkStrategy;
> import org.apache.flink.cdc.connectors.base.options.StartupOptions;
> import org.apache.flink.cdc.connectors.postgres.source.PostgresSourceBuilder;
> import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
> import org.apache.flink.configuration.Configuration;
> import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
> import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
> /** args: <checkpointDir> <heartbeatMs> [restoreFromCheckpointPath] */
> public class FlinkRepro {
> public static void main(String[] a) throws Exception {
> Configuration conf = new Configuration();
> conf.setString("state.checkpoints.dir", "file://" + a[0]);
> conf.setString("execution.checkpointing.externalized-checkpoint-retention", "RETAIN_ON_CANCELLATION");
> conf.setString("restart-strategy.type", "none");
> if (a.length > 2) conf.setString("execution.savepoint.path", "file://" + a[2]);
> StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
> env.setParallelism(1);
> env.enableCheckpointing(2000);
> Properties dbz = new Properties();
> dbz.setProperty("publication.name", "dbz_pub");
> dbz.setProperty("publication.autocreate.mode", "disabled");
> var src = PostgresSourceBuilder.PostgresIncrementalSource.<String>builder()
> .hostname("localhost").port(55432).database("postgres")
> .schemaList("public").tableList("public.a")
> .username("postgres").password("pw")
> .slotName("flink_slot").decodingPluginName("pgoutput")
> .deserializer(new JsonDebeziumDeserializationSchema())
> .startupOptions(StartupOptions.latest())
> .heartbeatInterval(Duration.ofMillis(Long.parseLong(a[1])))
> .debeziumProperties(dbz)
> .build();
> env.fromSource(src, WatermarkStrategy.noWatermarks(), "pg").addSink(new DiscardingSink<>());
> Runtime rt = Runtime.getRuntime();
> Thread t = new Thread(() -> {
> try {
> while (true) {
> Thread.sleep(5000);
> System.out.printf("MEM used=%dMiB max=%dMiB%n", (rt.totalMemory()-rt.freeMemory())>>20, rt.maxMemory()>>20);
> }
> } catch (Exception e) {}
> });
> t.setDaemon(true); t.start();
> env.execute("repro");
> }
> }
> {code}
> h3. Related
> * FLINK-40538, FLINK-39265
> * Debezium: [debezium/dbz#2905|https://github.com/debezium/dbz/issues/2905] (unbounded {{{}WalPositionLocator.lsnSeen{}}})
> [~loserwang1024] [~leonard] [~eskabetxe] since you have worked on the Postgres CDC restore path (FLINK-40538), I would appreciate your thoughts on the proposed fix.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)