Flink: Set generator parallelism to match input in DynamicIcebergSink

[sachinnn99]

Summary

• Set DynamicRecordProcessor (generator) operator parallelism to match upstream input parallelism in DynamicIcebergSink.Builder.append()
• Without this fix, the generator falls back to the environment default parallelism, ignoring any parallelism configured on the upstream operator
• Matches the established pattern used in IcebergSink and FlinkSink

Closes #15827

Test plan

• Added testGeneratorDefaultParallelism in TestDynamicIcebergSink (all 3 Flink versions)
• Verified existing testOperatorUidsFormat still passes
• spotlessCheck passes

[Below0]

thanks @sachinnn99! LGTM

[sachinnn99]

Thanks for the review!

[mxm]

Normally, we do the latest Flink version first, and then backport to the other versions, but the changes here are easy enough to review.

LGTM. Thanks @sachinnn99!

[mxm]

CC @pvary

[pvary]

Merged to main.
Thanks @sachinnn99 for the PR and @mxm for the review!

[manuzhang]

It looks this PR did not rebase on #15433 and brought in a test regression.

> Task :iceberg-flink:iceberg-flink-2.1:test
TestDynamicIcebergSink > testNoShuffleTopology() FAILED
    org.opentest4j.AssertionFailedError: 
    Expecting value to be true but was false
        at app//org.apache.iceberg.flink.sink.dynamic.TestDynamicIcebergSink.testNoShuffleTopology(TestDynamicIcebergSink.java:330)

#16026 has been submitted to fix it.

[sachinnn99]

Thanks @mxm for the review and the CC, and @pvary for the approval and merge!

[sachinnn99]

@manuzhang Thanks for catching this and submitting #16026 to fix it. Apologies for the regression — I should have rebased on #15433 before merging.