[
https://issues.apache.org/jira/browse/FLINK-41036?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Mohsen Rezaei updated FLINK-41036:
----------------------------------
Description:
If a Postgres CDC job is restored from a checkpoint taken while a large
transaction was in flight, the restore runs out of heap (OutOfMemoryError). The
transaction does not need to touch a captured table; its tables only need to be
in the publication.
The cause is that heartbeats move the checkpointed offset into the middle of
the transaction. On restore, Debezium then re-reads the transaction from BEGIN
and keeps one entry per message in memory until it reaches the stored offset
(unbounded set, reported in
[debezium/dbz#2905|https://github.com/debezium/dbz/issues/2905]).
Reproduced on unmodified apache/flink-cdc master (abf18d7a8), Debezium
1.9.8.Final, Flink 1.20.3, PostgreSQL 16, {{-Xmx512m}} on restore.
h3. When it happens
* The publication contains tables the source does not capture (e.g. a
user-managed or shared publication; the default {{filtered}} autocreate mode
cannot trigger it).
* One transaction on those tables with a very large number of WAL messages.
* A checkpoint is taken during it and the job is later restored from it (e.g.
after a TaskManager failure).
h3. Evidence
Table {{a}} is captured and gets one row; then one transaction inserts
8,000,000 rows into {{b}} (in the publication, not captured). Stream split
offset logged at each checkpoint:
||Checkpoint||heartbeat.interval.ms=1000||heartbeat.interval.ms=0||
|9 / 8|{{0/1524A10}} (last commit)|{{0/1524900}} (last commit)|
|15 / 12|{{0/10741438}} (inside the transaction)|{{0/1524900}}|
|18 / 16|{{0/3E1157C8}} (inside the transaction)|{{0/1524900}}|
No record for {{a}} is emitted during the transaction, so only the heartbeat
moves the offset ({{IncrementalSourceRecordEmitter}} L118-120).
Restore from the heartbeat run's checkpoint (kill -9 after the commit,
{{-Xmx512m}}):
{code}
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.HashSet.add(HashSet.java:221)
at
io.debezium.connector.postgresql.connection.WalPositionLocator.resumeFromLsn(WalPositionLocator.java:63)
at
io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.searchWalPosition(PostgresStreamingChangeEventSource.java:407)
...
{code}
The heap histogram shows about 2M {{HashMap$Node}} and {{Lsn}} instances after
10 s, roughly 60-80 bytes per message (heap dump available).
Control: restoring the {{heartbeat.interval.ms=0}} checkpoint (stored offset
equals the last commit) finds the resume position after 5 ms and stays at about
38 MiB.
{{heartbeat.interval.ms=0}} is not a workaround: the slot then does not advance
during the transaction ({{confirmed_flush_lsn}} stayed at {{0/1524900}} while
the server was at {{0/5682ADD8}}), and {{PostgresSourceConfigFactory}} L90-94
says the heartbeat is required for Postgres.
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.
* 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
Harness attached ({{FlinkRepro.java}}, {{run.sh}}, {{restore.sh}}).
# Postgres 16 with {{wal_level=logical}}; tables {{a}}, {{b}}; publication
{{FOR TABLE a, b}} ({{publication.autocreate.mode=disabled}}).
# Start {{FlinkRepro}} (-Xmx1g), heartbeat 1000 ms, checkpoint interval 2 s,
startup mode latest.
# Insert one row into {{a}}, then 8,000,000 rows into {{b}} in one transaction.
# About 12 s after the commit, kill -9 the job.
# Restart with {{-Xmx512m}} and {{execution.savepoint.path}} set to the latest
{{chk-N}} directory.
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:
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
>
> If a Postgres CDC job is restored from a checkpoint taken while a large
> transaction was in flight, the restore runs out of heap (OutOfMemoryError).
> The transaction does not need to touch a captured table; its tables only need
> to be in the publication.
> The cause is that heartbeats move the checkpointed offset into the middle of
> the transaction. On restore, Debezium then re-reads the transaction from
> BEGIN and keeps one entry per message in memory until it reaches the stored
> offset (unbounded set, reported in
> [debezium/dbz#2905|https://github.com/debezium/dbz/issues/2905]).
> Reproduced on unmodified apache/flink-cdc master (abf18d7a8), Debezium
> 1.9.8.Final, Flink 1.20.3, PostgreSQL 16, {{-Xmx512m}} on restore.
> h3. When it happens
> * The publication contains tables the source does not capture (e.g. a
> user-managed or shared publication; the default {{filtered}} autocreate mode
> cannot trigger it).
> * One transaction on those tables with a very large number of WAL messages.
> * A checkpoint is taken during it and the job is later restored from it (e.g.
> after a TaskManager failure).
> h3. Evidence
> Table {{a}} is captured and gets one row; then one transaction inserts
> 8,000,000 rows into {{b}} (in the publication, not captured). Stream split
> offset logged at each checkpoint:
> ||Checkpoint||heartbeat.interval.ms=1000||heartbeat.interval.ms=0||
> |9 / 8|{{0/1524A10}} (last commit)|{{0/1524900}} (last commit)|
> |15 / 12|{{0/10741438}} (inside the transaction)|{{0/1524900}}|
> |18 / 16|{{0/3E1157C8}} (inside the transaction)|{{0/1524900}}|
> No record for {{a}} is emitted during the transaction, so only the heartbeat
> moves the offset ({{IncrementalSourceRecordEmitter}} L118-120).
> Restore from the heartbeat run's checkpoint (kill -9 after the commit,
> {{-Xmx512m}}):
> {code}
> 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.HashSet.add(HashSet.java:221)
> at
> io.debezium.connector.postgresql.connection.WalPositionLocator.resumeFromLsn(WalPositionLocator.java:63)
> at
> io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.searchWalPosition(PostgresStreamingChangeEventSource.java:407)
> ...
> {code}
> The heap histogram shows about 2M {{HashMap$Node}} and {{Lsn}} instances
> after 10 s, roughly 60-80 bytes per message (heap dump available).
> Control: restoring the {{heartbeat.interval.ms=0}} checkpoint (stored offset
> equals the last commit) finds the resume position after 5 ms and stays at
> about 38 MiB.
> {{heartbeat.interval.ms=0}} is not a workaround: the slot then does not
> advance during the transaction ({{confirmed_flush_lsn}} stayed at
> {{0/1524900}} while the server was at {{0/5682ADD8}}), and
> {{PostgresSourceConfigFactory}} L90-94 says the heartbeat is required for
> Postgres.
> 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.
> * 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
> Harness attached ({{FlinkRepro.java}}, {{run.sh}}, {{restore.sh}}).
> # Postgres 16 with {{wal_level=logical}}; tables {{a}}, {{b}}; publication
> {{FOR TABLE a, b}} ({{publication.autocreate.mode=disabled}}).
> # Start {{FlinkRepro}} (-Xmx1g), heartbeat 1000 ms, checkpoint interval 2 s,
> startup mode latest.
> # Insert one row into {{a}}, then 8,000,000 rows into {{b}} in one
> transaction.
> # About 12 s after the commit, kill -9 the job.
> # Restart with {{-Xmx512m}} and {{execution.savepoint.path}} set to the
> latest {{chk-N}} directory.
> 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)
