[FLINK-35025][Runtime/State] Abstract stream operators for async state processing

[Zakelly]

What is the purpose of the change

As part of the async execution model of disaggregated state management, this PR gives the basic definition of StreamingOperator integrated with async execution.

The philosophy behind this PR is we are trying NOT to invade the previous code path. The most strict thing is that it MUST NOT affect any hot path without this feature.

By the way, the type parameter of record around AsyncExecutionController is erased, since the stream operator does not have the knowledge of record type, and the AEC does not need it.

Brief change log

• Erase the type parameter of record around AsyncExecutionController
• Define the AbstractAsyncStateStreamOperator and AbstractAsyncStateStreamOperatorV2 corresponding to the AbstracStreamOperator and AbstracStreamOperatorV2 respectively.
• Wire the record processing logic with new introduced abstract stream operators in RecordProcessorUtils.

Verifying this change

This change around AEC is already covered by existing tests. Basic UTs for new operators are added.

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): yes
• 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:

• 5af1006 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[yunfengzhou-hub]

Thanks for the PR. Left some comments as below.

[yunfengzhou-hub]

Making the async functionalities an abstract class would prevent the operator from extending from other abstract classes. For example, a RichMapFunction should be a subclass of AbstractUdfStreamOperator, and since it has extended from this abstract class, it cannot extend from AbstractAsyncStateStreamOperator at the same time.

[Zakelly]

The async processing starts variant at AbstractStreamOperator and AbstractStreamOperatorV2. My brief idea is, if some operator need integrate with the async processing, they should change to extends AbstractAsyncStateStreamOperator instead of AbstractStreamOperator. Moreover, the AbstractAsyncStateStreamOperator is fully compatible with AbstractStreamOperator, if we make AbstractAsyncStateStreamOperator#isAsyncStateProcessingEnabled configurable to false, everything is not changed compared with AbstractStreamOperator. That's what this PR trying to introduce.

Back to the AbstractUdfStreamOperator, it is a subclass of AbstractStreamOperator. We have two options, the first of which is a new introduced identical class AbstractAsyncStateUdfStreamOperator extending AbstractAsyncStateStreamOperator, the second is to make AbstractUdfStreamOperator extend AbstractAsyncStateStreamOperator. I'd prefer the former, we can discuss this in following PR.

[yunfengzhou-hub]

Instead of introducing setAsyncKeyedContextElement, postProcessElement and getRecordProcessor, how about the following implementation?

@Override
public void processElement(StreamRecord<T> record){
        lastProcessContext =
                asyncExecutionController.buildContext(
                        record.getValue(), keySelector.getKey(record.getValue()));
        lastProcessContext.retain();
        asyncExecutionController.setCurrentContext(lastProcessContext);

        super.processElement(record);

        lastProcessContext.release();

        // getRecordProcessor is removed, since the implementation in RecordProcessUtils should be enough
}

[Zakelly]

The processElement will be overriden by subclasses, not defined by the superclass. We should leave the processElement abstract and ship our implementation around this.

[yunfengzhou-hub]

It might be better to configure isAsyncStateProcessingEnabled globally on the whole Flink job instead of at operators' granularity, and in that case this method would be unnecessary.

[Zakelly]

It should be and will be read from the global config. All operators will have the same option here.

[fredia]

@Zakelly Thanks for the PR, I left some comments, PTAL

[Zakelly]

@fredia @yunfengzhou-hub Thanks a lot for your review. I have made another commit to introduce the element order of processElement as FLIP-425 said. This is strictly internal and not exposed to users for now.

[fredia]

@Zakelly Thanks for the update, I left some minor comments.

[fredia]

Thanks for the update, overall LGTM.

[yunfengzhou-hub]

Thanks for updating the PR. Left some minor comments as below.

[yunfengzhou-hub]

Thanks for the update! The changes on the stream operators LGTM.

[Zakelly]

Thanks @fredia and @yunfengzhou-hub for your detailed review! Really appreciate it. 🚀