[fix](streamingjob) fix postgres DML silently dropped on task restart by JNSimba · Pull Request #61481 · apache/doris

JNSimba · 2026-03-18T09:35:13Z

What problem does this PR solve?

Problem

When a streaming job restarts a task, the first DML of the new transaction
is occasionally silently dropped (10-20% failure rate). The affected record
never appears in the Doris target table, with no error logged — only
"identified as already processed" in cdc-client.log.

Root Cause

debezium 1.9.x hardcodes proto_version=1 (non-streaming pgoutput) for all
PG versions. In non-streaming mode, the walsender batches all changes of a
transaction and sends them after COMMIT, and all messages (BEGIN + DML) share
the same XLogData.data_start = the transaction's begin_lsn.

When this begin_lsn equals the previous transaction's commit_lsn (i.e.
the two transactions are adjacent in WAL with no other writes between them),
WalPositionLocator behaves incorrectly:

Find phase: COMMIT(T1) at lsn=Y sets storeLsnAfterLastEventStoredLsn=true.
BEGIN(T2) and INSERT(T2) both have lsn=Y, so they keep returning
Optional.empty(). Only COMMIT(T2) at lsn=Z sets
startStreamingLsn=Z, with lsnSeen={Y, Z}.
Actual streaming: INSERT(T2) arrives with lastReceiveLsn=Y.
skipMessage(Y): Y ∈ lsnSeen and Y ≠ startStreamingLsn(Z) → filtered.

The bug is intermittent because it only triggers when no other WAL activity
(autovacuum, other connections) occurs between the two transactions.

Fix

Override extractBinlogStateOffset() in PostgresSourceReader to strip
lsn_proc and lsn_commit from the offset before it is passed to debezium.
This constructs WalPositionLocator(lastCommitStoredLsn=null, lsn=Y), which
causes the find phase to exit immediately at the first received message
(startStreamingLsn=Y). In actual streaming, COMMIT(T1) triggers
switch-off (lastReceiveLsn=Y = startStreamingLsn), and all subsequent
messages including INSERT(T2) pass through.

See https://issues.apache.org/jira/browse/FLINK-39265.

Test

Run test_streaming_postgres_job multiple times. Before this fix the
'Apache' assertion fails ~10-20% of the time; after this fix it passes
consistently.

Release note

None

Check List (For Author)

Test
- Regression test
- Unit Test
- Manual test (add detailed scripts or steps below)
- No need to test or manual test. Explain why:
  - This is a refactor/code format and no logic has been changed.
  - Previous test can cover this change.
  - No code files have been changed.
  - Other reason
Behavior changed:
- No.
- Yes.
Does this need documentation?
- No.
- Yes.

Check List (For Reviewer who merge this PR)

Confirm the release note
Confirm test cases
Confirm document
Add branch pick label

JNSimba · 2026-03-18T09:35:41Z

run buildall

Thearas · 2026-03-18T09:35:42Z

Thank you for your contribution to Apache Doris.
Don't know what should be done next? See How to process your PR.

Please clearly describe your PR:

What problem was fixed (it's best to include specific error reporting information). How it was fixed.
Which behaviors were modified. What was the previous behavior, what is it now, why was it modified, and what possible impacts might there be.
What features were added. Why was this function added?
Which code was refactored and why was this part of the code refactored?
Which functions were optimized and what is the difference before and after the optimization?

JNSimba · 2026-03-18T10:54:06Z

run buildall

doris-robot · 2026-03-18T11:12:35Z

TPC-H: Total hot run time: 27048 ms

machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpch-tools
Tpch sf100 test result on commit 314e84b8a192ae2ead690f17b9d4a6d6a9b5839f, data reload: false
------ Round 1 ----------------------------------
orders	Doris	NULL	NULL	0	0	0	NULL	0	NULL	NULL	2023-12-26 18:27:23	2023-12-26 18:42:55	NULL	utf-8	NULL	NULL ============================================
q1	17600	4428	4284	4284
q2	q3	10640	771	544	544
q4	4680	355	251	251
q5	7561	1224	1012	1012
q6	187	178	153	153
q7	794	840	667	667
q8	9304	1514	1449	1449
q9	4891	4745	4762	4745
q10	6305	1910	1659	1659
q11	467	288	239	239
q12	692	581	468	468
q13	18039	2926	2183	2183
q14	223	231	223	223
q15	q16	758	731	667	667
q17	723	902	436	436
q18	5974	5403	5243	5243
q19	1293	999	621	621
q20	562	500	376	376
q21	4820	1873	1530	1530
q22	482	364	298	298
Total cold run time: 95995 ms
Total hot run time: 27048 ms
----- Round 2, with runtime_filter_mode=off -----
orders	Doris	NULL	NULL	150000000	42	6422171781	NULL	22778155	NULL	NULL	2023-12-26 18:27:23	2023-12-26 18:42:55	NULL	utf-8	NULL	NULL ============================================
q1	4771	4603	4664	4603
q2	q3	3889	4354	3822	3822
q4	944	1264	818	818
q5	4094	4389	4390	4389
q6	185	174	144	144
q7	1780	1678	1543	1543
q8	2532	2723	2968	2723
q9	7669	7472	7563	7472
q10	3761	3983	3546	3546
q11	524	443	417	417
q12	554	666	518	518
q13	2772	3164	2433	2433
q14	286	302	280	280
q15	q16	711	760	723	723
q17	1192	1277	1327	1277
q18	7164	6988	6550	6550
q19	975	982	1040	982
q20	2111	2109	2004	2004
q21	3997	3473	3326	3326
q22	472	426	376	376
Total cold run time: 50383 ms
Total hot run time: 47946 ms

doris-robot · 2026-03-18T11:23:20Z

TPC-DS: Total hot run time: 168708 ms

machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpcds-tools
TPC-DS sf100 test result on commit 314e84b8a192ae2ead690f17b9d4a6d6a9b5839f, data reload: false
query5	4342	654	501	501
query6	336	232	217	217
query7	4214	469	278	278
query8	347	253	241	241
query9	8786	2759	2732	2732
query10	514	427	342	342
query11	6992	5112	4936	4936
query12	182	133	129	129
query13	1274	494	366	366
query14	5728	3751	3504	3504
query14_1	2932	2885	2845	2845
query15	212	198	178	178
query16	979	460	451	451
query17	919	743	640	640
query18	2458	467	364	364
query19	218	210	193	193
query20	135	128	132	128
query21	215	143	113	113
query22	13325	14171	14748	14171
query23	16234	15827	15486	15486
query23_1	15738	15725	15360	15360
query24	7199	1647	1241	1241
query24_1	1264	1247	1243	1243
query25	590	479	397	397
query26	1239	272	148	148
query27	2784	480	303	303
query28	4492	1849	1834	1834
query29	823	566	474	474
query30	294	233	188	188
query31	1019	970	876	876
query32	93	74	76	74
query33	525	344	292	292
query34	928	875	515	515
query35	636	684	604	604
query36	1081	1169	1031	1031
query37	137	99	86	86
query38	2933	2971	2851	2851
query39	856	829	805	805
query39_1	793	784	789	784
query40	247	150	133	133
query41	64	60	62	60
query42	263	259	258	258
query43	239	247	221	221
query44 query45	199	190	185	185
query46	874	980	614	614
query47	2111	2142	2017	2017
query48	305	315	232	232
query49	635	462	382	382
query50	698	278	214	214
query51	4084	4089	4044	4044
query52	262	266	257	257
query53	303	340	292	292
query54	311	276	275	275
query55	94	86	86	86
query56	310	329	324	324
query57	1946	1811	1692	1692
query58	289	278	266	266
query59	2800	2962	2748	2748
query60	347	346	342	342
query61	159	155	180	155
query62	642	592	540	540
query63	316	294	283	283
query64	5079	1296	1032	1032
query65 query66	1466	457	359	359
query67	24299	24379	24199	24199
query68 query69	409	317	287	287
query70	976	961	920	920
query71	344	308	312	308
query72	2785	2674	2454	2454
query73	548	546	317	317
query74	9591	9555	9443	9443
query75	2878	2809	2476	2476
query76	2266	1047	687	687
query77	377	380	317	317
query78	10883	11099	10443	10443
query79	1134	772	563	563
query80	951	629	557	557
query81	516	269	234	234
query82	1330	160	123	123
query83	342	264	256	256
query84	294	117	100	100
query85	888	490	446	446
query86	403	297	299	297
query87	3129	3140	3121	3121
query88	3540	2663	2651	2651
query89	431	366	351	351
query90	1922	187	172	172
query91	172	185	144	144
query92	76	77	71	71
query93	930	823	496	496
query94	523	324	282	282
query95	586	349	327	327
query96	644	508	236	236
query97	2444	2512	2403	2403
query98	236	225	223	223
query99	1000	1012	923	923
Total cold run time: 248496 ms
Total hot run time: 168708 ms

hello-stephen · 2026-03-18T12:13:51Z

FE UT Coverage Report

Increment line coverage 0.00% (0/1) 🎉
Increment coverage report
Complete coverage report

JNSimba · 2026-03-19T02:05:50Z

run external

JNSimba · 2026-03-19T03:07:53Z

/review

Copilot

Pull request overview

Fixes an intermittent Postgres CDC failure where the first DML after a task restart could be silently dropped due to Debezium’s WalPositionLocator behavior in pgoutput non-streaming mode.

Changes:

Override extractBinlogStateOffset() in PostgresSourceReader to remove lsn_proc/lsn_commit keys before Debezium consumes the offset.
Clarify a log message in StreamingMultiTblTask to accurately reflect “timeout reason”.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 3 comments.

File	Description
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/postgres/PostgresSourceReader.java	Adjusts Postgres offset passed to Debezium to avoid `WalPositionLocator` incorrectly filtering DML on restart.
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTask.java	Fixes misleading error log text for timeout-reason retrieval failures.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

You can also share your feedback on Copilot code review. Take the survey.

github-actions · 2026-03-19T03:14:53Z

Code Review Summary

PR: fix fix postgres DML silently dropped on task restart

Overview

This PR fixes a bug where the first DML of a new Postgres streaming transaction is intermittently dropped (10-20% failure rate) on task restart, with no error logged. The fix strips lsn_proc and lsn_commit keys from the persisted binlog state offset so that Debezium's WalPositionLocator skips its "find phase" entirely, preventing incorrect filtering of DML events. This is a workaround for FLINK-39265.

Critical Checkpoint Conclusions

Goal / Correctness: The fix correctly addresses the root cause. In pgoutput non-streaming mode (proto_version=1, Debezium 1.9.x), BEGIN and DML messages share the same XLogData.data_start. When begin_lsn equals the previous commit_lsn, the find phase incorrectly marks DMLs as already-processed. Stripping lsn_proc/lsn_commit sets lastCommitStoredLsn=null in WalPositionLocator, causing immediate find-phase exit. Verified that all downstream code paths (Flink CDC PostgresOffset, Debezium Loader.load(), replication slot flushing) gracefully handle the absence of these keys.
Modification size / clarity: Very small and focused — a single method override (7 lines of code) plus an excellent Javadoc comment explaining the rationale. The log message fix in StreamingMultiTblTask.java is a trivial correctness improvement. No issues.
Concurrency: No concerns. extractBinlogStateOffset operates on a locally created HashMap returned by super.extractBinlogStateOffset(). No shared state is modified.
Lifecycle management: No new lifecycle concerns introduced.
Configuration items: None added. N/A.
Incompatible changes / rolling upgrade: The persisted offset format changes (two keys stripped). Verified safe: (a) Debezium's PostgresOffsetContext.Loader.load() falls back to the main lsn value when lsn_proc/lsn_commit are absent; (b) existing offsets containing these keys continue to work since stripping only occurs on extraction; (c) the live streaming offsetContext continues to populate these keys normally during operation, so replication slot management is unaffected.
Parallel code paths: MySqlSourceReader has its own extractBinlogStateOffset override that does not need this fix (MySQL binlog offsets are not affected by this Postgres-specific WAL position issue). No action needed.
Test coverage: No tests exist for PostgresSourceReader or extractBinlogStateOffset. The entire fs_brokers/cdc_client module has minimal test coverage (only SchemaChangeHelperTest and SmallFileMgrTest). While testing this specific Debezium integration bug requires a running Postgres instance with specific WAL patterns (making unit testing difficult), a unit test verifying that extractBinlogStateOffset strips the expected keys from a mock offset map would be straightforward and valuable.
Observability: The thorough Javadoc comment with JIRA reference provides good documentation. No additional logging needed for this fix path.
Performance: No concerns — two HashMap.remove() calls on a small map.
Other issues: None found. The log message fix ("Send get task fail reason" -> "Send get task timeout reason") correctly matches the method name getTimeoutReason().

Verdict

No blocking issues. The fix is correct, well-documented, and safe for production. The only suggestion is to consider adding a unit test for the extractBinlogStateOffset override to prevent future regressions, but this is not a blocker.

JNSimba · 2026-03-19T04:32:16Z

run external

liaoxin01

LGTM

github-actions · 2026-03-19T13:22:05Z

PR approved by at least one committer and no changes requested.

github-actions · 2026-03-19T13:22:08Z

PR approved by anyone and no changes requested.

JNSimba · 2026-03-20T02:32:00Z

run buildall

JNSimba · 2026-03-20T02:39:37Z

run buildall

github-actions · 2026-03-20T02:46:12Z

PR approved by at least one committer and no changes requested.

doris-robot · 2026-03-20T03:08:41Z

TPC-H: Total hot run time: 27069 ms

machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpch-tools
Tpch sf100 test result on commit 51475c668fedc1b591be8f3e37a03d9943d2966b, data reload: false
------ Round 1 ----------------------------------
orders	Doris	NULL	NULL	0	0	0	NULL	0	NULL	NULL	2023-12-26 18:27:23	2023-12-26 18:42:55	NULL	utf-8	NULL	NULL ============================================
q1	13597	4425	4290	4290
q2	q3	3165	772	550	550
q4	876	358	251	251
q5	1783	1256	1099	1099
q6	189	189	177	177
q7	850	852	676	676
q8	1148	1472	1452	1452
q9	4854	4709	4706	4706
q10	2055	1905	1691	1691
q11	408	248	236	236
q12	506	625	471	471
q13	2596	2976	2180	2180
q14	235	229	221	221
q15	q16	723	739	677	677
q17	730	854	455	455
q18	5858	5396	5281	5281
q19	727	974	637	637
q20	504	500	380	380
q21	2476	1852	1390	1390
q22	339	293	249	249
Total cold run time: 43619 ms
Total hot run time: 27069 ms
----- Round 2, with runtime_filter_mode=off -----
orders	Doris	NULL	NULL	150000000	42	6422171781	NULL	22778155	NULL	NULL	2023-12-26 18:27:23	2023-12-26 18:42:55	NULL	utf-8	NULL	NULL ============================================
q1	4349	4381	4369	4369
q2	q3	3881	4291	3798	3798
q4	860	1163	756	756
q5	4035	4331	4295	4295
q6	181	173	145	145
q7	1725	1613	1504	1504
q8	2430	2655	2505	2505
q9	7288	7011	7118	7011
q10	3650	3917	3532	3532
q11	507	435	416	416
q12	474	580	444	444
q13	2558	3012	2215	2215
q14	264	287	262	262
q15	q16	706	773	711	711
q17	1100	1223	1296	1223
q18	6955	6685	6655	6655
q19	862	865	932	865
q20	2001	2124	1922	1922
q21	3876	3468	3318	3318
q22	448	439	376	376
Total cold run time: 48150 ms
Total hot run time: 46322 ms

doris-robot · 2026-03-20T03:19:24Z

TPC-DS: Total hot run time: 167189 ms

machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpcds-tools
TPC-DS sf100 test result on commit 51475c668fedc1b591be8f3e37a03d9943d2966b, data reload: false
query5	4329	633	515	515
query6	343	234	216	216
query7	4214	481	281	281
query8	343	240	236	236
query9	8748	2743	2730	2730
query10	494	403	332	332
query11	6958	5087	4867	4867
query12	182	134	129	129
query13	1286	463	355	355
query14	6287	3808	3460	3460
query14_1	2814	2814	2832	2814
query15	204	193	177	177
query16	987	495	460	460
query17	1129	741	642	642
query18	2725	457	363	363
query19	217	207	184	184
query20	142	130	127	127
query21	213	137	107	107
query22	13253	13266	13176	13176
query23	15872	15592	15250	15250
query23_1	15484	15421	15344	15344
query24	7102	1632	1227	1227
query24_1	1242	1244	1216	1216
query25	583	500	449	449
query26	1237	281	158	158
query27	2713	490	304	304
query28	4433	1868	1855	1855
query29	868	590	520	520
query30	308	233	196	196
query31	1064	973	883	883
query32	88	75	75	75
query33	522	343	301	301
query34	906	876	531	531
query35	659	692	607	607
query36	1050	1142	1001	1001
query37	147	107	91	91
query38	2993	2967	2936	2936
query39	866	854	806	806
query39_1	799	797	793	793
query40	235	156	138	138
query41	69	64	64	64
query42	271	259	263	259
query43	240	261	234	234
query44 query45	208	198	190	190
query46	879	977	620	620
query47	2170	2171	2064	2064
query48	319	320	228	228
query49	637	466	387	387
query50	691	283	217	217
query51	4072	4021	4042	4021
query52	269	267	264	264
query53	293	334	286	286
query54	307	274	274	274
query55	90	91	89	89
query56	338	339	350	339
query57	1932	1875	1703	1703
query58	288	283	274	274
query59	2782	2950	2721	2721
query60	357	341	326	326
query61	158	189	155	155
query62	633	598	537	537
query63	313	291	277	277
query64	4959	1305	1004	1004
query65 query66	1402	461	359	359
query67	24209	24260	24089	24089
query68 query69	414	318	286	286
query70	945	946	955	946
query71	344	302	299	299
query72	2893	2662	2472	2472
query73	554	547	317	317
query74	9601	9547	9414	9414
query75	2873	2761	2451	2451
query76	2285	1039	676	676
query77	373	394	300	300
query78	10976	11046	10452	10452
query79	2471	766	599	599
query80	1760	636	567	567
query81	552	263	225	225
query82	992	151	120	120
query83	335	271	249	249
query84	252	120	100	100
query85	922	502	450	450
query86	427	312	298	298
query87	3164	3078	3019	3019
query88	3650	2689	2675	2675
query89	436	370	348	348
query90	1972	178	172	172
query91	165	166	137	137
query92	83	86	76	76
query93	1046	819	501	501
query94	646	324	290	290
query95	584	404	319	319
query96	643	516	227	227
query97	2495	2512	2407	2407
query98	240	226	229	226
query99	1008	1001	894	894
Total cold run time: 251174 ms
Total hot run time: 167189 ms

hello-stephen · 2026-03-20T03:58:02Z

FE UT Coverage Report

Increment line coverage 0.00% (0/1) 🎉
Increment coverage report
Complete coverage report

smallhibiscus

LGTM

…#61481) ### What problem does this PR solve? #### Problem When a streaming job restarts a task, the first DML of the new transaction is occasionally silently dropped (10-20% failure rate). The affected record never appears in the Doris target table, with no error logged — only "identified as already processed" in cdc-client.log. #### Root Cause debezium 1.9.x hardcodes `proto_version=1` (non-streaming pgoutput) for all PG versions. In non-streaming mode, the walsender batches all changes of a transaction and sends them after COMMIT, and all messages (BEGIN + DML) share the same `XLogData.data_start` = the transaction's `begin_lsn`. When this `begin_lsn` equals the previous transaction's `commit_lsn` (i.e. the two transactions are adjacent in WAL with no other writes between them), `WalPositionLocator` behaves incorrectly: 1. **Find phase**: `COMMIT(T1)` at `lsn=Y` sets `storeLsnAfterLastEventStoredLsn=true`. `BEGIN(T2)` and `INSERT(T2)` both have `lsn=Y`, so they keep returning `Optional.empty()`. Only `COMMIT(T2)` at `lsn=Z` sets `startStreamingLsn=Z`, with `lsnSeen={Y, Z}`. 2. **Actual streaming**: `INSERT(T2)` arrives with `lastReceiveLsn=Y`. `skipMessage(Y)`: `Y ∈ lsnSeen` and `Y ≠ startStreamingLsn(Z)` → filtered. The bug is intermittent because it only triggers when no other WAL activity (autovacuum, other connections) occurs between the two transactions. #### Fix Override `extractBinlogStateOffset()` in `PostgresSourceReader` to strip `lsn_proc` and `lsn_commit` from the offset before it is passed to debezium. This constructs `WalPositionLocator(lastCommitStoredLsn=null, lsn=Y)`, which causes the find phase to exit immediately at the first received message (`startStreamingLsn=Y`). In actual streaming, `COMMIT(T1)` triggers switch-off (`lastReceiveLsn=Y = startStreamingLsn`), and all subsequent messages including `INSERT(T2)` pass through. See https://issues.apache.org/jira/browse/FLINK-39265. #### Test Run `test_streaming_postgres_job` multiple times. Before this fix the 'Apache' assertion fails ~10-20% of the time; after this fix it passes consistently.

…task restart #61481 (#61564) Cherry-picked from #61481 Co-authored-by: wudi <wudi@selectdb.com>

…#61481) ### What problem does this PR solve? #### Problem When a streaming job restarts a task, the first DML of the new transaction is occasionally silently dropped (10-20% failure rate). The affected record never appears in the Doris target table, with no error logged — only "identified as already processed" in cdc-client.log. #### Root Cause debezium 1.9.x hardcodes `proto_version=1` (non-streaming pgoutput) for all PG versions. In non-streaming mode, the walsender batches all changes of a transaction and sends them after COMMIT, and all messages (BEGIN + DML) share the same `XLogData.data_start` = the transaction's `begin_lsn`. When this `begin_lsn` equals the previous transaction's `commit_lsn` (i.e. the two transactions are adjacent in WAL with no other writes between them), `WalPositionLocator` behaves incorrectly: 1. **Find phase**: `COMMIT(T1)` at `lsn=Y` sets `storeLsnAfterLastEventStoredLsn=true`. `BEGIN(T2)` and `INSERT(T2)` both have `lsn=Y`, so they keep returning `Optional.empty()`. Only `COMMIT(T2)` at `lsn=Z` sets `startStreamingLsn=Z`, with `lsnSeen={Y, Z}`. 2. **Actual streaming**: `INSERT(T2)` arrives with `lastReceiveLsn=Y`. `skipMessage(Y)`: `Y ∈ lsnSeen` and `Y ≠ startStreamingLsn(Z)` → filtered. The bug is intermittent because it only triggers when no other WAL activity (autovacuum, other connections) occurs between the two transactions. #### Fix Override `extractBinlogStateOffset()` in `PostgresSourceReader` to strip `lsn_proc` and `lsn_commit` from the offset before it is passed to debezium. This constructs `WalPositionLocator(lastCommitStoredLsn=null, lsn=Y)`, which causes the find phase to exit immediately at the first received message (`startStreamingLsn=Y`). In actual streaming, `COMMIT(T1)` triggers switch-off (`lastReceiveLsn=Y = startStreamingLsn`), and all subsequent messages including `INSERT(T2)` pass through. See https://issues.apache.org/jira/browse/FLINK-39265. #### Test Run `test_streaming_postgres_job` multiple times. Before this fix the 'Apache' assertion fails ~10-20% of the time; after this fix it passes consistently.

…apache#61481) ### What problem does this PR solve? #### Problem When a streaming job restarts a task, the first DML of the new transaction is occasionally silently dropped (10-20% failure rate). The affected record never appears in the Doris target table, with no error logged — only "identified as already processed" in cdc-client.log. #### Root Cause debezium 1.9.x hardcodes `proto_version=1` (non-streaming pgoutput) for all PG versions. In non-streaming mode, the walsender batches all changes of a transaction and sends them after COMMIT, and all messages (BEGIN + DML) share the same `XLogData.data_start` = the transaction's `begin_lsn`. When this `begin_lsn` equals the previous transaction's `commit_lsn` (i.e. the two transactions are adjacent in WAL with no other writes between them), `WalPositionLocator` behaves incorrectly: 1. **Find phase**: `COMMIT(T1)` at `lsn=Y` sets `storeLsnAfterLastEventStoredLsn=true`. `BEGIN(T2)` and `INSERT(T2)` both have `lsn=Y`, so they keep returning `Optional.empty()`. Only `COMMIT(T2)` at `lsn=Z` sets `startStreamingLsn=Z`, with `lsnSeen={Y, Z}`. 2. **Actual streaming**: `INSERT(T2)` arrives with `lastReceiveLsn=Y`. `skipMessage(Y)`: `Y ∈ lsnSeen` and `Y ≠ startStreamingLsn(Z)` → filtered. The bug is intermittent because it only triggers when no other WAL activity (autovacuum, other connections) occurs between the two transactions. #### Fix Override `extractBinlogStateOffset()` in `PostgresSourceReader` to strip `lsn_proc` and `lsn_commit` from the offset before it is passed to debezium. This constructs `WalPositionLocator(lastCommitStoredLsn=null, lsn=Y)`, which causes the find phase to exit immediately at the first received message (`startStreamingLsn=Y`). In actual streaming, `COMMIT(T1)` triggers switch-off (`lastReceiveLsn=Y = startStreamingLsn`), and all subsequent messages including `INSERT(T2)` pass through. See https://issues.apache.org/jira/browse/FLINK-39265. #### Test Run `test_streaming_postgres_job` multiple times. Before this fix the 'Apache' assertion fails ~10-20% of the time; after this fix it passes consistently.

…task restart #61481 (#61855) Cherry-picked from #61481 Co-authored-by: wudi <wudi@selectdb.com>

fix
7bb9685

fix
314e84b

JNSimba added the dev/4.1.x label Mar 19, 2026

JNSimba requested a review from Copilot March 19, 2026 03:07

CopilotAI reviewed Mar 19, 2026
View reviewed changes

liaoxin01 previously approved these changes Mar 19, 2026
View reviewed changes

github-actionsBot added approved Indicates a PR has been approved by one committer. reviewed labels Mar 19, 2026

CalvinKirs reviewed Mar 20, 2026
View reviewed changes

Comment thread...re/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTask.java Outdated

CalvinKirs requested changes Mar 20, 2026
View reviewed changes

github-actionsBot removed the approved Indicates a PR has been approved by one committer. label Mar 20, 2026

fix
51475c6

JNSimba dismissed liaoxin01’s stale review via 51475c6March 20, 2026 02:31

CalvinKirs approved these changes Mar 20, 2026
View reviewed changes

github-actionsBot added the approved Indicates a PR has been approved by one committer. label Mar 20, 2026

JNSimba marked this pull request as draft March 20, 2026 06:08

JNSimba marked this pull request as ready for review March 20, 2026 08:22

smallhibiscus approved these changes Mar 20, 2026
View reviewed changes

JNSimba merged commit ec12ab6 into apache:masterMar 20, 2026
30 of 32 checks passed

github-actionsBot mentioned this pull request Mar 20, 2026
branch-4.1: [fix](streamingjob) fix postgres DML silently dropped on task restart #61481 #61564
Merged

yiguolei pushed a commit that referenced this pull request Mar 21, 2026
branch-4.1: [fix](streamingjob) fix postgres DML silently dropped on …
004d740
…task restart #61481 (#61564) Cherry-picked from #61481 Co-authored-by: wudi <wudi@selectdb.com>

yiguolei added dev/4.1.0-merged and removed dev/4.1.x labels Mar 21, 2026

JNSimba added the dev/4.0.x label Mar 30, 2026

github-actionsBot mentioned this pull request Mar 30, 2026
branch-4.0: [fix](streamingjob) fix postgres DML silently dropped on task restart #61481 #61855
Merged

yiguolei pushed a commit that referenced this pull request Apr 7, 2026
branch-4.0: [fix](streamingjob) fix postgres DML silently dropped on …
4da1251
…task restart #61481 (#61855) Cherry-picked from #61481 Co-authored-by: wudi <wudi@selectdb.com>

yiguolei added dev/4.0.6-merged and removed dev/4.0.x labels Apr 7, 2026

JNSimba mentioned this pull request May 11, 2026
[fix](streaming-job) start counting task max interval after the first record is received #63141
Merged
3 tasks

morningman mentioned this pull request Jun 3, 2026
4.0.6 Release Notes #64065
Closed

Uh oh!

Conversation

JNSimba commented Mar 18, 2026

What problem does this PR solve?

Problem

Root Cause

Fix

Test

Release note

Check List (For Author)

Check List (For Reviewer who merge this PR)

Uh oh!

JNSimba commented Mar 18, 2026

Uh oh!

Thearas commented Mar 18, 2026

Uh oh!

JNSimba commented Mar 18, 2026

Uh oh!

doris-robot commented Mar 18, 2026

Uh oh!

doris-robot commented Mar 18, 2026

Uh oh!

hello-stephen commented Mar 18, 2026

FE UT Coverage Report

Uh oh!

JNSimba commented Mar 19, 2026

Uh oh!

JNSimba commented Mar 19, 2026

Uh oh!

CopilotAI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

github-actionsBot commented Mar 19, 2026

Code Review Summary

Overview

Critical Checkpoint Conclusions

Verdict

Uh oh!

JNSimba commented Mar 19, 2026

Uh oh!

liaoxin01 left a comment

Choose a reason for hiding this comment

Uh oh!

github-actionsBot commented Mar 19, 2026

Uh oh!

github-actionsBot commented Mar 19, 2026

Uh oh!

Uh oh!

JNSimba commented Mar 20, 2026

Uh oh!

JNSimba commented Mar 20, 2026

Uh oh!

github-actionsBot commented Mar 20, 2026

Uh oh!

doris-robot commented Mar 20, 2026

Uh oh!

doris-robot commented Mar 20, 2026

Uh oh!

hello-stephen commented Mar 20, 2026

FE UT Coverage Report

Uh oh!

smallhibiscus left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

9 participants