[FLINK-24907] Support side out late data for interval join #18118

chenyuzhi459 · 2021-12-15T11:03:15Z

What is the purpose of the change

This pull requst makes interval-join support side out late data.

Brief change log

Add OutputTag for left and right stream in org.apache.flink.streaming.api.operators.co.IntervalJoinOperator, and emit late data when isLate(ourTimestamp) = true

Verifying this change

IntervalJoinITCase.testIntervalJoinSideOutputLeftLateData
IntervalJoinITCase.testIntervalJoinSideOutputRightLateData

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)

flinkbot · 2021-12-15T11:06:24Z

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 d8e09a1 (Wed Dec 15 11:06:23 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.

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-12-15T11:11:39Z

CI report:

af9887e Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

chenyuzhi459 · 2021-12-16T07:07:16Z

@flinkbot re-run azure

MartijnVisser · 2022-08-31T08:16:56Z

@gyfora Is this a PR that you could review? I recall that this was mentioned in the Flink Slack channel?

gyfora · 2022-08-31T08:21:44Z

Thanks, I or someone from my team will review this ASAP .

gyfora · 2022-08-31T08:45:56Z

@gaborgsomogyi will help reviewing this feature.

gaborgsomogyi

The functionality looks good in general, I have only a minor comment to refactor the tests.

gaborgsomogyi · 2022-09-05T15:20:53Z

flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/IntervalJoinITCase.java

@@ -155,6 +157,134 @@ public void testJoinsCorrectlyWithMultipleKeys() throws Exception {
                "(key2,5):(key2,5)");
    }

+    @Test
+    public void testIntervalJoinSideOutputLeftLateData() throws Exception {


The 2 tests are 99% the same. Can we do something to merge then into a single function call w/ parameters?
For example the content of run can be a function pointer to the extracted function or something.

Thanks for your tips! I have merged the 2 tests into one test method by sideout the late data of left and right side at the same time .
Should I continue to extracted the run content ? Cause I am a little confuse about how to do that.

It would be better to have a private function which is called from 2 different tests in order not to mix-up left and right functionality but I'm personally fine w/ the actual code stand. It's a little bit harder to understand what's going on inside this single test but not horror.

The technique is create a function like

private void testIntervalJoinSideOutput(Consumer<SourceContext<Tuple2<String, Integer>>> streamOneRun) { ... DataStream<Tuple2<String, Integer>> streamOne = env.addSource( new SourceFunction<Tuple2<String, Integer>>() { @Override public void run(SourceContext<Tuple2<String, Integer>> ctx) { streamOneRun.accept(ctx); } @Override public void cancel() { // do nothing } }); ... }

and call it like this:

@Test public void testIntervalJoinSideOutputLeftLateData() throws Exception { testIntervalJoinSideOutput(ctx -> { ctx.collectWithTimestamp(Tuple2.of("key", 2), 2L); ctx.collectWithTimestamp(Tuple2.of("key", 3), 3L); ctx.emitWatermark(new Watermark(3)); ctx.collectWithTimestamp(Tuple2.of("key", 1), 1L); // late data }); ... expectInAnyOrder("(key,1)"); }

Be aware that this may or may not compile, just wanted to show the way :)

Well, I understand what you mean, thanks very much!
I have extracted the run content as a Consumer function

Looks good, thanks.

chenyuzhi459 · 2022-09-07T02:20:51Z

@flinkbot run azure

gaborgsomogyi · 2022-09-08T10:33:08Z

Test failed w/ unrelated error.

gaborgsomogyi · 2022-09-08T10:33:16Z

@flinkbot re-run azure

gaborgsomogyi · 2022-09-08T12:22:23Z

flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/IntervalJoinITCase.java

+                        .sideOutputLeftLateData(late)
+                        .process(new CombineToStringJoinFunction());
+
+        process.getSideOutput(late)


Just to be complete a last beautification can be added, namely we can create a function something like and call it from the 2 places:

private void addSinkToSideOutput(...) { ... }

Ok, it has been optimized.

gaborgsomogyi · 2022-09-08T12:23:29Z

When the last minor is resolved + unit tests passed then it's good to go.

gaborgsomogyi

LGTM.

gaborgsomogyi · 2022-09-09T09:09:00Z

@flinkbot run azure

gaborgsomogyi · 2022-09-09T12:03:18Z

Seems like there is a permanent issue w/ jenkins:

E: Unsupported file ./libssl1.0.0_1.0.2n-1ubuntu5.7_amd64.deb given on commandline

MartijnVisser · 2022-09-09T12:45:37Z

@gaborgsomogyi The PR needs to be rebated since it doesn't contains the latest necessary changes to run the CI

gaborgsomogyi · 2022-09-09T13:06:40Z

@MartijnVisser expected something like this. @chenyuzhi459 could you do a rebase to the latest master please?

chenyuzhi459 · 2022-09-09T13:22:42Z

@MartijnVisser expected something like this. @chenyuzhi459 could you do a rebase to the latest master please?

Ok, I have rebased it.

chenyuzhi459 · 2022-09-09T13:26:54Z

@flinkbot run azure

gaborgsomogyi · 2022-09-09T15:41:25Z

@flinkbot run azure

gaborgsomogyi · 2022-09-10T20:10:58Z

Jenkins passed, good to go from my perspective.

rmetzger added the component=<none> label Dec 15, 2021

gaborgsomogyi reviewed Sep 5, 2022

View reviewed changes

gaborgsomogyi reviewed Sep 8, 2022

View reviewed changes

gaborgsomogyi approved these changes Sep 8, 2022

View reviewed changes

chenyuzhi459 and others added 6 commits September 9, 2022 21:09

[FLINK-24907] Support side out late data for interval join

2ff5b33

[FLINK-24907] code style

8002cd2

[FLINK-24907] unchange comment

0f9facf

[FLINK-24907] merge test method

33d8391

[FLINK-24907] Optimize run content

cf83592

[FLINK-24907] Extract sideout method for testing

af9887e

chenyuzhi459 force-pushed the FLINK-24907-sideout-intervaljoin branch from d023ec5 to af9887e Compare September 9, 2022 13:21

gyfora merged commit b515da4 into apache:master Sep 19, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-24907] Support side out late data for interval join #18118

[FLINK-24907] Support side out late data for interval join #18118

chenyuzhi459 commented Dec 15, 2021 •

edited

flinkbot commented Dec 15, 2021

flinkbot commented Dec 15, 2021 •

edited

chenyuzhi459 commented Dec 16, 2021

MartijnVisser commented Aug 31, 2022

gyfora commented Aug 31, 2022

gyfora commented Aug 31, 2022

gaborgsomogyi left a comment

gaborgsomogyi Sep 5, 2022

chenyuzhi459 Sep 7, 2022 •

edited

gaborgsomogyi Sep 8, 2022

gaborgsomogyi Sep 8, 2022

chenyuzhi459 Sep 8, 2022 •

edited

gaborgsomogyi Sep 8, 2022

chenyuzhi459 commented Sep 7, 2022 •

edited

gaborgsomogyi commented Sep 8, 2022

gaborgsomogyi commented Sep 8, 2022

gaborgsomogyi Sep 8, 2022

chenyuzhi459 Sep 8, 2022

gaborgsomogyi commented Sep 8, 2022

gaborgsomogyi left a comment

gaborgsomogyi commented Sep 9, 2022

gaborgsomogyi commented Sep 9, 2022

MartijnVisser commented Sep 9, 2022

gaborgsomogyi commented Sep 9, 2022

chenyuzhi459 commented Sep 9, 2022

chenyuzhi459 commented Sep 9, 2022

gaborgsomogyi commented Sep 9, 2022

gaborgsomogyi commented Sep 10, 2022

[FLINK-24907] Support side out late data for interval join #18118

[FLINK-24907] Support side out late data for interval join #18118

Conversation

chenyuzhi459 commented Dec 15, 2021 • edited

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

flinkbot commented Dec 15, 2021

Automated Checks

Review Progress

flinkbot commented Dec 15, 2021 • edited

CI report:

chenyuzhi459 commented Dec 16, 2021

MartijnVisser commented Aug 31, 2022

gyfora commented Aug 31, 2022

gyfora commented Aug 31, 2022

gaborgsomogyi left a comment

Choose a reason for hiding this comment

gaborgsomogyi Sep 5, 2022

Choose a reason for hiding this comment

chenyuzhi459 Sep 7, 2022 • edited

Choose a reason for hiding this comment

gaborgsomogyi Sep 8, 2022

Choose a reason for hiding this comment

gaborgsomogyi Sep 8, 2022

Choose a reason for hiding this comment

chenyuzhi459 Sep 8, 2022 • edited

Choose a reason for hiding this comment

gaborgsomogyi Sep 8, 2022

Choose a reason for hiding this comment

chenyuzhi459 commented Sep 7, 2022 • edited

gaborgsomogyi commented Sep 8, 2022

gaborgsomogyi commented Sep 8, 2022

gaborgsomogyi Sep 8, 2022

Choose a reason for hiding this comment

chenyuzhi459 Sep 8, 2022

Choose a reason for hiding this comment

gaborgsomogyi commented Sep 8, 2022

gaborgsomogyi left a comment

Choose a reason for hiding this comment

gaborgsomogyi commented Sep 9, 2022

gaborgsomogyi commented Sep 9, 2022

MartijnVisser commented Sep 9, 2022

gaborgsomogyi commented Sep 9, 2022

chenyuzhi459 commented Sep 9, 2022

chenyuzhi459 commented Sep 9, 2022

gaborgsomogyi commented Sep 9, 2022

gaborgsomogyi commented Sep 10, 2022

chenyuzhi459 commented Dec 15, 2021 •

edited

flinkbot commented Dec 15, 2021 •

edited

chenyuzhi459 Sep 7, 2022 •

edited

chenyuzhi459 Sep 8, 2022 •

edited

chenyuzhi459 commented Sep 7, 2022 •

edited