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 34bd5a16e8 [flink] Fix migrate actions crashing when parallelism is
not specified (#8611)
34bd5a16e8 is described below
commit 34bd5a16e8745babbfb7453d2dcdfdc7eb5d38ce
Author: Eunbin Son <[email protected]>
AuthorDate: Wed Jul 15 09:06:35 2026 +0900
[flink] Fix migrate actions crashing when parallelism is not specified
(#8611)
---
.../flink/action/MigrateDatabaseActionFactory.java | 6 +-
.../flink/action/MigrateTableActionFactory.java | 6 +-
.../action/MigrateDatabaseActionFactoryTest.java | 67 ++++++++++++++++++++++
.../action/MigrateTableActionFactoryTest.java | 67 ++++++++++++++++++++++
4 files changed, 142 insertions(+), 4 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateDatabaseActionFactory.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateDatabaseActionFactory.java
index 15a0e28f89..a95e21f37c 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateDatabaseActionFactory.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateDatabaseActionFactory.java
@@ -41,7 +41,8 @@ public class MigrateDatabaseActionFactory implements
ActionFactory {
String sourceHiveDatabase = params.get(DATABASE);
Map<String, String> catalogConfig = catalogConfigMap(params);
String tableConf = params.get(OPTIONS);
- Integer parallelism = Integer.parseInt(params.get(PARALLELISM));
+ Integer parallelism =
+ params.has(PARALLELISM) ?
Integer.parseInt(params.get(PARALLELISM)) : null;
MigrateDatabaseAction migrateDatabaseAction =
new MigrateDatabaseAction(
@@ -62,6 +63,7 @@ public class MigrateDatabaseActionFactory implements
ActionFactory {
+ "--source_type hive \\\n"
+ "--database <database_name> \\\n"
+ "[--catalog_conf <key>=<value] \\\n"
- + "[--options <key>=<value>,<key>=<value>,...]");
+ + "[--options <key>=<value>,<key>=<value>,...] \\\n"
+ + "[--parallelism <parallelism>]");
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateTableActionFactory.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateTableActionFactory.java
index b2c96795a4..0f1ecb82b8 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateTableActionFactory.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateTableActionFactory.java
@@ -41,7 +41,8 @@ public class MigrateTableActionFactory implements
ActionFactory {
String sourceHiveTable = params.get(TABLE);
Map<String, String> catalogConfig = catalogConfigMap(params);
String tableConf = params.get(OPTIONS);
- Integer parallelism = Integer.parseInt(params.get(PARALLELISM));
+ Integer parallelism =
+ params.has(PARALLELISM) ?
Integer.parseInt(params.get(PARALLELISM)) : null;
MigrateTableAction migrateTableAction =
new MigrateTableAction(
@@ -61,6 +62,7 @@ public class MigrateTableActionFactory implements
ActionFactory {
+ "--source_type hive \\\n"
+ "--table <database.table_name> \\\n"
+ "[--catalog_conf <key>=<value] \\\n"
- + "[--options <key>=<value>,<key>=<value>,...]");
+ + "[--options <key>=<value>,<key>=<value>,...] \\\n"
+ + "[--parallelism <parallelism>]");
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/MigrateDatabaseActionFactoryTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/MigrateDatabaseActionFactoryTest.java
new file mode 100644
index 0000000000..56df4b9f0b
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/MigrateDatabaseActionFactoryTest.java
@@ -0,0 +1,67 @@
+/*
+ * 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;
+
+/** Tests for {@link MigrateDatabaseActionFactory#create}. */
+public class MigrateDatabaseActionFactoryTest extends ActionITCaseBase {
+
+ @Test
+ public void testCreateWithoutParallelismDoesNotThrow() {
+ // --parallelism is optional; omitting it must not throw
NumberFormatException.
+ assertThatCode(
+ () ->
+ assertThat(
+ createAction(
+
MigrateDatabaseAction.class,
+ "migrate_database",
+ "--warehouse",
+ warehouse,
+ "--source_type",
+ "hive",
+ "--database",
+ "default"))
+ .isNotNull())
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ public void testCreateWithParallelism() {
+ assertThatCode(
+ () ->
+ assertThat(
+ createAction(
+
MigrateDatabaseAction.class,
+ "migrate_database",
+ "--warehouse",
+ warehouse,
+ "--source_type",
+ "hive",
+ "--database",
+ "default",
+ "--parallelism",
+ "4"))
+ .isNotNull())
+ .doesNotThrowAnyException();
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/MigrateTableActionFactoryTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/MigrateTableActionFactoryTest.java
new file mode 100644
index 0000000000..f2185299d0
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/MigrateTableActionFactoryTest.java
@@ -0,0 +1,67 @@
+/*
+ * 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;
+
+/** Tests for {@link MigrateTableActionFactory#create}. */
+public class MigrateTableActionFactoryTest extends ActionITCaseBase {
+
+ @Test
+ public void testCreateWithoutParallelismDoesNotThrow() {
+ // --parallelism is optional; omitting it must not throw
NumberFormatException.
+ assertThatCode(
+ () ->
+ assertThat(
+ createAction(
+
MigrateTableAction.class,
+ "migrate_table",
+ "--warehouse",
+ warehouse,
+ "--source_type",
+ "hive",
+ "--table",
+ "default.T"))
+ .isNotNull())
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ public void testCreateWithParallelism() {
+ assertThatCode(
+ () ->
+ assertThat(
+ createAction(
+
MigrateTableAction.class,
+ "migrate_table",
+ "--warehouse",
+ warehouse,
+ "--source_type",
+ "hive",
+ "--table",
+ "default.T",
+ "--parallelism",
+ "4"))
+ .isNotNull())
+ .doesNotThrowAnyException();
+ }
+}
