GitHub user mxm opened a pull request:
https://github.com/apache/flink/pull/925
[FLINK-2371] improve AccumulatorLiveITCase
Instead of using Thread.sleep() to synchronize the checks of the
accumulator values, we rely on message passing here to synchronize the
task process.
Therefore, we let the task process signal to the task manager that it
has updated its accumulator values. The task manager lets the job
manager know and sends out the heartbeat which contains the
accumulators. When the job manager receives the accumulators and has
been notified previously, it sends a message to the subscribed test case
with the current accumulators.
This assures that all processes are always synchronized correctly and we
can verify the live accumulator results correctly.
In the course of rewriting the test, I had to change two things in the
implementation:
a) User accumulators are now immediately serialized as well. Otherwise,
Akka does not serialize in local one VM setups and passes the live
accumulator map through.
b) The asynchronous update of the accumulators can be disabled for
tests. This was necessary because we cannot guarantee when the Future
for updating the accumulators is executed. In real setups this is
neglectable.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/mxm/flink live-accumulators
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/925.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #925
----
commit 44687e783065a4157d2d3a695d9e94070ca6e8cd
Author: Maximilian Michels <m...@apache.org>
Date: 2015-07-20T09:55:11Z
[FLINK-2371] improve AccumulatorLiveITCase
Instead of using Thread.sleep() to synchronize the checks of the
accumulator values, we rely on message passing here to synchronize the
task process.
Therefore, we let the task process signal to the task manager that it
has updated its accumulator values. The task manager lets the job
manager know and sends out the heartbeat which contains the
accumulators. When the job manager receives the accumulators and has
been notified previously, it sends a message to the subscribed test case
with the current accumulators.
This assures that all processes are always synchronized correctly and we
can verify the live accumulator results correctly.
In the course of rewriting the test, I had to change two things in the
implementation:
a) User accumulators are now immediately serialized as well. Otherwise,
Akka does not serialize in local one VM setups and passes the live
accumulator map through.
b) The asynchronous update of the accumulators can be disabled for
tests. This was necessary because we cannot guarantee when the Future
for updating the accumulators is executed. In real setups this is
neglectable.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
