@Test public void testLockingWithUnsupportedMode() throws Exception { Map<String, String> parameters = new HashMap<>(); parameters.put(LockManagingProcess.PARAM_NODE_ID, "nodeid"); parameters.put(LockManagingProcess.PARAM_LOCK_MODE, "unsupported"); exception.expect(IllegalArgumentException.class); lockManager.run(portBindings, conf, parameters); }
@Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { Preconditions.checkArgument(parameters.containsKey(PARAM_NODE_ID), "node id not provided!"); Preconditions.checkArgument(parameters.containsKey(PARAM_LOCK_MODE), "lock mode not provided!"); String zkConnectionString = conf.get(ZKFailoverController.ZK_QUORUM_KEY); Preconditions.checkArgument(StringUtils.isNotBlank(zkConnectionString), "zookeeper quorum is unknown, invalid '%s' property value: %s", ZKFailoverController.ZK_QUORUM_KEY, zkConnectionString); int sessionTimeout = parameters.containsKey(PARAM_ZK_SESSION_TIMEOUT)? Integer.valueOf(parameters.get(PARAM_ZK_SESSION_TIMEOUT)) : DEFAULT_SESSION_TIMEOUT; final ZooKeeper zooKeeper = new ZooKeeper(zkConnectionString, sessionTimeout, (e) -> { }); if (zooKeeper.exists(DEFAULT_ROOT_NODE, false) == null) { log.info("initializing root node: " + DEFAULT_ROOT_NODE); try { zooKeeper.create(DEFAULT_ROOT_NODE, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); log.info("root node initialized"); } catch(NodeExistsException e) { log.warn("root node '" + DEFAULT_ROOT_NODE + "' was already created by different process"); } } final String nodePath = LockManagingProcessUtils.generatePath(parameters.get(PARAM_NODE_ID), DEFAULT_ROOT_NODE); switch(LockMode.valueOf(parameters.get(PARAM_LOCK_MODE))) { case obtain: { LockManagingProcessUtils.obtain(zooKeeper, nodePath); break; } case release: { LockManagingProcessUtils.release(zooKeeper, nodePath); break; } default: { throw new InvalidParameterException("unsupported lock mode: " + parameters.get(PARAM_LOCK_MODE)); } } }
LockManagingProcess implements eu.dnetlib.iis.common.java.Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { Preconditions.checkArgument(parameters.containsKey(PARAM_NODE_ID), "node id not provided!"); Preconditions.checkArgument(parameters.containsKey(PARAM_LOCK_MODE), "lock mode not provided!"); String zkConnectionString = conf.get(ZKFailoverController.ZK_QUORUM_KEY); Preconditions.checkArgument(StringUtils.isNotBlank(zkConnectionString), "zookeeper quorum is unknown, invalid '%s' property value: %s", ZKFailoverController.ZK_QUORUM_KEY, zkConnectionString); int sessionTimeout = parameters.containsKey(PARAM_ZK_SESSION_TIMEOUT)? Integer.valueOf(parameters.get(PARAM_ZK_SESSION_TIMEOUT)) : DEFAULT_SESSION_TIMEOUT; final ZooKeeper zooKeeper = new ZooKeeper(zkConnectionString, sessionTimeout, (e) -> { }); if (zooKeeper.exists(DEFAULT_ROOT_NODE, false) == null) { log.info("initializing root node: " + DEFAULT_ROOT_NODE); try { zooKeeper.create(DEFAULT_ROOT_NODE, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); log.info("root node initialized"); } catch(NodeExistsException e) { log.warn("root node '" + DEFAULT_ROOT_NODE + "' was already created by different process"); } } final String nodePath = LockManagingProcessUtils.generatePath(parameters.get(PARAM_NODE_ID), DEFAULT_ROOT_NODE); switch(LockMode.valueOf(parameters.get(PARAM_LOCK_MODE))) { case obtain: { LockManagingProcessUtils.obtain(zooKeeper, nodePath); break; } case release: { LockManagingProcessUtils.release(zooKeeper, nodePath); break; } default: { throw new InvalidParameterException("unsupported lock mode: " + parameters.get(PARAM_LOCK_MODE)); } } } }
LockManagingProcess implements eu.dnetlib.iis.common.java.Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { Preconditions.checkArgument(parameters.containsKey(PARAM_NODE_ID), "node id not provided!"); Preconditions.checkArgument(parameters.containsKey(PARAM_LOCK_MODE), "lock mode not provided!"); String zkConnectionString = conf.get(ZKFailoverController.ZK_QUORUM_KEY); Preconditions.checkArgument(StringUtils.isNotBlank(zkConnectionString), "zookeeper quorum is unknown, invalid '%s' property value: %s", ZKFailoverController.ZK_QUORUM_KEY, zkConnectionString); int sessionTimeout = parameters.containsKey(PARAM_ZK_SESSION_TIMEOUT)? Integer.valueOf(parameters.get(PARAM_ZK_SESSION_TIMEOUT)) : DEFAULT_SESSION_TIMEOUT; final ZooKeeper zooKeeper = new ZooKeeper(zkConnectionString, sessionTimeout, (e) -> { }); if (zooKeeper.exists(DEFAULT_ROOT_NODE, false) == null) { log.info("initializing root node: " + DEFAULT_ROOT_NODE); try { zooKeeper.create(DEFAULT_ROOT_NODE, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); log.info("root node initialized"); } catch(NodeExistsException e) { log.warn("root node '" + DEFAULT_ROOT_NODE + "' was already created by different process"); } } final String nodePath = LockManagingProcessUtils.generatePath(parameters.get(PARAM_NODE_ID), DEFAULT_ROOT_NODE); switch(LockMode.valueOf(parameters.get(PARAM_LOCK_MODE))) { case obtain: { LockManagingProcessUtils.obtain(zooKeeper, nodePath); break; } case release: { LockManagingProcessUtils.release(zooKeeper, nodePath); break; } default: { throw new InvalidParameterException("unsupported lock mode: " + parameters.get(PARAM_LOCK_MODE)); } } }  }
LockManagingProcess implements eu.dnetlib.iis.common.java.Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { Preconditions.checkArgument(parameters.containsKey(PARAM_NODE_ID), "node id not provided!"); Preconditions.checkArgument(parameters.containsKey(PARAM_LOCK_MODE), "lock mode not provided!"); String zkConnectionString = conf.get(ZKFailoverController.ZK_QUORUM_KEY); Preconditions.checkArgument(StringUtils.isNotBlank(zkConnectionString), "zookeeper quorum is unknown, invalid '%s' property value: %s", ZKFailoverController.ZK_QUORUM_KEY, zkConnectionString); int sessionTimeout = parameters.containsKey(PARAM_ZK_SESSION_TIMEOUT)? Integer.valueOf(parameters.get(PARAM_ZK_SESSION_TIMEOUT)) : DEFAULT_SESSION_TIMEOUT; final ZooKeeper zooKeeper = new ZooKeeper(zkConnectionString, sessionTimeout, (e) -> { }); if (zooKeeper.exists(DEFAULT_ROOT_NODE, false) == null) { log.info("initializing root node: " + DEFAULT_ROOT_NODE); try { zooKeeper.create(DEFAULT_ROOT_NODE, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); log.info("root node initialized"); } catch(NodeExistsException e) { log.warn("root node '" + DEFAULT_ROOT_NODE + "' was already created by different process"); } } final String nodePath = LockManagingProcessUtils.generatePath(parameters.get(PARAM_NODE_ID), DEFAULT_ROOT_NODE); switch(LockMode.valueOf(parameters.get(PARAM_LOCK_MODE))) { case obtain: { LockManagingProcessUtils.obtain(zooKeeper, nodePath); break; } case release: { LockManagingProcessUtils.release(zooKeeper, nodePath); break; } default: { throw new InvalidParameterException("unsupported lock mode: " + parameters.get(PARAM_LOCK_MODE)); } } }  @Override Map<String, PortType> getInputPorts(); @Override Map<String, PortType> getOutputPorts(); @Override void run(PortBindings portBindings, Configuration conf,
			Map<String, String> parameters); }
LockManagingProcess implements eu.dnetlib.iis.common.java.Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { Preconditions.checkArgument(parameters.containsKey(PARAM_NODE_ID), "node id not provided!"); Preconditions.checkArgument(parameters.containsKey(PARAM_LOCK_MODE), "lock mode not provided!"); String zkConnectionString = conf.get(ZKFailoverController.ZK_QUORUM_KEY); Preconditions.checkArgument(StringUtils.isNotBlank(zkConnectionString), "zookeeper quorum is unknown, invalid '%s' property value: %s", ZKFailoverController.ZK_QUORUM_KEY, zkConnectionString); int sessionTimeout = parameters.containsKey(PARAM_ZK_SESSION_TIMEOUT)? Integer.valueOf(parameters.get(PARAM_ZK_SESSION_TIMEOUT)) : DEFAULT_SESSION_TIMEOUT; final ZooKeeper zooKeeper = new ZooKeeper(zkConnectionString, sessionTimeout, (e) -> { }); if (zooKeeper.exists(DEFAULT_ROOT_NODE, false) == null) { log.info("initializing root node: " + DEFAULT_ROOT_NODE); try { zooKeeper.create(DEFAULT_ROOT_NODE, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); log.info("root node initialized"); } catch(NodeExistsException e) { log.warn("root node '" + DEFAULT_ROOT_NODE + "' was already created by different process"); } } final String nodePath = LockManagingProcessUtils.generatePath(parameters.get(PARAM_NODE_ID), DEFAULT_ROOT_NODE); switch(LockMode.valueOf(parameters.get(PARAM_LOCK_MODE))) { case obtain: { LockManagingProcessUtils.obtain(zooKeeper, nodePath); break; } case release: { LockManagingProcessUtils.release(zooKeeper, nodePath); break; } default: { throw new InvalidParameterException("unsupported lock mode: " + parameters.get(PARAM_LOCK_MODE)); } } }  @Override Map<String, PortType> getInputPorts(); @Override Map<String, PortType> getOutputPorts(); @Override void run(PortBindings portBindings, Configuration conf,
			Map<String, String> parameters); static final String DEFAULT_ROOT_NODE; static final String PARAM_ZK_SESSION_TIMEOUT; static final String PARAM_NODE_ID; static final String PARAM_LOCK_MODE; static final int DEFAULT_SESSION_TIMEOUT; static final Logger log; }