[FLINK-35355][State] Internal async aggregating state and correspondi…

[jectpro7]

This PR defines the V2 AggreatingState and its StateDescriptor. And provides the default InternalAggregatingState implementation which delegates all aycn request to AEC.

What is the purpose of the change

Supporting aggregate functionality regarding to FLIP-424.

Brief change log

1. Definition of v2 AggregatingState and AggregatingStateDescriptor
2. Default implementation of v2 AggregatingState

Verifying this change

1. Added UT for new class and changed code.

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): no
• 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 applicable

[flinkbot]

CI report:

• 972f0e4 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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