[FLINK-35939] Do not set empty config values via ConfigUtils#encodeCo…

…llectionToConfig
snuyanzin · Aug 19, 2024 · e4abd06 · e4abd06
1 parent adb8368
commit e4abd06
Show file tree

Hide file tree

Showing 4 changed files with 14 additions and 26 deletions.
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigUtils.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigUtils.java
@@ -100,7 +100,9 @@ public static <IN, OUT> void encodeCollectionToConfig(
                         .filter(Objects::nonNull)
                         .collect(Collectors.toCollection(ArrayList::new));
 
-        configuration.set(key, encodedOption);
+        if (!encodedOption.isEmpty()) {
+            configuration.set(key, encodedOption);
+        }
     }
 
     /**

diff --git a/flink-core/src/test/java/org/apache/flink/configuration/ConfigUtilsTest.java b/flink-core/src/test/java/org/apache/flink/configuration/ConfigUtilsTest.java
@@ -95,13 +95,13 @@ void nullArrayPutsNothingInConfig() {
     }
 
     @Test
-    void emptyCollectionPutsEmptyValueInConfig() {
+    void emptyCollectionPutsNothingInConfig() {
         final Configuration configurationUnderTest = new Configuration();
         ConfigUtils.encodeCollectionToConfig(
                 configurationUnderTest, TEST_OPTION, Collections.emptyList(), Object::toString);
 
         final List<String> recovered = configurationUnderTest.get(TEST_OPTION);
-        assertThat(recovered).isEmpty();
+        assertThat(recovered).isNull();
 
         final List<Integer> recoveredList =
                 ConfigUtils.decodeListFromConfig(
@@ -110,13 +110,13 @@ void emptyCollectionPutsEmptyValueInConfig() {
     }
 
     @Test
-    void emptyArrayPutsEmptyValueInConfig() {
+    void emptyArrayPutsNothingInConfig() {
         final Configuration configurationUnderTest = new Configuration();
         ConfigUtils.encodeArrayToConfig(
                 configurationUnderTest, TEST_OPTION, new Integer[5], Object::toString);
 
         final List<String> recovered = configurationUnderTest.get(TEST_OPTION);
-        assertThat(recovered).isEmpty();
+        assertThat(recovered).isNull();
 
         final List<Integer> recoveredList =
                 ConfigUtils.decodeListFromConfig(

diff --git a/flink-table/flink-sql-client/src/test/resources/sql/set.q b/flink-table/flink-sql-client/src/test/resources/sql/set.q
@@ -83,14 +83,12 @@ set;
 | execution.state-recovery.ignore-unclaimed-state |     false |
 |                                execution.target |    remote |
 |                          jobmanager.rpc.address | $VAR_JOBMANAGER_RPC_ADDRESS |
-|                             pipeline.classpaths |        [] |
-|                                   pipeline.jars |        [] |
 |                                       rest.port |     $VAR_REST_PORT |
 |              sql-client.display.print-time-cost |     false |
 |                sql-client.execution.result-mode |   tableau |
 |                table.exec.legacy-cast-behaviour |  DISABLED |
 +-------------------------------------------------+-----------+
-12 rows in set
+10 rows in set
 !ok
 
 # reset the configuration
@@ -108,11 +106,9 @@ set;
 | execution.state-recovery.ignore-unclaimed-state |     false |
 |                                execution.target |    remote |
 |                          jobmanager.rpc.address | $VAR_JOBMANAGER_RPC_ADDRESS |
-|                             pipeline.classpaths |        [] |
-|                                   pipeline.jars |        [] |
 |                                       rest.port |     $VAR_REST_PORT |
 +-------------------------------------------------+-----------+
-9 rows in set
+7 rows in set
 !ok
 
 # should fail because default dialect doesn't support hive dialect
@@ -149,12 +145,10 @@ set;
 | execution.state-recovery.ignore-unclaimed-state |     false |
 |                                execution.target |    remote |
 |                          jobmanager.rpc.address | $VAR_JOBMANAGER_RPC_ADDRESS |
-|                             pipeline.classpaths |        [] |
-|                                   pipeline.jars |        [] |
 |                                       rest.port |     $VAR_REST_PORT |
 |                              sql-client.verbose |      true |
 +-------------------------------------------------+-----------+
-10 rows in set
+8 rows in set
 !ok
 
 set 'execution.attached' = 'false';
@@ -175,12 +169,10 @@ set;
 | execution.state-recovery.ignore-unclaimed-state |     false |
 |                                execution.target |    remote |
 |                          jobmanager.rpc.address | $VAR_JOBMANAGER_RPC_ADDRESS |
-|                             pipeline.classpaths |        [] |
-|                                   pipeline.jars |        [] |
 |                                       rest.port |     $VAR_REST_PORT |
 |                              sql-client.verbose |      true |
 +-------------------------------------------------+-----------+
-10 rows in set
+8 rows in set
 !ok
 
 # test reset can work with add jar
@@ -207,12 +199,10 @@ set;
 | execution.state-recovery.ignore-unclaimed-state |     false |
 |                                execution.target |    remote |
 |                          jobmanager.rpc.address | $VAR_JOBMANAGER_RPC_ADDRESS |
-|                             pipeline.classpaths |        [] |
-|                                   pipeline.jars |        [] |
 |                                       rest.port |     $VAR_REST_PORT |
 |                              sql-client.verbose |      true |
 +-------------------------------------------------+-----------+
-10 rows in set
+8 rows in set
 !ok
 
 reset;

diff --git a/flink-table/flink-sql-gateway/src/test/resources/sql/set.q b/flink-table/flink-sql-gateway/src/test/resources/sql/set.q
@@ -36,11 +36,9 @@ set;
 | execution.state-recovery.ignore-unclaimed-state |     false |
 |                                execution.target |    remote |
 |                          jobmanager.rpc.address | localhost |
-|                             pipeline.classpaths |        [] |
-|                                   pipeline.jars |        [] |
 |                                       rest.port |     $VAR_REST_PORT |
 +-------------------------------------------------+-----------+
-9 rows in set
+7 rows in set
 !ok
 
 # set illegal value
@@ -60,9 +58,7 @@ set;
 | execution.state-recovery.ignore-unclaimed-state |     false |
 |                                execution.target |    remote |
 |                          jobmanager.rpc.address | localhost |
-|                             pipeline.classpaths |        [] |
-|                                   pipeline.jars |        [] |
 |                                       rest.port |     $VAR_REST_PORT |
 +-------------------------------------------------+-----------+
-9 rows in set
+7 rows in set
 !ok