You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2024/03/15 04:01:46 UTC
(incubator-paimon) branch master updated: [hotfix] Fix unstable testRandomCdcEventsUnawareBucket (#3014)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new bf4e8a224 [hotfix] Fix unstable testRandomCdcEventsUnawareBucket (#3014)
bf4e8a224 is described below
commit bf4e8a224919f7007835d81b89e3169592cd86c1
Author: yuzelin <33...@users.noreply.github.com>
AuthorDate: Fri Mar 15 12:01:42 2024 +0800
[hotfix] Fix unstable testRandomCdcEventsUnawareBucket (#3014)
---
.../sink/cdc/FlinkCdcSyncDatabaseSinkITCase.java | 36 ++++++++++------
.../sink/cdc/FlinkCdcSyncTableSinkITCase.java | 27 ++++++++----
.../apache/paimon/flink/sink/cdc/TestTable.java | 50 +++++++---------------
3 files changed, 56 insertions(+), 57 deletions(-)
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkITCase.java
index 167d56b47..d3369cef3 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkITCase.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkITCase.java
@@ -21,6 +21,7 @@ package org.apache.paimon.flink.sink.cdc;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogUtils;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.flink.FlinkCatalogFactory;
import org.apache.paimon.flink.util.AbstractTestBase;
import org.apache.paimon.fs.FileIO;
@@ -28,12 +29,15 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
+import org.apache.paimon.reader.RecordReaderIterator;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.SchemaUtils;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.TableScan;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FailingFileIO;
import org.apache.paimon.utils.TraceableFileIO;
@@ -41,7 +45,6 @@ import org.apache.paimon.utils.TraceableFileIO;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.io.TempDir;
@@ -64,23 +67,20 @@ public class FlinkCdcSyncDatabaseSinkITCase extends AbstractTestBase {
@TempDir java.nio.file.Path tempDir;
- @Disabled
@Test
@Timeout(120)
public void testRandomCdcEvents() throws Exception {
innerTestRandomCdcEvents(() -> ThreadLocalRandom.current().nextInt(5) + 1, false);
}
- @Disabled
@Test
- @Timeout(180)
+ @Timeout(120)
public void testRandomCdcEventsDynamicBucket() throws Exception {
innerTestRandomCdcEvents(() -> -1, false);
}
- @Disabled
@Test
- @Timeout(180)
+ @Timeout(120)
public void testRandomCdcEventsUnawareBucket() throws Exception {
innerTestRandomCdcEvents(() -> -1, true);
}
@@ -177,17 +177,21 @@ public class FlinkCdcSyncDatabaseSinkITCase extends AbstractTestBase {
// enable failure when running jobs if needed
FailingFileIO.reset(failingName, 2, 10000);
- if (unawareBucketMode) {
- // there's a compact operator which won't terminate
- env.executeAsync();
- } else {
- env.execute();
- }
+ env.execute();
// no failure when checking results
FailingFileIO.reset(failingName, 0, 1);
- for (int i = 0; i < fileStoreTables.size(); i++) {
- testTables.get(i).assertResult(fileStoreTables.get(i));
+ for (int i = 0; i < numTables; i++) {
+ FileStoreTable table = fileStoreTables.get(i).copyWithLatestSchema();
+ SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location());
+ TableSchema schema = schemaManager.latest().get();
+
+ ReadBuilder readBuilder = table.newReadBuilder();
+ TableScan.Plan plan = readBuilder.newScan().plan();
+ try (RecordReaderIterator<InternalRow> it =
+ new RecordReaderIterator<>(readBuilder.newRead().createReader(plan))) {
+ testTables.get(i).assertResult(schema, it);
+ }
}
}
@@ -204,6 +208,10 @@ public class FlinkCdcSyncDatabaseSinkITCase extends AbstractTestBase {
conf.set(CoreOptions.DYNAMIC_BUCKET_TARGET_ROW_NUM, 100L);
conf.set(CoreOptions.WRITE_BUFFER_SIZE, new MemorySize(4096 * 3));
conf.set(CoreOptions.PAGE_SIZE, new MemorySize(4096));
+ // disable compaction for unaware bucket mode to avoid instability
+ if (primaryKeys.isEmpty() && numBucket == -1) {
+ conf.set(CoreOptions.WRITE_ONLY, true);
+ }
TableSchema tableSchema =
SchemaUtils.forceCommit(
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncTableSinkITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncTableSinkITCase.java
index a47561741..57a7604c0 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncTableSinkITCase.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncTableSinkITCase.java
@@ -22,6 +22,7 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogUtils;
import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.flink.FlinkCatalogFactory;
import org.apache.paimon.flink.util.AbstractTestBase;
import org.apache.paimon.fs.FileIO;
@@ -29,12 +30,15 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
+import org.apache.paimon.reader.RecordReaderIterator;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.SchemaUtils;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.TableScan;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FailingFileIO;
import org.apache.paimon.utils.TraceableFileIO;
@@ -168,18 +172,21 @@ public class FlinkCdcSyncTableSinkITCase extends AbstractTestBase {
// enable failure when running jobs if needed
FailingFileIO.reset(failingName, 10, 10000);
-
- if (unawareBucketMode) {
- // there's a compact operator which won't terminate
- env.executeAsync();
- } else {
- env.execute();
- }
+ env.execute();
// no failure when checking results
FailingFileIO.reset(failingName, 0, 1);
- testTable.assertResult(table);
+ table = table.copyWithLatestSchema();
+ SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location());
+ TableSchema schema = schemaManager.latest().get();
+
+ ReadBuilder readBuilder = table.newReadBuilder();
+ TableScan.Plan plan = readBuilder.newScan().plan();
+ try (RecordReaderIterator<InternalRow> it =
+ new RecordReaderIterator<>(readBuilder.newRead().createReader(plan))) {
+ testTable.assertResult(schema, it);
+ }
}
private FileStoreTable createFileStoreTable(
@@ -195,6 +202,10 @@ public class FlinkCdcSyncTableSinkITCase extends AbstractTestBase {
conf.set(CoreOptions.DYNAMIC_BUCKET_TARGET_ROW_NUM, 100L);
conf.set(CoreOptions.WRITE_BUFFER_SIZE, new MemorySize(4096 * 3));
conf.set(CoreOptions.PAGE_SIZE, new MemorySize(4096));
+ // disable compaction for unaware bucket mode to avoid instability
+ if (primaryKeys.isEmpty() && numBucket == -1) {
+ conf.set(CoreOptions.WRITE_ONLY, true);
+ }
TableSchema tableSchema =
SchemaUtils.forceCommit(
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/TestTable.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/TestTable.java
index beb8786d1..525a05096 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/TestTable.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/TestTable.java
@@ -19,11 +19,7 @@
package org.apache.paimon.flink.sink.cdc;
import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.reader.RecordReaderIterator;
import org.apache.paimon.schema.TableSchema;
-import org.apache.paimon.table.FileStoreTable;
-import org.apache.paimon.table.source.ReadBuilder;
-import org.apache.paimon.table.source.TableScan;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
@@ -33,6 +29,7 @@ import org.apache.paimon.types.RowType;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@@ -188,40 +185,23 @@ public class TestTable {
return events;
}
- public void assertResult(FileStoreTable table) throws Exception {
+ public void assertResult(TableSchema schema, Iterator<InternalRow> it) {
Map<Integer, Map<String, String>> actual = new HashMap<>();
- while (true) {
- actual.clear();
- table = table.copyWithLatestSchema();
- TableSchema schema = table.schema();
- ReadBuilder readBuilder = table.newReadBuilder();
- TableScan.Plan plan = readBuilder.newScan().plan();
-
- try (RecordReaderIterator<InternalRow> it =
- new RecordReaderIterator<>(readBuilder.newRead().createReader(plan))) {
- while (it.hasNext()) {
- InternalRow row = it.next();
- Map<String, String> fields = new HashMap<>();
- for (int i = 0; i < schema.fieldNames().size(); i++) {
- if (!row.isNullAt(i)) {
- fields.put(
- schema.fieldNames().get(i),
- String.valueOf(
- schema.fields().get(i).type().equals(DataTypes.BIGINT())
- ? row.getLong(i)
- : row.getInt(i)));
- }
- }
- actual.put(Integer.valueOf(fields.get("k")), fields);
+ while (it.hasNext()) {
+ InternalRow row = it.next();
+ Map<String, String> fields = new HashMap<>();
+ for (int i = 0; i < schema.fieldNames().size(); i++) {
+ if (!row.isNullAt(i)) {
+ fields.put(
+ schema.fieldNames().get(i),
+ String.valueOf(
+ schema.fields().get(i).type().equals(DataTypes.BIGINT())
+ ? row.getLong(i)
+ : row.getInt(i)));
}
}
-
- if (actual.size() == expected.size()) {
- assertThat(actual).isEqualTo(expected);
- break;
- }
-
- Thread.sleep(500);
+ actual.put(Integer.valueOf(fields.get("k")), fields);
}
+ assertThat(actual).isEqualTo(expected);
}
}