ZKWorkerController

java.lang.Object
- edu.iu.dsc.tws.common.zk.ZKWorkerController

All Implemented Interfaces:

IWorkerController, IWorkerStatusUpdater
```
public class ZKWorkerController
extends java.lang.Object
implements IWorkerController, IWorkerStatusUpdater
```
we assume each worker is assigned a unique ID outside of this class.
We create following directories for a job by Job Master: persistent state znode (directory) ephemeral state znode (directory) default barrier znode (directory) init barrier znode (directory) events znode (directory) We create a persistent znode for each worker under the job persistent state znode. It has WorkerInfo object, the latest WorkerState and worker restart count. Job Master watches the children of this znode for worker state changes. We create an ephemeral znode for each worker under the job ephemeral state znode. The children of this znode is used for watching for worker failures and joins. Job Master watches the children of this znode. Persistent znodes for workers must be created before ephemeral znodes, because, JM gets workerInfo from persistent znode We have two types of barriers: default and init Workers wait at the init barrier when they are starting. They also come back to the init barrier in case of a failure. They all proceed from the init barrier after a failure We create a barrier znode for each worker under the barrier znode, When all workers created their znodes under this directory, all workers arrived on the barrier. Job master watches the children of this znode and informs all workers by publishing an event. Each worker is responsible for deletion of their znodes under the barrier directory, when they proceed through the barrier. Job Master deletes the worker barrier znodes in case of scaling down or job termination.
Events: All events are published on the events znode (directory) as new child znodes with sequential numerical names

Field Summary

Fields
Modifier and Type Field and Description

static java.util.logging.Logger LOG

Fields
Modifier and Type	Field and Description
`static java.util.logging.Logger`	`LOG`

Constructor Summary

Constructors
Constructor and Description
`ZKWorkerController(Config config, java.lang.String jobID, int numberOfWorkers, WorkerInfo workerInfo)` Construct ZKWorkerController but do not initialize yet

Method Summary

All Methods Instance Methods Concrete Methods
Modifier and Type	Method and Description
`boolean`	`addAllJoinedListener(IAllJoinedListener iAllJoinedListener)` add a single IAllJoinedListener if additional IAllJoinedListener tried to be added, do not add and return false
`boolean`	`addFailureListener(IWorkerFailureListener iWorkerFailureListener)` add a single IWorkerFailureListener if additional IWorkerFailureListener tried to be added, do not add and return false
`void`	`addJMFailureListener(IJobMasterFailureListener iJobMasterFailureListener)` TODO: jm restarted implemented, but jm failed not implemented yet Supports multiple IJobMasterFailureListeners
`boolean`	`addScalerListener(IScalerListener iScalerListener)` add a single IScalerListener if additional IScalerListener tried to be added, do not add and return false
`protected java.util.List<WorkerInfo>`	`cloneWorkers()`
`void`	`close()` close all local entities.
`java.util.List<WorkerInfo>`	`getAllWorkers()` get all workers in the job.
`CheckpointingClient`	`getCheckpointingClient()`
`java.util.List<WorkerInfo>`	`getCurrentWorkers()` Get current list of workers This list does not have the workers that have failed or already completed
`IWorkerFailureListener`	`getFailureListener()` Get the failure listener
`java.util.List<WorkerInfo>`	`getJoinedWorkers()` get all joined workers in this job, including the ones finished execution if there are some workers that have not joined yet, they may not be included in this list.
`int`	`getNumberOfCurrentWorkers()` get number of current workers in the job as seen from this worker
`int`	`getNumberOfWorkers()` return the number of all workers in this job, including non-started ones and finished ones
`WorkerInfo`	`getWorkerInfo()` return the WorkerInfo object for this worker
`WorkerInfo`	`getWorkerInfoForID(int id)` return the WorkerInfo object for the given ID
`WorkerState`	`getWorkerStatusForID(int id)` return the status of any worker in the job
`void`	`initialize(int restartCount1, long jsTime)` Initialize this ZKWorkerController Connect to the ZK server create a persistent znode for this worker set this WorkerInfo and its status in the body of that znode create an ephemeral znode for this worker create a cache for the persistent znode of the job for watching job scaling events initialState has to be either: WorkerState.STARTED or WorkerState.RESTARTED
`boolean`	`isRestarted()`
`void`	`setCheckpointingClient(CheckpointingClient checkpointingClient)`
`boolean`	`updateWorkerStatus(WorkerState newStatus)` Update worker status with new state return true if successful
`void`	`waitOnBarrier()` wait for all workers in the job to arrive at this barrier After waiting for the timeout specified in ControllerContext.maxWaitTimeOnBarrier if some workers still could not arrive at the barrier, throw an exception
`void`	`waitOnBarrier(long timeLimit)` All workers create a znode on the barrier directory Job master watches znode creations/removals on this directory when the number of znodes on that directory reaches the number of workers in the job, Job master publishes AllArrivedOnBarrier event Workers proceed when they get this event or when they time out
`void`	`waitOnInitBarrier()` init barrier the same algorithm as the default barrier
`int`	`workerRestartCount()` get worker restartCount zero means starting for the first time

Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

- Field Detail
  - LOG
```
public static final java.util.logging.Logger LOG
```
- Constructor Detail
  - ZKWorkerController
```
public ZKWorkerController(Config config,
                          java.lang.String jobID,
                          int numberOfWorkers,
                          WorkerInfo workerInfo)
```
    Construct ZKWorkerController but do not initialize yet
- Method Detail
  - initialize
```
public void initialize(int restartCount1,
                       long jsTime)
                throws java.lang.Exception
```
    Initialize this ZKWorkerController Connect to the ZK server create a persistent znode for this worker set this WorkerInfo and its status in the body of that znode create an ephemeral znode for this worker create a cache for the persistent znode of the job for watching job scaling events initialState has to be either: WorkerState.STARTED or WorkerState.RESTARTED
    The body of the persistent worker znode will be updated as the status of worker changes from STARTED, COMPLETED,
    
    Throws:
    
    java.lang.Exception
  - setCheckpointingClient
```
public void setCheckpointingClient(CheckpointingClient checkpointingClient)
```
  - isRestarted
```
public boolean isRestarted()
```
  - updateWorkerStatus
```
public boolean updateWorkerStatus(WorkerState newStatus)
```
    Update worker status with new state return true if successful
    Initially worker status is set as STARTED or RESTARTED. Therefore, there is no need to call this method after starting this IWorkerController This method should be called to change worker status to COMPLETED, FAILED, etc.
    
    Specified by:
    
    updateWorkerStatus in interface IWorkerStatusUpdater
    
    Returns:
  - getWorkerStatusForID
```
public WorkerState getWorkerStatusForID(int id)
```
    Description copied from interface: IWorkerStatusUpdater
    
    return the status of any worker in the job
    
    Specified by:
    
    getWorkerStatusForID in interface IWorkerStatusUpdater
    
    Returns:
  - getWorkerInfo
```
public WorkerInfo getWorkerInfo()
```
    Description copied from interface: IWorkerController
    
    return the WorkerInfo object for this worker
    
    Specified by:
    
    getWorkerInfo in interface IWorkerController
  - getWorkerInfoForID
```
public WorkerInfo getWorkerInfoForID(int id)
```
    Description copied from interface: IWorkerController
    
    return the WorkerInfo object for the given ID
    
    Specified by:
    
    getWorkerInfoForID in interface IWorkerController
  - getNumberOfWorkers
```
public int getNumberOfWorkers()
```
    Description copied from interface: IWorkerController
    
    return the number of all workers in this job, including non-started ones and finished ones
    
    Specified by:
    
    getNumberOfWorkers in interface IWorkerController
  - getJoinedWorkers
```
public java.util.List<WorkerInfo> getJoinedWorkers()
```
    Description copied from interface: IWorkerController
    
    get all joined workers in this job, including the ones finished execution if there are some workers that have not joined yet, they may not be included in this list. users can compare the total number of workers to the size of this list and understand whether there are non-joined workers
    
    Specified by:
    
    getJoinedWorkers in interface IWorkerController
  - getAllWorkers
```
public java.util.List<WorkerInfo> getAllWorkers()
                                         throws TimeoutException
```
    Description copied from interface: IWorkerController
    
    get all workers in the job. If some workers has not joined the job yet, wait for them. After waiting for the timeout specified in ControllerContext.maxWaitTimeForAllToJoin if some workers still could not join, throw an exception
    return all workers in the job including the ones that have already left, if any
    
    Specified by:
    
    getAllWorkers in interface IWorkerController
    
    Throws:
    
    TimeoutException
  - cloneWorkers
```
protected java.util.List<WorkerInfo> cloneWorkers()
```
  - addFailureListener
```
public boolean addFailureListener(IWorkerFailureListener iWorkerFailureListener)
```
    add a single IWorkerFailureListener if additional IWorkerFailureListener tried to be added, do not add and return false
  - addAllJoinedListener
```
public boolean addAllJoinedListener(IAllJoinedListener iAllJoinedListener)
```
    add a single IAllJoinedListener if additional IAllJoinedListener tried to be added, do not add and return false
  - addScalerListener
```
public boolean addScalerListener(IScalerListener iScalerListener)
```
    add a single IScalerListener if additional IScalerListener tried to be added, do not add and return false
  - addJMFailureListener
```
public void addJMFailureListener(IJobMasterFailureListener iJobMasterFailureListener)
```
    TODO: jm restarted implemented, but jm failed not implemented yet Supports multiple IJobMasterFailureListeners
  - getCurrentWorkers
```
public java.util.List<WorkerInfo> getCurrentWorkers()
```
    Get current list of workers This list does not have the workers that have failed or already completed
  - getNumberOfCurrentWorkers
```
public int getNumberOfCurrentWorkers()
                              throws java.lang.Exception
```
    get number of current workers in the job as seen from this worker
    
    Throws:
    
    java.lang.Exception
  - waitOnBarrier
```
public void waitOnBarrier()
                   throws TimeoutException
```
    Description copied from interface: IWorkerController
    
    wait for all workers in the job to arrive at this barrier After waiting for the timeout specified in ControllerContext.maxWaitTimeOnBarrier if some workers still could not arrive at the barrier, throw an exception
    
    Specified by:
    
    waitOnBarrier in interface IWorkerController
    
    Throws:
    
    TimeoutException
  - waitOnBarrier
```
public void waitOnBarrier(long timeLimit)
                   throws TimeoutException
```
    All workers create a znode on the barrier directory Job master watches znode creations/removals on this directory when the number of znodes on that directory reaches the number of workers in the job, Job master publishes AllArrivedOnBarrier event Workers proceed when they get this event or when they time out
    Workers remove their znodes after they proceed through the barrier so that they can wait on the barrier again Workers are responsible for creating and removing znodes on the barrier Job master removes barrier znode after the job completion or scale down. if timeout is reached, throws TimeoutException.
    
    Specified by:
    
    waitOnBarrier in interface IWorkerController
    
    Throws:
    
    TimeoutException
  - waitOnInitBarrier
```
public void waitOnInitBarrier()
                       throws TimeoutException
```
    init barrier the same algorithm as the default barrier
    
    Specified by:
    
    waitOnInitBarrier in interface IWorkerController
    
    Throws:
    
    TimeoutException
  - workerRestartCount
```
public int workerRestartCount()
```
    Description copied from interface: IWorkerController
    
    get worker restartCount zero means starting for the first time
    
    Specified by:
    
    workerRestartCount in interface IWorkerController
  - getFailureListener
```
public IWorkerFailureListener getFailureListener()
```
    Description copied from interface: IWorkerController
    
    Get the failure listener
    
    Specified by:
    
    getFailureListener in interface IWorkerController
    
    Returns:
    
    the failure listener
  - getCheckpointingClient
```
public CheckpointingClient getCheckpointingClient()
```
    Specified by:
    
    getCheckpointingClient in interface IWorkerController
  - close
```
public void close()
```
    close all local entities.

Class ZKWorkerController

Field Summary

Constructor Summary

Method Summary

Methods inherited from class java.lang.Object

Field Detail

LOG

Constructor Detail

ZKWorkerController

Method Detail

initialize

setCheckpointingClient

isRestarted

updateWorkerStatus

getWorkerStatusForID

getWorkerInfo

getWorkerInfoForID

getNumberOfWorkers

getJoinedWorkers

getAllWorkers

cloneWorkers

addFailureListener

addAllJoinedListener

addScalerListener

addJMFailureListener

getCurrentWorkers

getNumberOfCurrentWorkers

waitOnBarrier

waitOnBarrier

waitOnInitBarrier

workerRestartCount

getFailureListener

getCheckpointingClient

close