From d411dff568a703944c50394e32efc71c551e95eb Mon Sep 17 00:00:00 2001 From: Eunbin Son Date: Thu, 6 Aug 2026 10:49:00 +0900 Subject: [PATCH] [flink] Report which required argument is missing in action factories Four required action arguments were read with params.get(...) and dereferenced immediately, so omitting one failed with a bare NullPointerException (merge_actions) or NumberFormatException: null (timestamp, watermark) that never named the missing argument. Read them with params.getRequired(...), like every sibling argument in the same methods, so the failure reports the argument name. Generated-by: Claude Code --- .../CreateTagFromTimestampActionFactory.java | 2 +- .../CreateTagFromWatermarkActionFactory.java | 4 +- .../flink/action/MergeIntoActionFactory.java | 2 +- ...eateTagFromTimestampActionFactoryTest.java | 70 +++++++++++++ ...eateTagFromWatermarkActionFactoryTest.java | 89 +++++++++++++++++ .../action/MergeIntoActionFactoryTest.java | 98 +++++++++++++++++++ 6 files changed, 261 insertions(+), 4 deletions(-) create mode 100644 paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CreateTagFromTimestampActionFactoryTest.java create mode 100644 paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CreateTagFromWatermarkActionFactoryTest.java create mode 100644 paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/MergeIntoActionFactoryTest.java 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 658f3e03f0cb..639d5624bf71 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 String identifier() { @Override public Optional create(MultipleParameterToolAdapter params) { - Long timestamp = Long.parseLong(params.get(TIMESTAMP)); + Long timestamp = Long.parseLong(params.getRequired(TIMESTAMP)); String timeRetained = params.get(TIME_RETAINED); Map 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 1f4e5b1f92b9..47ca8e461aae 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 String identifier() { @Override public Optional 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 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 e39b6397c9bc..fab8865c505f 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 Optional create(MultipleParameterToolAdapter params) { action.withMergeCondition(params.getRequired(ON)); List 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 000000000000..a15091f7c615 --- /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 000000000000..bac8b9907689 --- /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 000000000000..266b3eec65d9 --- /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(); + } +}