[FLINK-24915] [DataStream] fix StreamElementSerializer#deserialize(reuse, source) forgets to handle tag == TAG_STREAM_STATUS. #17838

gmzz123 · 2021-11-19T12:16:27Z

What is the purpose of the change

This pull request fixes StreamElementSerializer#deserialize(StreamElement reuse, DataInputView source) not handle tag == TAG_STREAM_STATUS.

Brief change log

deserialize(StreamElement reuse, DataInputView source) return the WatermarkStatus when tag == TAG_STREAM_STATUS.
copy(StreamElement from, StreamElement reuse) need not check reuse is a StreamRecord.
copy(DataInputView source, DataOutputView target), there is no need to deserialize then serialize. As for each type of StreamElement, the binary length is fixed. So we can use binary copy to speed up.

Verifying this change

This change added tests and can be verified as follows:

add test for copy(with/without object reuse). this copy will not go through DataInputView/DataOutputView.
add test for StreamElement type WatermarkStatus.

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: (yes)
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 / docs / JavaDocs / not documented)

…use, source) forgets to handle tag == TAG_STREAM_STATUS.

flinkbot · 2021-11-19T12:18:42Z

CI report:

2e59912 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

flinkbot · 2021-11-19T12:20:46Z

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 6658dc7 (Fri Nov 19 12:20:46 UTC 2021)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!
This pull request references an unassigned Jira ticket. According to the code contribution guide, tickets need to be assigned before starting with the implementation work.

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

Details

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

gmzz123 · 2021-11-23T07:00:14Z

@AHeise
Could you help review it? Thanks!

AHeise

Thank you very much for your patience and your contribution. The change looks good to me and I also like the secondary changes.

Could you please extract some changes to a separate commit in this PR? So we should have
[hotfix][DataStream] Use diamond operator in StreamElementSerializer(Test)
[FLINK-24915][DataStream] Improve StreamElementSerializer#copy
[FLINK-24915][DataStream] Fix StreamElementSerializer#deserialize for StreamStatus

If Azure pipeline is green, could you please also prepare backport PRs against Flink 1.14, 1.13 and maybe 1.12 if trivial (we don't usually release a new 1.12 build but sometimes we do).

AHeise · 2021-12-06T16:12:16Z

...c/test/java/org/apache/flink/streaming/runtime/streamrecord/StreamElementSerializerTest.java

+        StreamRecord<String> withoutTimestamp = new StreamRecord<>("test 1 2 分享基督耶穌的愛給們，開拓雙贏!");
+        assertEquals(
+                withoutTimestamp,
+                serializeAndDeserializeWithReuse(withoutTimestamp, serializer, reuse));
+
+        StreamRecord<String> withTimestamp = new StreamRecord<>("one more test 拓 們 分", 77L);
+        assertEquals(
+                withTimestamp, serializeAndDeserializeWithReuse(withTimestamp, serializer, reuse));
+
+        StreamRecord<String> negativeTimestamp = new StreamRecord<>("他", Long.MIN_VALUE);
+        assertEquals(
+                negativeTimestamp,
+                serializeAndDeserializeWithReuse(negativeTimestamp, serializer, reuse));


I'd also like to see an assert that reuse is actually returned.

You could switch to assertj-style assertions to make it more beautiful:

assertThat(serializeAndDeserializeWithReuse(negativeTimestamp, serializer, reuse)) .isEqualTo(negativeTimestamp) .isSameAs(reuse)

AHeise · 2021-12-06T16:14:49Z

...a/src/main/java/org/apache/flink/streaming/runtime/streamrecord/StreamElementSerializer.java

-            target.writeLong(source.readLong());
+            target.write(source, 8);
            typeSerializer.copy(source, target);
        } else if (tag == TAG_REC_WITHOUT_TIMESTAMP) {
            typeSerializer.copy(source, target);
        } else if (tag == TAG_WATERMARK) {
-            target.writeLong(source.readLong());
+            target.write(source, 8);
        } else if (tag == TAG_STREAM_STATUS) {
-            target.writeInt(source.readInt());
+            target.write(source, 4);
        } else if (tag == TAG_LATENCY_MARKER) {
-            target.writeLong(source.readLong());
-            target.writeLong(source.readLong());
-            target.writeLong(source.readLong());
-            target.writeInt(source.readInt());
+            target.write(source, 28);


Please extract these performance optimizations into a separate commit.

AHeise · 2021-12-06T16:15:06Z

...a/src/main/java/org/apache/flink/streaming/runtime/streamrecord/StreamElementSerializer.java

-        if (from.isRecord() && reuse.isRecord()) {
+        // need not check reuse is really a StreamRecord, otherwise reuse.asRecord() will throw
+        // ClassCastException, similar as cannot copy does.
+        if (from.isRecord()) {


Please extract these performance optimizations into a separate commit.

AHeise · 2021-12-06T16:15:52Z

...a/src/main/java/org/apache/flink/streaming/runtime/streamrecord/StreamElementSerializer.java

    public StreamElementSerializer<T> duplicate() {
        TypeSerializer<T> copy = typeSerializer.duplicate();
-        return (copy == typeSerializer) ? this : new StreamElementSerializer<T>(copy);
+        return (copy == typeSerializer) ? this : new StreamElementSerializer<>(copy);


Please extract these warning fixes to a separate commit.

github-actions · 2025-01-15T00:16:51Z

This PR is being marked as stale since it has not had any activity in the last 180 days.
If you would like to keep this PR alive, please leave a comment asking for a review.
If the PR has merge conflicts, update it with the latest from the base branch.

If you are having difficulty finding a reviewer, please reach out to the [community](https://flink.apache.org/what-is-flink/community/).

If this PR is no longer valid or desired, please feel free to close it. If no activity occurs in the next 90 days, it will be automatically closed.

github-actions · 2025-03-30T06:26:28Z

This PR has been closed since it has not had any activity in 120 days.
If you feel like this was a mistake, or you would like to continue working on it,
please feel free to re-open the PR and ask for a review.

[FLINK-24915] [DataStream] fix StreamElementSerializer#deserialize(re…

6658dc7

…use, source) forgets to handle tag == TAG_STREAM_STATUS.

rmetzger added the component=API/DataStream label Nov 19, 2021

gmzz123 added 3 commits November 22, 2021 16:46

Merge branch 'apache:master' into streamStatus

c97856d

Merge branch 'apache:master' into streamStatus

c820468

Merge branch 'apache:master' into streamStatus

2e59912

AHeise reviewed Dec 6, 2021

View reviewed changes

1996fanrui added the stale label Jan 17, 2025

github-actions bot added the closed-stale label Mar 30, 2025

github-actions bot closed this Mar 30, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[FLINK-24915] [DataStream] fix StreamElementSerializer#deserialize(reuse, source) forgets to handle tag == TAG_STREAM_STATUS. #17838

[FLINK-24915] [DataStream] fix StreamElementSerializer#deserialize(reuse, source) forgets to handle tag == TAG_STREAM_STATUS. #17838

Uh oh!

gmzz123 commented Nov 19, 2021

Uh oh!

flinkbot commented Nov 19, 2021 •

edited

Loading

Uh oh!

flinkbot commented Nov 19, 2021

Uh oh!

gmzz123 commented Nov 23, 2021

Uh oh!

AHeise left a comment

Uh oh!

AHeise Dec 6, 2021

Uh oh!

AHeise Dec 6, 2021

Uh oh!

AHeise Dec 6, 2021

Uh oh!

AHeise Dec 6, 2021

Uh oh!

github-actions bot commented Jan 15, 2025

Uh oh!

github-actions bot commented Mar 30, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

[FLINK-24915] [DataStream] fix StreamElementSerializer#deserialize(reuse, source) forgets to handle tag == TAG_STREAM_STATUS. #17838

[FLINK-24915] [DataStream] fix StreamElementSerializer#deserialize(reuse, source) forgets to handle tag == TAG_STREAM_STATUS. #17838

Uh oh!

Conversation

gmzz123 commented Nov 19, 2021

What is the purpose of the change

Brief change log

Verifying this change

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

Documentation

Uh oh!

flinkbot commented Nov 19, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

flinkbot commented Nov 19, 2021

Automated Checks

Review Progress

Uh oh!

gmzz123 commented Nov 23, 2021

Uh oh!

AHeise left a comment

Choose a reason for hiding this comment

Uh oh!

AHeise Dec 6, 2021

Choose a reason for hiding this comment

Uh oh!

AHeise Dec 6, 2021

Choose a reason for hiding this comment

Uh oh!

AHeise Dec 6, 2021

Choose a reason for hiding this comment

Uh oh!

AHeise Dec 6, 2021

Choose a reason for hiding this comment

Uh oh!

github-actions bot commented Jan 15, 2025

Uh oh!

github-actions bot commented Mar 30, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

flinkbot commented Nov 19, 2021 •

edited

Loading