[FLINK-33398][runtime] Support switching from batch to stream mode for one input stream operator

[Sxnan]

What is the purpose of the change

This PR supports switching from batch to stream mode to improve the performance during processing backlog data for one input stream operator.

Brief change log

• Introduce RecordAttributes to notify the downstream whether the records are backlog data.
• Propagate the RecordAttributes at runtime.
• Automatically sort the backlog data for one input operator.

Verifying this change

This change added tests and can be verified as follows:

• Added integration test for keyed aggregation and keyed windowed aggregation with backlog data.

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

[flinkbot]

CI report:

• 6959d8f Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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

[Sxnan]

@flinkbot run azure

[yunfengzhou-hub]

Thanks for the PR. Left some comments as below.

[yunfengzhou-hub]

Thanks for the update. Left some comments as below.

Besides, this PR does not cover all the changes proposed in FLIP-327. According to offline discussions, we need more PRs to support state cache and built-in multi-input operators. Therefore, it might be better to create child tickets for FLINK-33202 that cover the whole implementation plan, and assign this PR to one of these child tickets.

[Sxnan]

@yunfengzhou-hub Thanks for the review! I updated the PR. Can you have another look?

[yunfengzhou-hub]

Thanks for the update. Left some minor comments as below. @xintongsong Could you please take a look at this PR?

[Sxnan]

@xintongsong Could you help review this PR?

[SmirAlex]

Hi @Sxnan! Sorry, I understand that I'm not a reviewer, but it happened that I was testing functionality from this MR recently, and I found a bug (in my opinion). It concerns the logic of RecordAttributesValve.
There was very little backlog data in my test at source + it had parallelism 4 (actually, any value > 1 suits the case). Due to very short backlog phase, time interval between sending RecordAttributes(isBacklog=true) and RecordAttributes(isBacklog=false) was also very short. In addition, due to the high parallelism one source subtask could send RecordAttributes(isBacklog=false) even before RecordAttributes(isBacklog=true) of another subtask. As a result, race condition have occurred in RecordAttributesValve#inputRecordAttributes. backlogChannelsCnt was incrementing and decrementing simultaneously, which led to not reaching numInputChannels, so no RecordAttributes was emitted from RecordAttributesValve at all.

I suggest to have different counters for RecordAttributes(isBacklog=true) and for RecordAttributes(isBacklog=false). Therefore, the race condition I mentioned earlier won't affect the result. Something like this:

    if (recordAttributes.isBacklog()) {
        backlogChannelsCnt += 1;
        if (backlogChannelsCnt != numInputChannels) {
            return;
        }
        backlogChannelsCnt = 0;
    } else {
        nonBacklogChannelsCnt += 1;
        if (nonBacklogChannelsCnt != numInputChannels) {
            return;
        }
        nonBacklogChannelsCnt = 0;
    }

    if (lastOutputAttributes == null
            || lastOutputAttributes.isBacklog() != recordAttributes.isBacklog()) {
        if (lastOutputAttributes != null && !lastOutputAttributes.isBacklog()) {
            LOG.warn(
                    "Switching from non-backlog to backlog is currently not supported. Backlog status remains.");
            return;
        }
        lastOutputAttributes = recordAttributes;
        output.emitRecordAttributes(recordAttributes);
    }

WDYT? Also, it is good to have a test for aforementioned case.

[Sxnan]

Hi @Sxnan! Sorry, I understand that I'm not a reviewer, but it happened that I was testing functionality from this MR recently, and I found a bug (in my opinion). It concerns the logic of RecordAttributesValve. There was very little backlog data in my test at source + it had parallelism 4 (actually, any value > 1 suits the case). Due to very short backlog phase, time interval between sending RecordAttributes(isBacklog=true) and RecordAttributes(isBacklog=false) was also very short. In addition, due to the high parallelism one source subtask could send RecordAttributes(isBacklog=false) even before RecordAttributes(isBacklog=true) of another subtask. As a result, race condition have occurred in RecordAttributesValve#inputRecordAttributes. backlogChannelsCnt was incrementing and decrementing simultaneously, which led to not reaching numInputChannels, so no RecordAttributes was emitted from RecordAttributesValve at all.

I suggest to have different counters for RecordAttributes(isBacklog=true) and for RecordAttributes(isBacklog=false). Therefore, the race condition I mentioned earlier won't affect the result. Something like this:

    if (recordAttributes.isBacklog()) {
        backlogChannelsCnt += 1;
        if (backlogChannelsCnt != numInputChannels) {
            return;
        }
        backlogChannelsCnt = 0;
    } else {
        nonBacklogChannelsCnt += 1;
        if (nonBacklogChannelsCnt != numInputChannels) {
            return;
        }
        nonBacklogChannelsCnt = 0;
    }

    if (lastOutputAttributes == null
            || lastOutputAttributes.isBacklog() != recordAttributes.isBacklog()) {
        if (lastOutputAttributes != null && !lastOutputAttributes.isBacklog()) {
            LOG.warn(
                    "Switching from non-backlog to backlog is currently not supported. Backlog status remains.");
            return;
        }
        lastOutputAttributes = recordAttributes;
        output.emitRecordAttributes(recordAttributes);
    }

WDYT? Also, it is good to have a test for aforementioned case.

Hi @Smir. Thanks for trying this out! The RecordAttributesValve combines the RecodAttributes from different input channels from the same input. The input is considered in backlog state if and only if all the input channels are backlog = true. Otherwise, some non-backlog records will be considered as backlog records.

Back to your case, where there is very little backlog data and it has parallelism greater than 1. It is possible that an input channel switches to non-backlog state before other input channels switch to backlog state. When that happens, we just process those data as if they are non-backlog data.

[SmirAlex]

Thanks for the response, @Sxnan. I got your points. BTW, the reason why delivery guarantees of RecordAttributes elements to downstream operators was important to me is that I was testing the possibility to use them in order to properly implement Processing Time Temporal Join. I described my thoughts about this in FLIP-326 discussion thread. Thus, if eventually it will be decided to reuse RecordAttributes logic in order to solve FLINK-19830, the problem I described will arise one way or another.

Also, I should mention that there is another workaround that guarantees delivery of RecordAttributes(isBacklog=false) (but not isBacklog=true), but at the same time prevents considering some non-backlog records as backlog records. Pseudocode may be like this:

if (lastOutputAttributes == null && nonBacklogChannelsCnt == numInputChannels) {
    nonBacklogChannelsCnt = 0;
    output.emitRecordAttributes(RecordAttributes(isBacklog=false));
}

The downside is that it is possible to have RecordAttributes(isBacklog=false) without RecordAttributes(isBacklog=true) previously, but probably, it's ok. In other respects, this approach worked fine for me.

[xintongsong]

What does newKey == null mean? And why we need to update the current watermark under such condition?

[xintongsong]

It feels hacky to expose an internal queue from one component and use it to construct another component. If the queue is meant to be shared, I think the correct way is to create the queue out side the two components and pass it in as constructor arguments.

[xintongsong]

TBH, I'm not a big fan of inherit. The problem is that, by directly accessing protected fields of the superclass from the subclasses, the contract between the classes becomes obscure. As subclass can see internals of the superclass, changes to the superclass can easily break things that subclasses depend on thus cause problems, and such issues are very hard to avoid.

In many cases, an inherit approach can be replaced by an equivalent approach based on composition, which has clearer contracts between classes. But admittedly it's no always possible and sometimes gets costly.

For the classes below, I guess I haven't spent enough time to explore whether it's possible to switch to a composition based approach. Do you think it's feasible?

• BacklogTimeService
• BatchExecutionInternalTimeService
• InternalBacklogAwareTimerServiceManagerImpl
• InternalTimeServiceManagerImpl

[xintongsong]

Same here for deciding whether to enable the mixed mode.

[xintongsong]

I'm not familiar with Scala. What is this for? Shouldn't the method from the super class be called anyway if we don't override it?

[Sxnan]

After an offline discussion with @xintongsong, we will split this PR into two PRs. This PR will focus on optimizing the one input operator during backlog. The other PR will focus on propagating the RecordAttributes through the job graph.

This PR should be merged after the other PR.