public List < WorkerStatus > findAllWorkerStatusForPlan ( Plan plan , List < String > otherWorkers ) throws WorkerDaoException { List < WorkerStatus > results = new ArrayList < WorkerStatus > ( ) ; try { Stat preCheck = framework . getCuratorFramework ( ) . checkExists ( ) . forPath ( PLAN_WORKERS_ZK + "/" + plan . getName ( ) ) ; if ( preCheck == null ) { return results ; } } catch ( Exception e1 ) { LOGGER . warn ( e1 ) ; throw new WorkerDaoException ( e1 ) ; } for ( String worker : otherWorkers ) { String lookAtPath = PLAN_WORKERS_ZK + "/" + plan . getName ( ) + "/" + worker ; try { Stat stat = framework . getCuratorFramework ( ) . checkExists ( ) . forPath ( lookAtPath ) ; byte [ ] data = framework . getCuratorFramework ( ) . getData ( ) . storingStatIn ( stat ) . forPath ( lookAtPath ) ; results . add ( MAPPER . readValue ( data , WorkerStatus . class ) ) ; } catch ( Exception e ) { LOGGER . warn ( e ) ; throw new WorkerDaoException ( e ) ; } } return results ; }
Gets the status of each worker . The status contains the partitionId being consumed . This information helps the next worker bind to an unconsumed partition