Flink: Set generator parallelism to match input in DynamicIcebergSink

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java

@@ -408,6 +408,7 @@ public DataStreamSink<DynamicRecordInternal> append() {
                       tableCreator,
                       caseSensitive,
                       dropUnusedColumns))
+              .setParallelism(input.getParallelism())
               .uid(prefixIfNotNull(uidPrefix, "-generator"))
               .name(operatorName("generator"))
               .returns(type);

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java

@@ -1196,6 +1196,33 @@ void testOperatorUidsFormat() {
             "Sink Committer: --sink");
   }
 
+  @Test
+  void testGeneratorDefaultParallelism() {
+    StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
+    streamEnv.setParallelism(4);
+
+    DataStreamSource<DynamicIcebergDataImpl> source =
+        streamEnv.fromData(Collections.emptySet(), TypeInformation.of(new TypeHint<>() {}));
+    source.setParallelism(8);
+
+    DynamicIcebergSink.forInput(source)
+        .generator(new Generator())
+        .catalogLoader(CATALOG_EXTENSION.catalogLoader())
+        .uidPrefix("test")
+        .append();
+
+    // Since the generator parallelism is not directly accessible via the returned DataStreamSink,
+    // inspect the stream graph to verify the generator inherits the input source parallelism.
+    int generatorParallelism =
+        streamEnv.getStreamGraph().getStreamNodes().stream()
+            .filter(node -> "test--generator".equals(node.getTransformationUID()))
+            .findFirst()
+            .map(StreamNode::getParallelism)
+            .orElseThrow(() -> new AssertionError("Generator node not found"));
+
+    assertThat(generatorParallelism).isEqualTo(source.getParallelism());
+  }
+
   private Set<String> createSinkAndReturnUIds(String uidPrefix) {
     StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
 

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java

@@ -407,6 +407,7 @@ public DataStreamSink<DynamicRecordInternal> append() {
                       tableCreator,
                       caseSensitive,
                       dropUnusedColumns))
+              .setParallelism(input.getParallelism())
               .uid(prefixIfNotNull(uidPrefix, "-generator"))
               .name(operatorName("generator"))
               .returns(type);

flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java

@@ -1196,6 +1196,33 @@ void testOperatorUidsFormat() {
             "Sink Committer: --sink");
   }
 
+  @Test
+  void testGeneratorDefaultParallelism() {
+    StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
+    streamEnv.setParallelism(4);
+
+    DataStreamSource<DynamicIcebergDataImpl> source =
+        streamEnv.fromData(Collections.emptySet(), TypeInformation.of(new TypeHint<>() {}));
+    source.setParallelism(8);
+
+    DynamicIcebergSink.forInput(source)
+        .generator(new Generator())
+        .catalogLoader(CATALOG_EXTENSION.catalogLoader())
+        .uidPrefix("test")
+        .append();
+
+    // Since the generator parallelism is not directly accessible via the returned DataStreamSink,
+    // inspect the stream graph to verify the generator inherits the input source parallelism.
+    int generatorParallelism =
+        streamEnv.getStreamGraph().getStreamNodes().stream()
+            .filter(node -> "test--generator".equals(node.getTransformationUID()))
+            .findFirst()
+            .map(StreamNode::getParallelism)
+            .orElseThrow(() -> new AssertionError("Generator node not found"));
+
+    assertThat(generatorParallelism).isEqualTo(source.getParallelism());
+  }
+
   private Set<String> createSinkAndReturnUIds(String uidPrefix) {
     StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
 

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java

@@ -407,6 +407,7 @@ public DataStreamSink<DynamicRecordInternal> append() {
                       tableCreator,
                       caseSensitive,
                       dropUnusedColumns))
+              .setParallelism(input.getParallelism())
               .uid(prefixIfNotNull(uidPrefix, "-generator"))
               .name(operatorName("generator"))
               .returns(type);

flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java

@@ -1196,6 +1196,33 @@ void testOperatorUidsFormat() {
             "Sink Committer: --sink");
   }
 
+  @Test
+  void testGeneratorDefaultParallelism() {
+    StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
+    streamEnv.setParallelism(4);
+
+    DataStreamSource<DynamicIcebergDataImpl> source =
+        streamEnv.fromData(Collections.emptySet(), TypeInformation.of(new TypeHint<>() {}));
+    source.setParallelism(8);
+
+    DynamicIcebergSink.forInput(source)
+        .generator(new Generator())
+        .catalogLoader(CATALOG_EXTENSION.catalogLoader())
+        .uidPrefix("test")
+        .append();
+
+    // Since the generator parallelism is not directly accessible via the returned DataStreamSink,
+    // inspect the stream graph to verify the generator inherits the input source parallelism.
+    int generatorParallelism =
+        streamEnv.getStreamGraph().getStreamNodes().stream()
+            .filter(node -> "test--generator".equals(node.getTransformationUID()))
+            .findFirst()
+            .map(StreamNode::getParallelism)
+            .orElseThrow(() -> new AssertionError("Generator node not found"));
+
+    assertThat(generatorParallelism).isEqualTo(source.getParallelism());
+  }
+
   private Set<String> createSinkAndReturnUIds(String uidPrefix) {
     StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();