Flink: Dynamic Sink: Use max write parallelism if none is supplied

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

@@ -39,6 +39,18 @@ public class DynamicRecord {
   private boolean upsertMode;
   @Nullable private Set<String> equalityFields;
 
+  /**
+   * Constructs a new DynamicRecord.
+   *
+   * @param tableIdentifier The target table identifier.
+   * @param branch The target table branch.
+   * @param schema The target table schema.
+   * @param rowData The data matching the provided schema.
+   * @param partitionSpec The target table {@link PartitionSpec}.
+   * @param distributionMode The {@link DistributionMode}.
+   * @param writeParallelism The number of parallel writers. If set to {@literal <= 0}, will
+   *     automatically configure the maximum available write parallelism.
+   */
   public DynamicRecord(
       TableIdentifier tableIdentifier,
       String branch,

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

@@ -62,4 +62,14 @@ static int safeAbs(int input) {
 
     return -input;
   }
+
+  static int firstPositive(int first, int second) {
+    if (first > 0) {
+      return first;
+    }
+    if (second > 0) {
+      return second;
+    }
+    throw new IllegalArgumentException("None of the supplied ints were positive!");
+  }
 }

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

@@ -99,7 +99,8 @@ int generateKey(
                         dynamicRecord.distributionMode(), DistributionMode.NONE),
                     MoreObjects.firstNonNull(
                         dynamicRecord.equalityFields(), Collections.emptySet()),
-                    dynamicRecord.writeParallelism()));
+                    DynamicSinkUtil.firstPositive(
+                        dynamicRecord.writeParallelism(), maxWriteParallelism)));
     try {
       return keySelector.getKey(
           overrideRowData != null ? overrideRowData : dynamicRecord.rowData());

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

@@ -157,6 +157,28 @@ void testEqualityKeys() throws Exception {
     assertThat(getSubTaskId(writeKey3, writeParallelism, maxWriteParallelism)).isEqualTo(0);
   }
 
+  @Test
+  void testUseMaxWriteParallelismIfWriteParallelismUnspecified() throws Exception {
+    final int maxWriteParallelism = 5;
+    HashKeyGenerator generator = new HashKeyGenerator(16, maxWriteParallelism);
+
+    Set<Integer> writeKeys = Sets.newHashSet();
+    for (int i = 0; i < maxWriteParallelism; i++) {
+      GenericRowData row = GenericRowData.of(i, StringData.fromString("z"));
+      writeKeys.add(
+          getWriteKey(
+              generator,
+              PartitionSpec.unpartitioned(),
+              DistributionMode.NONE,
+              // Use a writeParallelism <= 0
+              -i,
+              Collections.emptySet(),
+              row));
+    }
+
+    assertThat(writeKeys).hasSize(maxWriteParallelism);
+  }
+
   @Test
   void testCapAtMaxWriteParallelism() throws Exception {
     int writeParallelism = 10;