[FLINK-25557][checkpoint] Introduce incremental/full checkpoint size stats

[Myasuka]

What is the purpose of the change

Introduce incremental/full checkpoint size stats.

Brief change log

• Introduce getIncrementalStateSize for state objects.
• Let PlaceHolderStateHandler hold state size information
• Enable changelog related state handles also consider incremental state size.

Verifying this change

Added new tests in StateUtilTest.java and IncrementalRemoteKeyedStateHandleTest.java

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: 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? yes
• If yes, how is the feature documented? Documented

[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 1a403ce (Tue Jan 11 09:20:29 UTC 2022)

Warnings:

• No documentation files were touched! Remember to keep the Flink docs up to date!
• Invalid pull request title: No valid Jira ID provided

_{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:

• 0f6b718 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[rkhachatryan]

Thanks for the PR @Myasuka , I think it's a very valuable addition!

I have some general questions:

1. How does this work after up-scaling? Will state size be reported e.g. twice? (if so, I guess it's an existing problem, right?)
2. I think documentation needs to be updated (https://nightlies.apache.org/flink/flink-docs-master/docs/ops/monitoring/checkpoint_monitoring/)
3. If checkpoint is aborted, full size is reported, but incremental is not (in AsyncCheckpointRunnable). I think it's fine, but this should also be documented

I've also left some comments in the code, PTAL.

[Myasuka]

@rkhachatryan , what do you mean "How does this work after up-scaling? Will state size be reported e.g. twice"? Checkpoint state size would only report during taking checkpoint.
After restore, we have two choices for incremental state size:

1. same as previous complete checkpoint reported.
2. reset as same as full checkpoint size.
   It seems the 1st one is better, WDYT?

[rkhachatryan]

Checkpoint state size would only report during taking checkpoint.

Right, and RocksDB doesn't re-use SST currently by more than one task, IIRC. So for the next checkpoint no two tasks will report the same file usage.

It's different for changelog though as it can re-use files on up-scaling.
The only solution I see is to deduplicate files on JM (which will also help reporting the number of created files).

Could you explain the two options you proposed? (or let's chat offline maybe)

[rkhachatryan]

Thanks for updating the PR.
I've left some comments, PTAL.

My major concern is that for changelog, incremental size may not include most of the state.

[rkhachatryan]

Thanks for updating the PR.
I have a suggestion on wording of the tooltip message.
Everything else LGTM.

[Myasuka]

@flinkbot run azure