GitHub user shanthoosh opened a pull request:
https://github.com/apache/samza/pull/526
Schedule ZkBarrierChangeHandler into debounce thread for execution.
In existing implementation, `ZkBarrierChangeHandler` is executed from the
`ZkEventThread` and has following drawbacks:
* `ZkWatch` events are buffered into a in-memory queue(maintained by
ZkClient) and delivered one at a time to ZkClient listener implementations. If
the exeuction of a delivered `ZkWatch` event is in progress, then no other
`ZkWatch` event will be delivered to the listeners. If `ZkBarrierChangeHandler`
is executed from `ZkEventThread`, any increase in processing latency will
delay the delivery of other `ZkWatch` events(buffered in in-memory queue of
ZkClient).
* During session expiration(zkConnection error scenario), buffering all
events into `ScheduleAfterDebounceTime` helps us to garbage collect older
generation events(to ensure correctness and not execute older generation
`ZkWatch` events).
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/shanthoosh/samza
schedule_barrier_change_in_debounce_thread
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/samza/pull/526.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 #526
----
commit 7c2540dcb2a9327b342679c02babc7fe7475c0ac
Author: Shanthoosh Venkataraman <svenkataraman@...>
Date: 2018-04-30T04:28:14Z
SAMZA-1692: Standalone stability fixes.
- Currently, on session expiration processorListener with incorrect
generationId is registered with zookeeper(ZkUtils generationId is incremented
on reconnect but the generationId in processorListener is zero all the time).
When a session reconnect happens to a processor successive to leader, leader
expiration event will be skipped. This will prevent leader re-election on a
current leader death and will stall the processors group. Fix is to
reinstantiate and then register processorChangeListener on session expiration.
- Add processorId to debounce thread name (this can aid debugging when
multiple processors are running within a jvm).
- After ScheduleAfterDebounceTime queue is shutdown, don't accept new
schedule requests. Current ZkJobCoordinator shutdown sequence comprise of the
following steps
- Shutdown the ScheduleAfterDebounceTime queue.
- Stop the zkClient and relinquish it's resources.
After we shutdown ScheduleAfterDebounceTime and before zkclient is stopped,
any new operations can be scheduled in ScheduleAfterDebounceTime queue. This
will result in RejectedExecutionException, since executorService is stopped.
```
Caused by: java.util.concurrent.RejectedExecutionException: Task
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask@23f962a8
rejected from java.util.concurrent.ScheduledThreadPoolExecutor@43408be8
```
commit ee27c18502b1e6398b2d3f739033b29c1dfb3cf1
Author: Shanthoosh Venkataraman <santhoshvenkat1988@...>
Date: 2018-05-07T20:23:58Z
Review comments.
commit 0f9a5a810c39f690cfb40ee5f90896e23a6b5ad9
Author: Shanthoosh Venkataraman <santhoshvenkat1988@...>
Date: 2018-05-17T23:25:22Z
Changes:
In current implementation, barrier completion logic is executed from
ZkClient EventThread.
Schedule this logic to be executed from debounce thread.
----
---
