[FLINK-18685] [runtime] Make MiniClusterJobClient#getAccumulators non-blocking in Streaming mode #14558

echauchot · 2021-01-05T10:00:54Z

What is the purpose of the change

Make MiniClusterJobClient#getAccumulators() non-blocking in Streaming mode

Brief change log

Get the serialized accumulators from the execution graph rather than getting the accumulators from the JobExecutionResult.

Verifying this change

Changed AccumulatorLiveITCase so that the verification of the accumulators is done by calling getAccumulators() with both ClusterClient (as it was in previous test version) and MiniClusterJobClient

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, Kubernetes/Yarn/Mesos, 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? it was a bug that was not compliant with the javadocs. Now it is, so javadoc is untouched by this PR.

flinkbot · 2021-01-05T10:03:45Z

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 aa00e2a (Fri May 28 08:14:28 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-01-05T10:16:18Z

CI report:

aa00e2a 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

echauchot · 2021-01-05T16:28:06Z

R:@rmetzger. There are some test failures that I missed during my local tests. I'll ping you when this PR is ready for review.

echauchot · 2021-01-08T15:19:53Z

@rmetzger I fixed one of the failing tests. But for UnalignedCheckpointCompatibilityITCase savepoint tests, I know we need to get the accumulators from JobExecutionResult for the test to work. But I struggle to find the proper condition when to get the accumulators from JobExecutionResult rather than getting them from the ExecutionGraph as usual. I don't know Flink enough yet, can you give me a clue on this condition?

gaoyunhaii · 2021-01-12T08:42:18Z

flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterJobClient.java

+        //TODO: this is not the only case when we need to get the accumulators from JobExecutionResult.
+        // It is needed also for UnalignedCheckpointCompatibilityITCase savepoints tests to pass.
+        // What is the complete condition ?
+        if (!miniCluster.isRunning()) {


I'm wondering if the miniCluster.isRunning() might not be the proper condition ? Since a mini-cluster could be viewed as a cluster that could run multiple jobs, the miniCluster would still be able to run after the job is done logically. Perhaps we could use jobResultFuture.isDone() ?

agree miniCluster.isRunning() is more a side effect of PerJobMiniClusterFactoryTest.testJobExecution stopping the minicluster when the job is done. jobResultFuture.isDone() is definitely a better condition.

gaoyunhaii · 2021-01-12T08:56:45Z

@rmetzger I fixed one of the failing tests. But for UnalignedCheckpointCompatibilityITCase savepoint tests, I know we need to get the accumulators from JobExecutionResult for the test to work. But I struggle to find the proper condition when to get the accumulators from JobExecutionResult rather than getting them from the ExecutionGraph as usual. I don't know Flink enough yet, can you give me a clue on this condition?

Hi @echauchot, from my point of view, there would be two possible semantics when users want to acquire accumulators:

Users want to retrieve the current value of accumulators immediately.
Users knows the job is bounded and want to acquire the final accumulator values after the job is finished.

Previously getAccumulator() takes the second semantics, and now we want to move it to the first semantics to make it consistent with the other client implementation. Therefore, if users still want to use the second semantics, I think they would need to call getJobExecutionResult().thenApply(JobExecutionResult::getAllAccumulatorResults) explicitly. Thus for me to fix the failed tests we need to modify the tests since we would not be able to distinguish the first and second semantics automatically. What do you think about it ?

echauchot · 2021-01-14T11:12:11Z

@rmetzger I fixed one of the failing tests. But for UnalignedCheckpointCompatibilityITCase savepoint tests, I know we need to get the accumulators from JobExecutionResult for the test to work. But I struggle to find the proper condition when to get the accumulators from JobExecutionResult rather than getting them from the ExecutionGraph as usual. I don't know Flink enough yet, can you give me a clue on this condition?

Hi @echauchot, from my point of view, there would be two possible semantics when users want to acquire accumulators:

Users want to retrieve the current value of accumulators immediately.

Users knows the job is bounded and want to acquire the final accumulator values after the job is finished.

Previously getAccumulator() takes the second semantics, and now we want to move it to the first semantics to make it consistent with the other client implementation. Therefore, if users still want to use the second semantics, I think they would need to call getJobExecutionResult().thenApply(JobExecutionResult::getAllAccumulatorResults) explicitly. Thus for me to fix the failed tests we need to modify the tests since we would not be able to distinguish the first and second semantics automatically. What do you think about it ?

@gaoyunhaii thanks for your comments. I totally agree with them. I also concluded that the only way was to change UnalignedCheckpointCompatibilityITCase to wait for the job end. Thanks for the confirmation !

echauchot · 2021-01-14T11:15:53Z

@flinkbot run azure

echauchot · 2021-01-14T11:23:50Z

@flinkbot run travis

echauchot · 2021-01-14T11:35:43Z

@rmetzger this PR is ready for review PTAL.

…ocking and returns correct accumulators.

…t requests the accumulators from the JobExecutionResult as before

echauchot · 2021-01-22T13:31:25Z

Friendly ping: @rmetzger do you have time to review/merge this PR? if not, can you ping someone to look ? Thanks.

echauchot · 2021-01-27T11:32:57Z

@tillrohrmann can you please take a look at this PR as I saw you in the git history of this part of code ? Thanks

rmetzger · 2021-01-27T11:44:06Z

@echauchot I'm really sorry for the delay. I'll take a look at this PR today!

echauchot · 2021-01-27T13:49:01Z

@rmetzger don't worry, I know what it is :)

rmetzger

Thanks a lot for this fix!

I also manually verified the fix according to the Jira description.

I'll merge this now!

rmetzger added the review=description? label Jan 5, 2021

rmetzger added the component=API/DataStream label Jan 5, 2021

echauchot force-pushed the FLINK-18685-blockingGetAccumulators branch from 4480110 to 5543d0d Compare January 8, 2021 15:15

gaoyunhaii reviewed Jan 12, 2021

View reviewed changes

echauchot force-pushed the FLINK-18685-blockingGetAccumulators branch from 5543d0d to 792763a Compare January 14, 2021 11:07

echauchot force-pushed the FLINK-18685-blockingGetAccumulators branch from 792763a to c875ca9 Compare January 14, 2021 11:15

echauchot added 3 commits January 14, 2021 13:07

[FLINK-18685] Make MiniClusterJobClient#getAccumulators non-blocking

2400f83

[FLINK-18685] Test that MiniClusterJobClient.getAccumulator is not bl…

c88b6ca

…ocking and returns correct accumulators.

[FLINK-18685] Change UnalignedCheckpointCompatibilityITCase so that i…

aa00e2a

…t requests the accumulators from the JobExecutionResult as before

echauchot force-pushed the FLINK-18685-blockingGetAccumulators branch from c875ca9 to aa00e2a Compare January 14, 2021 12:07

rmetzger approved these changes Jan 27, 2021

View reviewed changes

rmetzger closed this in beae8e7 Jan 27, 2021

echauchot deleted the FLINK-18685-blockingGetAccumulators branch February 3, 2021 11:58

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-18685] [runtime] Make MiniClusterJobClient#getAccumulators non-blocking in Streaming mode #14558

[FLINK-18685] [runtime] Make MiniClusterJobClient#getAccumulators non-blocking in Streaming mode #14558

echauchot commented Jan 5, 2021

flinkbot commented Jan 5, 2021 •

edited

Loading

flinkbot commented Jan 5, 2021 •

edited

Loading

echauchot commented Jan 5, 2021

echauchot commented Jan 8, 2021 •

edited

Loading

gaoyunhaii Jan 12, 2021 •

edited

Loading

echauchot Jan 14, 2021

gaoyunhaii commented Jan 12, 2021

echauchot commented Jan 14, 2021

echauchot commented Jan 14, 2021

echauchot commented Jan 14, 2021

echauchot commented Jan 14, 2021

echauchot commented Jan 22, 2021

echauchot commented Jan 27, 2021

rmetzger commented Jan 27, 2021

echauchot commented Jan 27, 2021

rmetzger left a comment

[FLINK-18685] [runtime] Make MiniClusterJobClient#getAccumulators non-blocking in Streaming mode #14558

[FLINK-18685] [runtime] Make MiniClusterJobClient#getAccumulators non-blocking in Streaming mode #14558

Conversation

echauchot commented Jan 5, 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

flinkbot commented Jan 5, 2021 • edited Loading

Automated Checks

Review Progress

flinkbot commented Jan 5, 2021 • edited Loading

CI report:

echauchot commented Jan 5, 2021

echauchot commented Jan 8, 2021 • edited Loading

gaoyunhaii Jan 12, 2021 • edited Loading

Choose a reason for hiding this comment

echauchot Jan 14, 2021

Choose a reason for hiding this comment

gaoyunhaii commented Jan 12, 2021

echauchot commented Jan 14, 2021

echauchot commented Jan 14, 2021

echauchot commented Jan 14, 2021

echauchot commented Jan 14, 2021

echauchot commented Jan 22, 2021

echauchot commented Jan 27, 2021

rmetzger commented Jan 27, 2021

echauchot commented Jan 27, 2021

rmetzger left a comment

Choose a reason for hiding this comment

flinkbot commented Jan 5, 2021 •

edited

Loading

flinkbot commented Jan 5, 2021 •

edited

Loading

echauchot commented Jan 8, 2021 •

edited

Loading

gaoyunhaii Jan 12, 2021 •

edited

Loading