@Test public void testAssignment() throws Exception { TaskSetup taskSetup = new TaskSetup(zookeeperAddress); final String clusterName = "test-s4-cluster"; taskSetup.clean(clusterName); taskSetup.setup(clusterName, 10); final CountDownLatch latch = new CountDownLatch(10); for (int i = 0; i < 10; i++) { Runnable runnable = new Runnable() { @Override public void run() { AssignmentFromZK assignmentFromZK; try { assignmentFromZK = new AssignmentFromZK(clusterName, zookeeperAddress, 30000, 30000); ClusterNode assignClusterNode = assignmentFromZK.assignClusterNode(); latch.countDown(); } catch (Exception e) { e.printStackTrace(); } } }; Thread t = new Thread(runnable); t.start(); } boolean await = latch.await(3, TimeUnit.SECONDS); assertEquals(true, await); }
@Override public ClusterNode assignClusterNode() { lock.lock(); try { while (!currentlyOwningTask.get()) { taskAcquired.awaitUninterruptibly(); } } catch (Exception e) { logger.error("Exception while waiting for task to be acquired"); return null; } finally { lock.unlock(); } return clusterNodeRef.get(); }
AssignmentFromZK implements Assignment, IZkChildListener, IZkStateListener, IZkDataListener { @Override public ClusterNode assignClusterNode() { lock.lock(); try { while (!currentlyOwningTask.get()) { taskAcquired.awaitUninterruptibly(); } } catch (Exception e) { logger.error("Exception while waiting for task to be acquired"); return null; } finally { lock.unlock(); } return clusterNodeRef.get(); } }
AssignmentFromZK implements Assignment, IZkChildListener, IZkStateListener, IZkDataListener { @Override public ClusterNode assignClusterNode() { lock.lock(); try { while (!currentlyOwningTask.get()) { taskAcquired.awaitUninterruptibly(); } } catch (Exception e) { logger.error("Exception while waiting for task to be acquired"); return null; } finally { lock.unlock(); } return clusterNodeRef.get(); } @Inject  AssignmentFromZK(@Named("cluster.name") String clusterName,
            @Named("cluster.zk_address") String zookeeperAddress,
            @Named("cluster.zk_session_timeout") int sessionTimeout,
            @Named("cluster.zk_connection_timeout") int connectionTimeout); }
AssignmentFromZK implements Assignment, IZkChildListener, IZkStateListener, IZkDataListener { @Override public ClusterNode assignClusterNode() { lock.lock(); try { while (!currentlyOwningTask.get()) { taskAcquired.awaitUninterruptibly(); } } catch (Exception e) { logger.error("Exception while waiting for task to be acquired"); return null; } finally { lock.unlock(); } return clusterNodeRef.get(); } @Inject  AssignmentFromZK(@Named("cluster.name") String clusterName,
            @Named("cluster.zk_address") String zookeeperAddress,
            @Named("cluster.zk_session_timeout") int sessionTimeout,
            @Named("cluster.zk_connection_timeout") int connectionTimeout); @Override void handleDataChange(String dataPath, Object data); @Override void handleDataDeleted(String dataPath); @Override void handleStateChanged(KeeperState state); @Override void handleNewSession(); @Override void handleChildChange(String parentPath, List<String> currentChilds); @Override ClusterNode assignClusterNode(); }
AssignmentFromZK implements Assignment, IZkChildListener, IZkStateListener, IZkDataListener { @Override public ClusterNode assignClusterNode() { lock.lock(); try { while (!currentlyOwningTask.get()) { taskAcquired.awaitUninterruptibly(); } } catch (Exception e) { logger.error("Exception while waiting for task to be acquired"); return null; } finally { lock.unlock(); } return clusterNodeRef.get(); } @Inject  AssignmentFromZK(@Named("cluster.name") String clusterName,
            @Named("cluster.zk_address") String zookeeperAddress,
            @Named("cluster.zk_session_timeout") int sessionTimeout,
            @Named("cluster.zk_connection_timeout") int connectionTimeout); @Override void handleDataChange(String dataPath, Object data); @Override void handleDataDeleted(String dataPath); @Override void handleStateChanged(KeeperState state); @Override void handleNewSession(); @Override void handleChildChange(String parentPath, List<String> currentChilds); @Override ClusterNode assignClusterNode();  }