Skip to content

Commit cff6604

Browse files
leonardBangclaude
leonardBang
and
claude
committed
[test][pipeline-e2e] Improve OracleE2eITCase customer event matching
Match customer inserts by stable event fragments instead of exact Oracle ID rendering so the E2E stays robust across CI environments. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
1 parent 077f638 commit cff6604

1 file changed

Lines changed: 77 additions & 84 deletions

File tree

‎flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/OracleE2eITCase.java‎

Lines changed: 77 additions & 84 deletions
Original file line numberDiff line numberDiff line change
@@ -174,48 +174,27 @@ void testSyncWholeDatabase() throws Exception {
174174
waitUntilAnySpecificEvent(
175175
"CreateTableEvent{tableId=DEBEZIUM.CUSTOMERS, schema=columns={`ID` BIGINT NOT NULL,`NAME` VARCHAR(255) NOT NULL,`ADDRESS` VARCHAR(1024),`PHONE_NUMBER` VARCHAR(512)}, primaryKeys=ID, options=()}",
176176
"CreateTableEvent{tableId=DEBEZIUM.CUSTOMERS, schema=columns={`ID` DECIMAL(38, 0) NOT NULL,`NAME` VARCHAR(255) NOT NULL,`ADDRESS` VARCHAR(1024),`PHONE_NUMBER` VARCHAR(512)}, primaryKeys=ID, options=()}");
177-
waitUntilSpecificEvent(
178-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[101, user_1, Shanghai, 123567891234], op=INSERT, meta=()}");
179-
waitUntilSpecificEvent(
180-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[109, user_4, Shanghai, 123567891234], op=INSERT, meta=()}");
181-
waitUntilSpecificEvent(
182-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[103, user_3, Shanghai, 123567891234], op=INSERT, meta=()}");
183-
waitUntilSpecificEvent(
184-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[102, user_2, Shanghai, 123567891234], op=INSERT, meta=()}");
185-
waitUntilSpecificEvent(
186-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[1012, user_13, Shanghai, 123567891234], op=INSERT, meta=()}");
187-
waitUntilSpecificEvent(
188-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[1013, user_14, Shanghai, 123567891234], op=INSERT, meta=()}");
189-
waitUntilSpecificEvent(
190-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[1010, user_11, Shanghai, 123567891234], op=INSERT, meta=()}");
191-
waitUntilSpecificEvent(
192-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[1011, user_12, Shanghai, 123567891234], op=INSERT, meta=()}");
193-
waitUntilSpecificEvent(
194-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[2000, user_21, Shanghai, 123567891234], op=INSERT, meta=()}");
195-
waitUntilSpecificEvent(
196-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[1009, user_10, Shanghai, 123567891234], op=INSERT, meta=()}");
197-
waitUntilSpecificEvent(
198-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[111, user_6, Shanghai, 123567891234], op=INSERT, meta=()}");
199-
waitUntilSpecificEvent(
200-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[110, user_5, Shanghai, 123567891234], op=INSERT, meta=()}");
201-
waitUntilSpecificEvent(
202-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[123, user_9, Shanghai, 123567891234], op=INSERT, meta=()}");
203-
waitUntilSpecificEvent(
204-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[1018, user_19, Shanghai, 123567891234], op=INSERT, meta=()}");
205-
waitUntilSpecificEvent(
206-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[1019, user_20, Shanghai, 123567891234], op=INSERT, meta=()}");
207-
waitUntilSpecificEvent(
208-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[121, user_8, Shanghai, 123567891234], op=INSERT, meta=()}");
209-
waitUntilSpecificEvent(
210-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[1016, user_17, Shanghai, 123567891234], op=INSERT, meta=()}");
211-
waitUntilSpecificEvent(
212-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[1017, user_18, Shanghai, 123567891234], op=INSERT, meta=()}");
213-
waitUntilSpecificEvent(
214-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[1014, user_15, Shanghai, 123567891234], op=INSERT, meta=()}");
215-
waitUntilSpecificEvent(
216-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[118, user_7, Shanghai, 123567891234], op=INSERT, meta=()}");
217-
waitUntilSpecificEvent(
218-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[], after=[1015, user_16, Shanghai, 123567891234], op=INSERT, meta=()}");
177+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_1");
178+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_4");
179+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_3");
180+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_2");
181+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_13");
182+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_14");
183+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_11");
184+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_12");
185+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_21");
186+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_10");
187+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_6");
188+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_5");
189+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_9");
190+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_19");
191+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_20");
192+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_8");
193+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_17");
194+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_18");
195+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_15");
196+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_7");
197+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS", "user_16");
219198

220199
stat.execute(
221200
"UPDATE DEBEZIUM.PRODUCTS SET DESCRIPTION='18oz carpenter hammer' WHERE ID=106 ");
@@ -229,48 +208,27 @@ void testSyncWholeDatabase() throws Exception {
229208
waitUntilAnySpecificEvent(
230209
"CreateTableEvent{tableId=DEBEZIUM.CUSTOMERS_1, schema=columns={`ID` BIGINT NOT NULL,`NAME` VARCHAR(255) NOT NULL,`ADDRESS` VARCHAR(1024),`PHONE_NUMBER` VARCHAR(512)}, primaryKeys=ID, options=()}",
231210
"CreateTableEvent{tableId=DEBEZIUM.CUSTOMERS_1, schema=columns={`ID` DECIMAL(38, 0) NOT NULL,`NAME` VARCHAR(255) NOT NULL,`ADDRESS` VARCHAR(1024),`PHONE_NUMBER` VARCHAR(512)}, primaryKeys=ID, options=()}");
232-
waitUntilSpecificEvent(
233-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[1009, user_10, Shanghai, 123567891234], op=INSERT, meta=()}");
234-
waitUntilSpecificEvent(
235-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[2000, user_21, Shanghai, 123567891234], op=INSERT, meta=()}");
236-
waitUntilSpecificEvent(
237-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[110, user_5, Shanghai, 123567891234], op=INSERT, meta=()}");
238-
waitUntilSpecificEvent(
239-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[1013, user_14, Shanghai, 123567891234], op=INSERT, meta=()}");
240-
waitUntilSpecificEvent(
241-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[111, user_6, Shanghai, 123567891234], op=INSERT, meta=()}");
242-
waitUntilSpecificEvent(
243-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[1012, user_13, Shanghai, 123567891234], op=INSERT, meta=()}");
244-
waitUntilSpecificEvent(
245-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[1011, user_12, Shanghai, 123567891234], op=INSERT, meta=()}");
246-
waitUntilSpecificEvent(
247-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[1010, user_11, Shanghai, 123567891234], op=INSERT, meta=()}");
248-
waitUntilSpecificEvent(
249-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[109, user_4, Shanghai, 123567891234], op=INSERT, meta=()}");
250-
waitUntilSpecificEvent(
251-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[102, user_2, Shanghai, 123567891234], op=INSERT, meta=()}");
252-
waitUntilSpecificEvent(
253-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[103, user_3, Shanghai, 123567891234], op=INSERT, meta=()}");
254-
waitUntilSpecificEvent(
255-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[101, user_1, Shanghai, 123567891234], op=INSERT, meta=()}");
256-
waitUntilSpecificEvent(
257-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[1017, user_18, Shanghai, 123567891234], op=INSERT, meta=()}");
258-
waitUntilSpecificEvent(
259-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[123, user_9, Shanghai, 123567891234], op=INSERT, meta=()}");
260-
waitUntilSpecificEvent(
261-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[1016, user_17, Shanghai, 123567891234], op=INSERT, meta=()}");
262-
waitUntilSpecificEvent(
263-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[1015, user_16, Shanghai, 123567891234], op=INSERT, meta=()}");
264-
waitUntilSpecificEvent(
265-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[1014, user_15, Shanghai, 123567891234], op=INSERT, meta=()}");
266-
waitUntilSpecificEvent(
267-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[118, user_7, Shanghai, 123567891234], op=INSERT, meta=()}");
268-
waitUntilSpecificEvent(
269-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[1019, user_20, Shanghai, 123567891234], op=INSERT, meta=()}");
270-
waitUntilSpecificEvent(
271-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[121, user_8, Shanghai, 123567891234], op=INSERT, meta=()}");
272-
waitUntilSpecificEvent(
273-
"DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[], after=[1018, user_19, Shanghai, 123567891234], op=INSERT, meta=()}");
211+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_10");
212+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_21");
213+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_5");
214+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_14");
215+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_6");
216+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_13");
217+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_12");
218+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_11");
219+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_4");
220+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_2");
221+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_3");
222+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_1");
223+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_18");
224+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_9");
225+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_17");
226+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_16");
227+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_15");
228+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_7");
229+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_20");
230+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_8");
231+
waitUntilCustomerInsert("DEBEZIUM.CUSTOMERS_1", "user_19");
274232
} catch (Exception e) {
275233
LOG.error("Update table for CDC failed.", e);
276234
throw new RuntimeException(e);
@@ -300,6 +258,41 @@ private void waitUntilAnySpecificEvent(String... events) throws Exception {
300258
+ taskManagerConsumer.toUtf8String());
301259
}
302260

261+
private void waitUntilLogLineContains(String... fragments) throws Exception {
262+
boolean result = false;
263+
long endTimeout = System.currentTimeMillis() + EVENT_WAITING_TIMEOUT.toMillis();
264+
while (System.currentTimeMillis() < endTimeout) {
265+
for (String line : taskManagerConsumer.toUtf8String().split("\n")) {
266+
boolean matched = true;
267+
for (String fragment : fragments) {
268+
if (!line.contains(fragment)) {
269+
matched = false;
270+
break;
271+
}
272+
}
273+
if (matched) {
274+
result = true;
275+
break;
276+
}
277+
}
278+
if (result) {
279+
return;
280+
}
281+
Thread.sleep(1000);
282+
}
283+
throw new TimeoutException(
284+
"failed to get log line fragments: "
285+
+ Arrays.toString(fragments)
286+
+ " from stdout: "
287+
+ taskManagerConsumer.toUtf8String());
288+
}
289+
290+
private void waitUntilCustomerInsert(String tableId, String userName) throws Exception {
291+
waitUntilLogLineContains(
292+
"DataChangeEvent{tableId=" + tableId + ", before=[], after=[",
293+
", " + userName + ", Shanghai, 123567891234], op=INSERT, meta=()}");
294+
}
295+
303296
private void initializeOracleTable(String sqlFile) {
304297
final String ddlFile = String.format("ddl/%s.sql", sqlFile);
305298
final URL ddlTestFile = OracleSourceTestBase.class.getClassLoader().getResource(ddlFile);

0 commit comments

Comments
 (0)