[FLINK-25583] Support compacting small files for FileSink.

[pltbkd]

What is the purpose of the change

This pull request makes the FileSink to support compacting small files, based on the new extended sink API.

Brief change log

• Users may enable compaction when building FileSink by calling enableCompact.
• Committables emitted from FileWriter will be collected by CompactCoordinator and reorgnized into CompactRequests according to the user defined FileCompactStrategy.
• CompactRequests will be sent to CompactorOperators then executed, and the results will be sent to the committers, as if it's created by the writers.
• FileCompactor is introduced to execute the compaction, and several basic implementations are provided.

Verifying this change

This change added some unit tests.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): (yes)
• The serializers: (yes)
• The runtime per-record code paths (performance sensitive): (no)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
• The S3 file system connector: (no)

Documentation

• Does this pull request introduce a new feature? (yes)
• If yes, how is the feature documented? (JavaDocs)

[flinkbot]

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 2ff898d (Wed Feb 09 08:53:09 UTC 2022)

Warnings:

• 1 pom.xml files were touched: Check for build and licensing issues.
• No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

• ❓ 1. The [description] looks good.
• ❓ 2. There is [consensus] that the contribution should go into to Flink.
• ❓ 3. Needs [attention] from.
• ❓ 4. The change fits into the overall [architecture].
• ❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
• @flinkbot approve all to approve all aspects
• @flinkbot approve-until architecture to approve everything until architecture
• @flinkbot attention @username1 [@username2 ..] to require somebody's attention
• @flinkbot disapprove architecture to remove an approval you gave earlier

[flinkbot]

CI report:

• 2ffd741 Azure: CANCELED
• 4841282 Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[pltbkd]

The branch is force pushed to add a hotfix before all commits, to fix the bug in SinkTransformationTranslator that sets parallelism of transformations added in the per-commit topology without checking if the parallelism is set.

[pltbkd]

@gaoyunhaii
Thanks for the suggestions.
The commits are reorgnized to be more clear, and about 2/3 comments are resolved. I'm still working on the remaining ones.

[pltbkd]

All comments so far are resolved now.

[gaoyunhaii]

Thanks @pltbkd for the update! I try to fixed some tests. Also, could you have a look that if the two *StateHandle needs snapshotState ?

[pltbkd]

Thanks @pltbkd for the update! I try to fixed some tests. Also, could you have a look that if the two *StateHandle needs snapshotState ?

Cleaning up of the state is added when the remaining state is drained in the handlers. Nothing special is necessary while snapshotting the state. I suppose we don't need to implement a customized snapshotState. I have added some comments to explain this.

[pltbkd]

@flinkbot run azure

[gaoyunhaii]

LGTM, will merge after ci get greeen.

[gaoyunhaii]

@flinkbot run azure