From c99341e16c0df3a6e4a784d9e5a834a88aa73ca6 Mon Sep 17 00:00:00 2001 From: lvyanquan Date: Wed, 27 Nov 2024 14:49:07 +0800 Subject: [PATCH] [FLINK-36524][pipeline-connector][paimon] 1) bump version to 0.9.0. 2) Forcibly add cache-enabled = false in catalog option to avoid using previous table schema. --- .../cdc/connectors/paimon/sink/v2/PaimonCommitter.java | 2 -- .../connectors/paimon/sink/PaimonMetadataApplierTest.java | 6 +++--- .../cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java | 2 -- 3 files changed, 3 insertions(+), 7 deletions(-) diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonCommitter.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonCommitter.java index aa012f6019c..03c0be6be15 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonCommitter.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonCommitter.java @@ -27,8 +27,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import javax.annotation.Nullable; - import java.io.IOException; import java.util.Collection; import java.util.Collections; diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplierTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplierTest.java index 08a88829d74..7b362ee0d40 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplierTest.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplierTest.java @@ -271,10 +271,10 @@ public void testCreateTableWithOptions(String metastore) Arrays.asList( new DataField(0, "col1", DataTypes.STRING().notNull()), new DataField(1, "col2", DataTypes.STRING()), - new DataField(2, "col3", DataTypes.STRING()), - new DataField(3, "col4", DataTypes.STRING()))); + new DataField(2, "col3", DataTypes.STRING().notNull()), + new DataField(3, "col4", DataTypes.STRING().notNull()))); Assertions.assertEquals(tableSchema, table.rowType()); - Assertions.assertEquals(Collections.singletonList("col1"), table.primaryKeys()); + Assertions.assertEquals(Arrays.asList("col1", "col3", "col4"), table.primaryKeys()); Assertions.assertEquals(Arrays.asList("col3", "col4"), table.partitionKeys()); Assertions.assertEquals("-1", table.options().get("bucket")); } diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java index cababd0e085..0dd60c2f7a4 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java @@ -128,8 +128,6 @@ private void initialize(String metastore) + "'hadoop-conf-dir'='%s', " + "'hive-conf-dir'='%s', " + "'cache-enabled'='false'" - + "'hive-conf-dir'='%s', " - + "'cache-enabled'='false' " + ")", warehouse, HADOOP_CONF_DIR, HIVE_CONF_DIR)); } else {