Mohsen Rezaei created FLINK-41036:
-------------------------------------
Summary: [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
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.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
