[FLINK-23474] Extract internal version of InputStatus #16581

dawidwys · 2021-07-23T11:48:49Z

What is the purpose of the change

This commit separates internal and user facing versions of InputStatus.
User sources should never return e.g. the END_OF_RECOVERY status and
thus we need an internal status.

Verifying this change

This change is a trivial rework / code cleanup without any test coverage.

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

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

dawidwys · 2021-07-23T11:48:58Z

cc @pnowojski

flinkbot · 2021-07-23T11:51:30Z

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 b632e35 (Sat Aug 28 12:22:19 UTC 2021)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{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 · 2021-07-23T12:17:55Z

CI report:

b632e35 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

pnowojski

LGTM % a couple of minor comments

pnowojski · 2021-07-23T15:02:55Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java

+    private DataInputStatus convertToInternalStatus(InputStatus inputStatus) {
+        switch (inputStatus) {
+            case MORE_AVAILABLE:
+                return DataInputStatus.MORE_AVAILABLE;
+            case NOTHING_AVAILABLE:
+                return DataInputStatus.NOTHING_AVAILABLE;
+            case END_OF_INPUT:
+                return DataInputStatus.END_OF_INPUT;
+            default:
+                throw new IllegalArgumentException("Unknown input status: " + inputStatus);
+        }


I would be curious how well JIT can optimise this... probably as long as sourceReader.pollNext() is de-virtualised and inlined this will be for free. But otherwise I would guess it will cost a couple of CPU cycles. Either way I don't expect this to be visible as a performance regression, but I'm just curious...

pnowojski · 2021-07-23T15:04:09Z

flink-core/src/main/java/org/apache/flink/core/io/InputStatus.java

-    /** Indicator that all persisted data of the data exchange has been successfully restored. */
-    END_OF_RECOVERY,
-


This change requires a release note 🙄

pnowojski · 2021-07-23T15:06:13Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/DataInputStatus.java

+ * An {@code InputStatus} indicates the availability of data from an asynchronous input. When asking
+ * an asynchronous input to produce data, it returns this status to indicate how to proceed.
+ *
+ * <p>It is an internal equivalent of {@link org.apache.flink.core.io.InputStatus} that provides
+ * additional non public statuses.


nitty nit: reverse order of the paragraphs? Start with:

DataInputStatus is an internal equivalent of {@link org.apache.flink.core.io.InputStatus} that provides additional non public statuses.

?

dawidwys · 2021-07-25T12:19:29Z

@flinkbot run azure

This commit separates internal and user facing versions of InputStatus. User sources should never return e.g. the END_OF_RECOVERY status and thus we need an internal status.

dawidwys · 2021-07-26T06:37:29Z

@flinkbot run azure

rmetzger added review=description? component=API/DataStream component=Runtime/Task labels Jul 23, 2021

pnowojski approved these changes Jul 23, 2021

View reviewed changes

dawidwys force-pushed the flink-23474 branch 2 times, most recently from 33acb9c to 2a96fd5 Compare July 25, 2021 10:32

[FLINK-23474] Extract internal version of InputStatus

b632e35

This commit separates internal and user facing versions of InputStatus. User sources should never return e.g. the END_OF_RECOVERY status and thus we need an internal status.

dawidwys force-pushed the flink-23474 branch from 2a96fd5 to b632e35 Compare July 26, 2021 06:14

dawidwys merged commit 684d56e into apache:master Jul 26, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-23474] Extract internal version of InputStatus #16581

[FLINK-23474] Extract internal version of InputStatus #16581

dawidwys commented Jul 23, 2021

dawidwys commented Jul 23, 2021

flinkbot commented Jul 23, 2021 •

edited

Loading

flinkbot commented Jul 23, 2021 •

edited

Loading

pnowojski left a comment

pnowojski Jul 23, 2021

pnowojski Jul 23, 2021

pnowojski Jul 23, 2021

dawidwys commented Jul 25, 2021

dawidwys commented Jul 26, 2021

		/** Indicator that all persisted data of the data exchange has been successfully restored. */
		END_OF_RECOVERY,

[FLINK-23474] Extract internal version of InputStatus #16581

[FLINK-23474] Extract internal version of InputStatus #16581

Conversation

dawidwys commented Jul 23, 2021

What is the purpose of the change

Verifying this change

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

Documentation

dawidwys commented Jul 23, 2021

flinkbot commented Jul 23, 2021 • edited Loading

Automated Checks

Review Progress

flinkbot commented Jul 23, 2021 • edited Loading

CI report:

pnowojski left a comment

Choose a reason for hiding this comment

pnowojski Jul 23, 2021

Choose a reason for hiding this comment

pnowojski Jul 23, 2021

Choose a reason for hiding this comment

pnowojski Jul 23, 2021

Choose a reason for hiding this comment

dawidwys commented Jul 25, 2021

dawidwys commented Jul 26, 2021

flinkbot commented Jul 23, 2021 •

edited

Loading

flinkbot commented Jul 23, 2021 •

edited

Loading