This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new e7c1e5e0f3 [flink] Report which required argument is missing in action
factories (#9059)
e7c1e5e0f3 is described below
commit e7c1e5e0f303a3b2e4aa3ca0fa3a1f844c78c551
Author: Eunbin Son <[email protected]>
AuthorDate: Fri Aug 7 14:15:20 2026 +0900
[flink] Report which required argument is missing in action factories
(#9059)
---
.../CreateTagFromTimestampActionFactory.java | 2 +-
.../CreateTagFromWatermarkActionFactory.java | 4 +-
.../flink/action/MergeIntoActionFactory.java | 2 +-
.../CreateTagFromTimestampActionFactoryTest.java | 70 ++++++++++++++++
.../CreateTagFromWatermarkActionFactoryTest.java | 89 ++++++++++++++++++++
.../flink/action/MergeIntoActionFactoryTest.java | 98 ++++++++++++++++++++++
6 files changed, 261 insertions(+), 4 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CreateTagFromTimestampActionFactory.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CreateTagFromTimestampActionFactory.java
index 658f3e03f0..639d5624bf 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CreateTagFromTimestampActionFactory.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CreateTagFromTimestampActionFactory.java
@@ -38,7 +38,7 @@ public class CreateTagFromTimestampActionFactory implements
ActionFactory {
@Override
public Optional<Action> create(MultipleParameterToolAdapter params) {
- Long timestamp = Long.parseLong(params.get(TIMESTAMP));
+ Long timestamp = Long.parseLong(params.getRequired(TIMESTAMP));
String timeRetained = params.get(TIME_RETAINED);
Map<String, String> catalogConfig = catalogConfigMap(params);
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CreateTagFromWatermarkActionFactory.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CreateTagFromWatermarkActionFactory.java
index 1f4e5b1f92..47ca8e461a 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CreateTagFromWatermarkActionFactory.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CreateTagFromWatermarkActionFactory.java
@@ -39,8 +39,8 @@ public class CreateTagFromWatermarkActionFactory implements
ActionFactory {
@Override
public Optional<Action> create(MultipleParameterToolAdapter params) {
- String tag = params.get(TAG);
- Long watermark = Long.parseLong(params.get(WATERMARK));
+ String tag = params.getRequired(TAG);
+ Long watermark = Long.parseLong(params.getRequired(WATERMARK));
String timeRetained = params.get(TIME_RETAINED);
Map<String, String> catalogConfig = catalogConfigMap(params);
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MergeIntoActionFactory.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MergeIntoActionFactory.java
index e39b6397c9..fab8865c50 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MergeIntoActionFactory.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MergeIntoActionFactory.java
@@ -80,7 +80,7 @@ public class MergeIntoActionFactory implements ActionFactory {
action.withMergeCondition(params.getRequired(ON));
List<String> actions =
- Arrays.stream(params.get(MERGE_ACTIONS).split(","))
+ Arrays.stream(params.getRequired(MERGE_ACTIONS).split(","))
.map(String::trim)
.collect(Collectors.toList());
if (actions.contains(MATCHED_UPSERT)) {
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CreateTagFromTimestampActionFactoryTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CreateTagFromTimestampActionFactoryTest.java
new file mode 100644
index 0000000000..a15091f7c6
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CreateTagFromTimestampActionFactoryTest.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.action;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link CreateTagFromTimestampActionFactory#create}. */
+public class CreateTagFromTimestampActionFactoryTest extends ActionITCaseBase {
+
+ @Test
+ public void testMissingTimestampReportsRequiredArgument() {
+ assertThatThrownBy(
+ () ->
+ createAction(
+ CreateTagFromTimestampAction.class,
+ "create_tag_from_timestamp",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--tag",
+ "tag_1"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Argument 'timestamp' is required");
+ }
+
+ @Test
+ public void testCreateWithTimestamp() {
+ assertThatCode(
+ () ->
+ assertThat(
+ createAction(
+
CreateTagFromTimestampAction.class,
+
"create_tag_from_timestamp",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--tag",
+ "tag_1",
+ "--timestamp",
+ "1000"))
+ .isNotNull())
+ .doesNotThrowAnyException();
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CreateTagFromWatermarkActionFactoryTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CreateTagFromWatermarkActionFactoryTest.java
new file mode 100644
index 0000000000..bac8b99076
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CreateTagFromWatermarkActionFactoryTest.java
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.action;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link CreateTagFromWatermarkActionFactory#create}. */
+public class CreateTagFromWatermarkActionFactoryTest extends ActionITCaseBase {
+
+ @Test
+ public void testMissingWatermarkReportsRequiredArgument() {
+ assertThatThrownBy(
+ () ->
+ createAction(
+ CreateTagFromWatermarkAction.class,
+ "create_tag_from_watermark",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--tag",
+ "tag_1"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Argument 'watermark' is required");
+ }
+
+ @Test
+ public void testMissingTagReportsRequiredArgument() {
+ assertThatThrownBy(
+ () ->
+ createAction(
+ CreateTagFromWatermarkAction.class,
+ "create_tag_from_watermark",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--watermark",
+ "1000"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Argument 'tag' is required");
+ }
+
+ @Test
+ public void testCreateWithTagAndWatermark() {
+ assertThatCode(
+ () ->
+ assertThat(
+ createAction(
+
CreateTagFromWatermarkAction.class,
+
"create_tag_from_watermark",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--tag",
+ "tag_1",
+ "--watermark",
+ "1000"))
+ .isNotNull())
+ .doesNotThrowAnyException();
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/MergeIntoActionFactoryTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/MergeIntoActionFactoryTest.java
new file mode 100644
index 0000000000..266b3eec65
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/MergeIntoActionFactoryTest.java
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.action;
+
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link MergeIntoActionFactory#create}. */
+public class MergeIntoActionFactoryTest extends ActionITCaseBase {
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ // merge_into loads the target table before parsing --merge_actions,
so it must exist and
+ // must have primary keys
+ DataType[] fieldTypes = new DataType[] {DataTypes.INT(),
DataTypes.STRING()};
+ RowType rowType = RowType.of(fieldTypes, new String[] {"k", "v"});
+ createFileStoreTable(
+ rowType,
+ Collections.emptyList(),
+ Collections.singletonList("k"),
+ Collections.emptyList(),
+ new HashMap<>());
+ }
+
+ @Test
+ public void testMissingMergeActionsReportsRequiredArgument() {
+ assertThatThrownBy(
+ () ->
+ createAction(
+ MergeIntoAction.class,
+ "merge_into",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--source_table",
+ "S",
+ "--on",
+ "T.k = S.k"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Argument 'merge_actions' is required");
+ }
+
+ @Test
+ public void testCreateWithMergeActions() {
+ assertThatCode(
+ () ->
+ assertThat(
+ createAction(
+ MergeIntoAction.class,
+ "merge_into",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--source_table",
+ "S",
+ "--on",
+ "T.k = S.k",
+ "--merge_actions",
+ "matched-upsert",
+ "--matched_upsert_set",
+ "v = S.v"))
+ .isNotNull())
+ .doesNotThrowAnyException();
+ }
+}
