[FLINK-14890] [tests] Add missing test harnesses for broadcast functions

[afedulov]

What is the purpose of the change

Adds missing test harnesses for broadcast functions

Brief change log

Two new test harnesses are added - one for keyed one for non-keyed broadcast functions.

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): 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): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, 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]

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 6484be5 (Wed Dec 04 15:16:52 UTC 2019)

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]

CI report:

• 2a81c89 : SUCCESS Build
• 0082506 : FAILURE Build
• 2d8039c : UNKNOWN
• 76da268 : SUCCESS Build
• a5ba7d7 : SUCCESS Build
• c14d233 : CANCELED Build
• 6484be5 : FAILURE Build

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run travis re-run the last Travis build

[aljoscha]

Thanks for your contribution!

I think this mixes up to (potentially) orthogonal issues:

1. The addition of convenience methods for creating harnesses for (Keyed)BroadcastProcessFunction. These don't exist for regular (Keyed)ProcessFunctions as of now, where we just have operators and the (Keyed)TwoInputStreamOperatorTestHarness.
2. The addition of an operator test harness for operators with broadcast input. Now that I look at the code I think it's even possible to re-use the (Keyed)TwoInputStreamOperatorTestHarness for this, it is just that the method will be called processElement2() and not processBroadcastElement().

I would not be opposed to fixing both issues, but they should be considered separately (at least separate commits). If we want to fix 1. we should also add convenience methods for the other user functions but that seems to be a bigger undertaking.

Also, we should have tests for newly added code.

[afedulov]

@aljoscha
Thanks for the feedback!
I have addressed the comments. I will follow up with the tests if the general structure is OK.

[aljoscha]

I would prefer method names like forProcessFunction(), forKeyedProcessFunction() and so on.

[aljoscha]

I would prefer KeyedBroadcastOperatorTestHarness (and BroadcastOperatorTestHarness for the other one). These names are more generic and we don't have to tie the implementation to CoBroadcastWithKeyedOperator, you could put any operator in one of these harnesses.

[afedulov]

Addressed the comments. One question - I have nothing smart to say in the javadocs about throws Exception in the utility methods signatures coming from the respective wrapped test harnesses, any ideas?

[aljoscha]

The utility functions look very good now! ☕Could you please also add a simple unit test that checks each of the functions. Pushing in an element and verifying the result should be enough here.

And please change the titles of the PRs to include [FLINK-14890] at the beginning.

Regarding the exceptions, I also don't have anything clever to say. We could even think about wrapping them in a RuntimeException and not adding them to the method signature.

[aljoscha]

I would relax this constraint and maybe say A test harness for testing a {@link TwoInputStreamOperator} in a broadcast context.

[afedulov]

Relaxing this constraint and switching to TwoInputStreamOperator would prevent access to the BroadcastState via the current method getBroadcastState() in the new test harnesses. It seemed to me like this ability would be handy from the user perspective. What do you think?

[aljoscha]

The code below, i.e.

twoInputOperator.getOperatorStateBackend().getBroadcastState(stateDescriptor);

works just as well on a regular TwoInputStreamOperator. Is that what you're referring to?

[aljoscha]

See above, this could be TwoInputStreamOperator.

[aljoscha]

Same as for the other operator.

[afedulov]

I was also initially thinking about wrapping it into a RuntimeException, but given the fact that these methods will be called in tests, which typically are declared with throws Exception themselves, I am not sure about the benefit. All of the useful methods from the harnesses like processElement already throw exceptions anyhow.

[aljoscha]

Thanks! I merged this and put some minor changes on top. 👌