[FLINK-28474][checkpoint] Fix the bug ChannelStateWriteResult might not fail after checkpoint abort

[1996fanrui]

What is the purpose of the change

Simplify the ChannelStateWriterImpl due to concurrent unaligned checkpoint isn't supported.

Brief change log

• Simplify the ChannelStateWriterImpl due to concurrent unaligned checkpoint isn't supported.
• If the checkpointId is aborted, the checkpoint will be ignored within ChannelStateWriter#start.

Verifying this change

This change added tests and can be verified as follows:

• ChannelStateWriterImplTest#testStartMultipleTimes()
• ChannelStateWriterImplTest#testBuffersRecycledOnAbortedCheckpoint()
• ChannelStateWriterImplTest#testOldCheckpointIsAborted()

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

• Dependencies (does it add or upgrade a dependency): yes
• The public API, i.e., is any changed class annotated with @Public(Evolving): yes
• The serializers: no
• 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? no
• If yes, how is the feature documented? not documented

[flinkbot]

CI report:

• 52f3fdb Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[pnowojski]

Thanks for the changes, please see a couple of my comments.

[pnowojski]

Thanks @1996fanrui, I think the code looks mostly good. I've started a benchmark request to check the impact of replacing ConcurrentHashMap with a lock. I don't expect it to be an issue, but let's check that.

http://codespeed.dak8s.net:8080/job/flink-benchmark-request/171/

Also FYI @rkhachatryan, I've already reviewed this code, but you might be interested in this issue.

[1996fanrui]

Thanks @1996fanrui, I think the code looks mostly good. I've started a benchmark request to check the impact of replacing ConcurrentHashMap with a lock. I don't expect it to be an issue, but let's check that.

http://codespeed.dak8s.net:8080/job/flink-benchmark-request/171/

Also FYI @rkhachatryan, I've already reviewed this code, but you might be interested in this issue.

Checkpoint is a relatively low-frequency behavior. My understanding is that this change should have little or no impact on data processing.

[pnowojski]

Checkpoint is a relatively low-frequency behavior. My understanding is that this change should have little or no impact on data processing.

Checkpointing yes, but you have added isCheckpointSubsumedOrAborted call for every addInputData() call for example, which might cause some performance regression

I've checked the benchmark results and didn't notice anything suspicious, however our benchmark coverage around checkpointing is pretty scarce. I have left a couple of more comments however.

[1996fanrui]

Hi @pnowojski , I updated this PR based on the first version. Please help take a look in your free time, thanks a lot~

[pnowojski]

Thanks @1996fanrui for the update. I think this version looks much simpler compared to the previous. And sorry again for the detour.

I've left a couple of smaller comments.

[1996fanrui]

Thanks @1996fanrui for the update. I think this version looks much simpler compared to the previous. And sorry again for the detour.

I've left a couple of smaller comments.

Hi @pnowojski , thanks for your review, I have addressed all comments.

[pnowojski]

Thanks @1996fanrui for the update LGTM % last final minor issue that I think came back after reverting to the earlier version:
#20233 (comment)

[pnowojski]

LGTM, let's wait for green azure