Change in accumutors semantics with jobClient

classic Classic list List threaded Threaded
1 message Options
Reply | Threaded
Open this post in threaded view
|

Change in accumutors semantics with jobClient

Etienne Chauchot
Hi all,

I did a fix some time ago regarding accumulators:
the/JobClient.getAccumulators()/ was infinitely  blocking in local
environment for a streaming job (1). The change (2) consisted of giving
the current accumulators value for the running job. And when fixing this
in the PR, it appeared that I had to change the accumulators semantics
with /JobClient/ and I just realized that I forgot to bring this back to
the ML:

Previously /JobClient/ assumed that getAccumulator() was called on a
bounded pipeline and that the user wanted to acquire the *final
accumulator values* after the job is finished.

But now it returns the *current value of accumulators* immediately to be
compatible with unbounded pipelines.

If it is run on a bounded pipeline, then to get the final accumulator
values after the job is finished, one needs to call
/getJobExecutionResult().thenApply(JobExecutionResult::getAllAccumulatorResults)/

(1): https://issues.apache.org/jira/browse/FLINK-18685

(2): https://github.com/apache/flink/pull/14558#


Cheers,

Etienne