int claimResource ( ZooKeeper zookeeper , String poolNode , int poolSize ) throws KeeperException , InterruptedException { logger . debug ( "Trying to claim a resource." ) ; List < String > claimedResources = zookeeper . getChildren ( poolNode , false ) ; if ( claimedResources . size ( ) >= poolSize ) { logger . debug ( "No resources available at the moment (pool size: {}), waiting." , poolSize ) ; // No resources available. Wait for a resource to become available. final CountDownLatch latch = new CountDownLatch ( 1 ) ; zookeeper . getChildren ( poolNode , event -> latch . countDown ( ) ) ; latch . await ( ) ; return claimResource ( zookeeper , poolNode , poolSize ) ; } // Try to claim an available resource. for ( int i = 0 ; i < poolSize ; i ++ ) { String resourcePath = Integer . toString ( i ) ; if ( ! claimedResources . contains ( resourcePath ) ) { String node ; try { logger . debug ( "Trying to claim seemingly available resource {}." , resourcePath ) ; node = zookeeper . create ( poolNode + "/" + resourcePath , new byte [ 0 ] , ZooDefs . Ids . OPEN_ACL_UNSAFE , CreateMode . EPHEMERAL ) ; } catch ( KeeperException e ) { if ( e . code ( ) == KeeperException . Code . NODEEXISTS ) { // Failed to claim this resource for some reason. continue ; } else { // Unexpected failure. throw e ; } } zookeeper . exists ( node , event -> { if ( event . getType ( ) == EventType . NodeDeleted ) { // Invalidate our claim when the node is deleted by some other process. logger . debug ( "Resource-claim node unexpectedly deleted ({})" , resource ) ; close ( true ) ; } } ) ; logger . debug ( "Successfully claimed resource {}." , resourcePath ) ; return i ; } } return claimResource ( zookeeper , poolNode , poolSize ) ; }
Try to claim an available resource from the resource pool .