[FLINK-38491][pipeline-connector][iceberg] fixed iceberg compaction parallelism (#4150)

Author: fcfangcc

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/IcebergDataSinkFactory.java

@@ -122,6 +122,7 @@ public Set<ConfigOption<?>> optionalOptions() {
         options.add(IcebergDataSinkOptions.PARTITION_KEY);
         options.add(IcebergDataSinkOptions.SINK_COMPACTION_ENABLED);
         options.add(IcebergDataSinkOptions.SINK_COMPACTION_COMMIT_INTERVAL);
+        options.add(IcebergDataSinkOptions.SINK_COMPACTION_PARALLELISM);
         return options;
     }
 }

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergSink.java

@@ -136,7 +136,7 @@ public void addPostCommitTopology(
             // Shuffle by different table id.
             DataStream<CommittableMessage<WriteResultWrapper>> keyedStream =
                     committableMessageDataStream.partitionCustom(
-                            (bucket, numPartitions) -> bucket % numPartitions,
+                            Math::floorMod,
                             (committableMessage) -> {
                                 if (committableMessage instanceof CommittableWithLineage) {
                                     WriteResultWrapper multiTableCommittable =

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/test/java/org/apache/flink/cdc/connectors/iceberg/sink/IcebergDataSinkFactoryTest.java

@@ -40,6 +40,7 @@ void testCreateDataSink() {
 
         Configuration conf = Configuration.fromMap(ImmutableMap.<String, String>builder().build());
         conf.set(IcebergDataSinkOptions.WAREHOUSE, "/tmp/warehouse");
+        conf.set(IcebergDataSinkOptions.SINK_COMPACTION_PARALLELISM, 4);
         DataSink dataSink =
                 sinkFactory.createDataSink(
                         new FactoryHelper.DefaultContext(

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/test/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/CompactionOperatorTest.java

@@ -150,7 +150,8 @@ public void testCompationOperator() throws IOException, InterruptedException {
         }
         CompactionOperator compactionOperator =
                 new CompactionOperator(
-                        catalogOptions, CompactionOptions.builder().commitInterval(1).build());
+                        catalogOptions,
+                        CompactionOptions.builder().commitInterval(1).parallelism(4).build());
         compactionOperator.processElement(
                 new StreamRecord<>(
                         new CommittableWithLineage<>(