helix-core/src/test/java/org/apache/helix/mock/participant/DummyProcess.java - helix - Git at Google

 package org.apache.helix.mock.participant;

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */

 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.CommandLineParser;
 import org.apache.commons.cli.GnuParser;
 import org.apache.commons.cli.HelpFormatter;
 import org.apache.commons.cli.Option;
 import org.apache.commons.cli.OptionBuilder;
 import org.apache.commons.cli.OptionGroup;
 import org.apache.commons.cli.Options;
 import org.apache.commons.cli.ParseException;
 import org.apache.helix.HelixManager;
 import org.apache.helix.HelixManagerFactory;
 import org.apache.helix.InstanceType;
 import org.apache.helix.NotificationContext;
 import org.apache.helix.model.Message;
 import org.apache.helix.participant.StateMachineEngine;
 import org.apache.helix.participant.statemachine.StateModel;
 import org.apache.helix.participant.statemachine.StateModelFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 public class DummyProcess {
   private static final Logger logger = LoggerFactory.getLogger(DummyProcess.class);
   public static final String zkServer = "zkSvr";
   public static final String cluster = "cluster";
   public static final String hostAddress = "host";
   public static final String hostPort = "port";
   public static final String relayCluster = "relayCluster";
   public static final String help = "help";
   public static final String transDelay = "transDelay";
   public static final String helixManagerType = "helixManagerType";
   // public static final String rootNamespace = "rootNamespace";

   private final String _zkConnectString;
   private final String _clusterName;
   private final String _instanceName;
   private DummyMasterSlaveStateModelFactory stateModelFactory;
   // private StateMachineEngine genericStateMachineHandler;

   private int _transDelayInMs = 0;
   private final String _clusterMangerType;

   public DummyProcess(String zkConnectString, String clusterName, String instanceName,
       String clusterMangerType, int delay) {
     _zkConnectString = zkConnectString;
     _clusterName = clusterName;
     _instanceName = instanceName;
     _clusterMangerType = clusterMangerType;
     _transDelayInMs = delay > 0 ? delay : 0;
   }

   static void sleep(long transDelay) {
     try {
       if (transDelay > 0) {
         Thread.sleep(transDelay);
       }
     } catch (InterruptedException e) {
       // TODO Auto-generated catch block
       e.printStackTrace();
     }
   }

   public HelixManager start() throws Exception {
     HelixManager manager = null;
     // zk cluster manager
     if (_clusterMangerType.equalsIgnoreCase("zk")) {
       manager =
           HelixManagerFactory.getZKHelixManager(_clusterName, _instanceName,
               InstanceType.PARTICIPANT, _zkConnectString);
     } else {
       throw new IllegalArgumentException("Unsupported cluster manager type:" + _clusterMangerType);
     }

     stateModelFactory = new DummyMasterSlaveStateModelFactory(_transDelayInMs);
     DummyLeaderStandbyStateModelFactory stateModelFactory1 =
         new DummyLeaderStandbyStateModelFactory(_transDelayInMs);
     DummyOnlineOfflineStateModelFactory stateModelFactory2 =
         new DummyOnlineOfflineStateModelFactory(_transDelayInMs);
     // genericStateMachineHandler = new StateMachineEngine();
     StateMachineEngine stateMach = manager.getStateMachineEngine();
     stateMach.registerStateModelFactory("MasterSlave", stateModelFactory);
     stateMach.registerStateModelFactory("LeaderStandby", stateModelFactory1);
     stateMach.registerStateModelFactory("OnlineOffline", stateModelFactory2);

     manager.connect();
     // manager.getMessagingService().registerMessageHandlerFactory(MessageType.STATE_TRANSITION.name(),
     // genericStateMachineHandler);
     return manager;
   }

   public static class DummyMasterSlaveStateModelFactory extends StateModelFactory<DummyMasterSlaveStateModel> {
     int _delay;

     public DummyMasterSlaveStateModelFactory(int delay) {
       _delay = delay;
     }

     @Override
     public DummyMasterSlaveStateModel createNewStateModel(String resourceName, String stateUnitKey) {
       DummyMasterSlaveStateModel model = new DummyMasterSlaveStateModel();
       model.setDelay(_delay);
       return model;
     }
   }

   public static class DummyLeaderStandbyStateModelFactory extends
       StateModelFactory<DummyLeaderStandbyStateModel> {
     int _delay;

     public DummyLeaderStandbyStateModelFactory(int delay) {
       _delay = delay;
     }

     @Override
     public DummyLeaderStandbyStateModel createNewStateModel(String resourceName, String stateUnitKey) {
       DummyLeaderStandbyStateModel model = new DummyLeaderStandbyStateModel();
       model.setDelay(_delay);
       return model;
     }
   }

   public static class DummyOnlineOfflineStateModelFactory extends
       StateModelFactory<DummyOnlineOfflineStateModel> {
     int _delay;

     public DummyOnlineOfflineStateModelFactory(int delay) {
       _delay = delay;
     }

     @Override
     public DummyOnlineOfflineStateModel createNewStateModel(String resourceName, String stateUnitKey) {
       DummyOnlineOfflineStateModel model = new DummyOnlineOfflineStateModel();
       model.setDelay(_delay);
       return model;
     }
   }

   public static class DummyMasterSlaveStateModel extends StateModel {
     int _transDelay = 0;

     public void setDelay(int delay) {
       _transDelay = delay > 0 ? delay : 0;
     }

     public void onBecomeSlaveFromOffline(Message message, NotificationContext context) {
       String db = message.getPartitionName();
       String instanceName = context.getManager().getInstanceName();
       DummyProcess.sleep(_transDelay);

       logger.info("DummyStateModel.onBecomeSlaveFromOffline(), instance:" + instanceName + ", db:"
           + db);
     }

     public void onBecomeSlaveFromMaster(Message message, NotificationContext context) {
       DummyProcess.sleep(_transDelay);

       logger.info("DummyStateModel.onBecomeSlaveFromMaster()");

     }

     public void onBecomeMasterFromSlave(Message message, NotificationContext context) {
       DummyProcess.sleep(_transDelay);

       logger.info("DummyStateModel.onBecomeMasterFromSlave()");

     }

     public void onBecomeOfflineFromSlave(Message message, NotificationContext context) {
       DummyProcess.sleep(_transDelay);

       logger.info("DummyStateModel.onBecomeOfflineFromSlave()");

     }

     public void onBecomeDroppedFromOffline(Message message, NotificationContext context) {
       DummyProcess.sleep(_transDelay);

       logger.info("DummyStateModel.onBecomeDroppedFromOffline()");

     }
   }

   public static class DummyOnlineOfflineStateModel extends StateModel {
     int _transDelay = 0;

     public void setDelay(int delay) {
       _transDelay = delay > 0 ? delay : 0;
     }

     public void onBecomeOnlineFromOffline(Message message, NotificationContext context) {
       String db = message.getPartitionName();
       String instanceName = context.getManager().getInstanceName();
       DummyProcess.sleep(_transDelay);

       logger.info("DummyStateModel.onBecomeOnlineFromOffline(), instance:" + instanceName + ", db:"
           + db);
     }

     public void onBecomeOfflineFromOnline(Message message, NotificationContext context) {
       DummyProcess.sleep(_transDelay);

       logger.info("DummyStateModel.onBecomeOfflineFromOnline()");

     }

     public void onBecomeDroppedFromOffline(Message message, NotificationContext context) {
       DummyProcess.sleep(_transDelay);

       logger.info("DummyStateModel.onBecomeDroppedFromOffline()");

     }
   }

   public static class DummyLeaderStandbyStateModel extends StateModel {
     int _transDelay = 0;

     public void setDelay(int delay) {
       _transDelay = delay > 0 ? delay : 0;
     }

     public void onBecomeLeaderFromStandby(Message message, NotificationContext context) {
       String db = message.getPartitionName();
       String instanceName = context.getManager().getInstanceName();
       DummyProcess.sleep(_transDelay);
       logger.info("DummyLeaderStandbyStateModel.onBecomeLeaderFromStandby(), instance:"
           + instanceName + ", db:" + db);
     }

     public void onBecomeStandbyFromLeader(Message message, NotificationContext context) {
       DummyProcess.sleep(_transDelay);

       logger.info("DummyLeaderStandbyStateModel.onBecomeStandbyFromLeader()");

     }

     public void onBecomeDroppedFromOffline(Message message, NotificationContext context) {
       DummyProcess.sleep(_transDelay);

       logger.info("DummyLeaderStandbyStateModel.onBecomeDroppedFromOffline()");

     }

     public void onBecomeStandbyFromOffline(Message message, NotificationContext context) {
       DummyProcess.sleep(_transDelay);

       logger.info("DummyLeaderStandbyStateModel.onBecomeStandbyFromOffline()");

     }

     public void onBecomeOfflineFromStandby(Message message, NotificationContext context) {
       DummyProcess.sleep(_transDelay);

       logger.info("DummyLeaderStandbyStateModel.onBecomeOfflineFromStandby()");

     }
   }

   // TODO hack OptionBuilder is not thread safe
   @SuppressWarnings("static-access")
   synchronized private static Options constructCommandLineOptions() {
     Option helpOption =
         OptionBuilder.withLongOpt(help).withDescription("Prints command-line options info")
             .create();

     Option clusterOption =
         OptionBuilder.withLongOpt(cluster).withDescription("Provide cluster name").create();
     clusterOption.setArgs(1);
     clusterOption.setRequired(true);
     clusterOption.setArgName("Cluster name (Required)");

     Option hostOption =
         OptionBuilder.withLongOpt(hostAddress).withDescription("Provide host name").create();
     hostOption.setArgs(1);
     hostOption.setRequired(true);
     hostOption.setArgName("Host name (Required)");

     Option portOption =
         OptionBuilder.withLongOpt(hostPort).withDescription("Provide host port").create();
     portOption.setArgs(1);
     portOption.setRequired(true);
     portOption.setArgName("Host port (Required)");

     Option cmTypeOption =
         OptionBuilder
             .withLongOpt(helixManagerType)
             .withDescription(
                 "Provide cluster manager type (e.g. 'zk', 'static-file', or 'dynamic-file'")
             .create();
     cmTypeOption.setArgs(1);
     cmTypeOption.setRequired(true);
     cmTypeOption
         .setArgName("Clsuter manager type (e.g. 'zk', 'static-file', or 'dynamic-file') (Required)");

     Option zkServerOption =
         OptionBuilder.withLongOpt(zkServer).withDescription("Provide zookeeper address").create();
     zkServerOption.setArgs(1);
     zkServerOption.setRequired(true);
     zkServerOption.setArgName("ZookeeperServerAddress(Required for zk-based cluster manager)");

     // Option rootNsOption = OptionBuilder.withLongOpt(rootNamespace)
     // .withDescription("Provide root namespace for dynamic-file based cluster manager").create();
     // rootNsOption.setArgs(1);
     // rootNsOption.setRequired(true);
     // rootNsOption.setArgName("Root namespace (Required for dynamic-file based cluster manager)");

     Option transDelayOption =
         OptionBuilder.withLongOpt(transDelay).withDescription("Provide state trans delay").create();
     transDelayOption.setArgs(1);
     transDelayOption.setRequired(false);
     transDelayOption.setArgName("Delay time in state transition, in MS");

     OptionGroup optionGroup = new OptionGroup();
     optionGroup.addOption(zkServerOption);

     Options options = new Options();
     options.addOption(helpOption);
     options.addOption(clusterOption);
     options.addOption(hostOption);
     options.addOption(portOption);
     options.addOption(transDelayOption);
     options.addOption(cmTypeOption);

     options.addOptionGroup(optionGroup);

     return options;
   }

   public static void printUsage(Options cliOptions) {
     HelpFormatter helpFormatter = new HelpFormatter();
     helpFormatter.printHelp("java " + DummyProcess.class.getName(), cliOptions);
   }

   public static CommandLine processCommandLineArgs(String[] cliArgs) throws Exception {
     CommandLineParser cliParser = new GnuParser();
     Options cliOptions = constructCommandLineOptions();
     // CommandLine cmd = null;

     try {
       return cliParser.parse(cliOptions, cliArgs);
     } catch (ParseException pe) {
       System.err.println("CommandLineClient: failed to parse command-line options: "
           + pe.toString());
       printUsage(cliOptions);
       System.exit(1);
     }
     return null;
   }

   public static void main(String[] args) throws Exception {
     String cmType = "zk";
     String zkConnectString = "localhost:2181";
     String clusterName = "testCluster";
     String instanceName = "localhost_8900";
     String cvFileStr = null;
     // String rootNs = null;
     int delay = 0;

     if (args.length > 0) {
       CommandLine cmd = processCommandLineArgs(args);
       zkConnectString = cmd.getOptionValue(zkServer);
       clusterName = cmd.getOptionValue(cluster);

       String host = cmd.getOptionValue(hostAddress);
       String portString = cmd.getOptionValue(hostPort);
       int port = Integer.parseInt(portString);
       instanceName = host + "_" + port;
       cmType = cmd.getOptionValue(helixManagerType);
       if (cmd.hasOption(transDelay)) {
         try {
           delay = Integer.parseInt(cmd.getOptionValue(transDelay));
           if (delay < 0) {
             throw new Exception("delay must be positive");
           }
         } catch (Exception e) {
           e.printStackTrace();
           delay = 0;
         }
       }
     }
     // Espresso_driver.py will consume this
     logger.info("Dummy process started, instanceName:" + instanceName);

     DummyProcess process =
         new DummyProcess(zkConnectString, clusterName, instanceName, cmType, delay);
     HelixManager manager = process.start();

     try {
       Thread.currentThread().join();
     } catch (InterruptedException e) {
       // ClusterManagerFactory.disconnectManagers(instanceName);
       logger.info("participant:" + instanceName + ", " + Thread.currentThread().getName()
           + " interrupted");
       // if (manager != null)
       // {
       // manager.disconnect();
       // }
     } finally {
       if (manager != null) {
         manager.disconnect();
       }
     }

   }
 }
	package org.apache.helix.mock.participant;

	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/

	import org.apache.commons.cli.CommandLine;
	import org.apache.commons.cli.CommandLineParser;
	import org.apache.commons.cli.GnuParser;
	import org.apache.commons.cli.HelpFormatter;
	import org.apache.commons.cli.Option;
	import org.apache.commons.cli.OptionBuilder;
	import org.apache.commons.cli.OptionGroup;
	import org.apache.commons.cli.Options;
	import org.apache.commons.cli.ParseException;
	import org.apache.helix.HelixManager;
	import org.apache.helix.HelixManagerFactory;
	import org.apache.helix.InstanceType;
	import org.apache.helix.NotificationContext;
	import org.apache.helix.model.Message;
	import org.apache.helix.participant.StateMachineEngine;
	import org.apache.helix.participant.statemachine.StateModel;
	import org.apache.helix.participant.statemachine.StateModelFactory;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	public class DummyProcess {
	private static final Logger logger = LoggerFactory.getLogger(DummyProcess.class);
	public static final String zkServer = "zkSvr";
	public static final String cluster = "cluster";
	public static final String hostAddress = "host";
	public static final String hostPort = "port";
	public static final String relayCluster = "relayCluster";
	public static final String help = "help";
	public static final String transDelay = "transDelay";
	public static final String helixManagerType = "helixManagerType";
	// public static final String rootNamespace = "rootNamespace";

	private final String _zkConnectString;
	private final String _clusterName;
	private final String _instanceName;
	private DummyMasterSlaveStateModelFactory stateModelFactory;
	// private StateMachineEngine genericStateMachineHandler;

	private int _transDelayInMs = 0;
	private final String _clusterMangerType;

	public DummyProcess(String zkConnectString, String clusterName, String instanceName,
	String clusterMangerType, int delay) {
	_zkConnectString = zkConnectString;
	_clusterName = clusterName;
	_instanceName = instanceName;
	_clusterMangerType = clusterMangerType;
	_transDelayInMs = delay > 0 ? delay : 0;
	}

	static void sleep(long transDelay) {
	try {
	if (transDelay > 0) {
	Thread.sleep(transDelay);
	}
	} catch (InterruptedException e) {
	// TODO Auto-generated catch block
	e.printStackTrace();
	}
	}

	public HelixManager start() throws Exception {
	HelixManager manager = null;
	// zk cluster manager
	if (_clusterMangerType.equalsIgnoreCase("zk")) {
	manager =
	HelixManagerFactory.getZKHelixManager(_clusterName, _instanceName,
	InstanceType.PARTICIPANT, _zkConnectString);
	} else {
	throw new IllegalArgumentException("Unsupported cluster manager type:" + _clusterMangerType);
	}

	stateModelFactory = new DummyMasterSlaveStateModelFactory(_transDelayInMs);
	DummyLeaderStandbyStateModelFactory stateModelFactory1 =
	new DummyLeaderStandbyStateModelFactory(_transDelayInMs);
	DummyOnlineOfflineStateModelFactory stateModelFactory2 =
	new DummyOnlineOfflineStateModelFactory(_transDelayInMs);
	// genericStateMachineHandler = new StateMachineEngine();
	StateMachineEngine stateMach = manager.getStateMachineEngine();
	stateMach.registerStateModelFactory("MasterSlave", stateModelFactory);
	stateMach.registerStateModelFactory("LeaderStandby", stateModelFactory1);
	stateMach.registerStateModelFactory("OnlineOffline", stateModelFactory2);

	manager.connect();
	// manager.getMessagingService().registerMessageHandlerFactory(MessageType.STATE_TRANSITION.name(),
	// genericStateMachineHandler);
	return manager;
	}

	public static class DummyMasterSlaveStateModelFactory extends StateModelFactory<DummyMasterSlaveStateModel> {
	int _delay;

	public DummyMasterSlaveStateModelFactory(int delay) {
	_delay = delay;
	}

	@Override
	public DummyMasterSlaveStateModel createNewStateModel(String resourceName, String stateUnitKey) {
	DummyMasterSlaveStateModel model = new DummyMasterSlaveStateModel();
	model.setDelay(_delay);
	return model;
	}
	}

	public static class DummyLeaderStandbyStateModelFactory extends
	StateModelFactory<DummyLeaderStandbyStateModel> {
	int _delay;

	public DummyLeaderStandbyStateModelFactory(int delay) {
	_delay = delay;
	}

	@Override
	public DummyLeaderStandbyStateModel createNewStateModel(String resourceName, String stateUnitKey) {
	DummyLeaderStandbyStateModel model = new DummyLeaderStandbyStateModel();
	model.setDelay(_delay);
	return model;
	}
	}

	public static class DummyOnlineOfflineStateModelFactory extends
	StateModelFactory<DummyOnlineOfflineStateModel> {
	int _delay;

	public DummyOnlineOfflineStateModelFactory(int delay) {
	_delay = delay;
	}

	@Override
	public DummyOnlineOfflineStateModel createNewStateModel(String resourceName, String stateUnitKey) {
	DummyOnlineOfflineStateModel model = new DummyOnlineOfflineStateModel();
	model.setDelay(_delay);
	return model;
	}
	}

	public static class DummyMasterSlaveStateModel extends StateModel {
	int _transDelay = 0;

	public void setDelay(int delay) {
	_transDelay = delay > 0 ? delay : 0;
	}

	public void onBecomeSlaveFromOffline(Message message, NotificationContext context) {
	String db = message.getPartitionName();
	String instanceName = context.getManager().getInstanceName();
	DummyProcess.sleep(_transDelay);

	logger.info("DummyStateModel.onBecomeSlaveFromOffline(), instance:" + instanceName + ", db:"
	+ db);
	}

	public void onBecomeSlaveFromMaster(Message message, NotificationContext context) {
	DummyProcess.sleep(_transDelay);

	logger.info("DummyStateModel.onBecomeSlaveFromMaster()");

	}

	public void onBecomeMasterFromSlave(Message message, NotificationContext context) {
	DummyProcess.sleep(_transDelay);

	logger.info("DummyStateModel.onBecomeMasterFromSlave()");

	}

	public void onBecomeOfflineFromSlave(Message message, NotificationContext context) {
	DummyProcess.sleep(_transDelay);

	logger.info("DummyStateModel.onBecomeOfflineFromSlave()");

	}

	public void onBecomeDroppedFromOffline(Message message, NotificationContext context) {
	DummyProcess.sleep(_transDelay);

	logger.info("DummyStateModel.onBecomeDroppedFromOffline()");

	}
	}

	public static class DummyOnlineOfflineStateModel extends StateModel {
	int _transDelay = 0;

	public void setDelay(int delay) {
	_transDelay = delay > 0 ? delay : 0;
	}

	public void onBecomeOnlineFromOffline(Message message, NotificationContext context) {
	String db = message.getPartitionName();
	String instanceName = context.getManager().getInstanceName();
	DummyProcess.sleep(_transDelay);

	logger.info("DummyStateModel.onBecomeOnlineFromOffline(), instance:" + instanceName + ", db:"
	+ db);
	}

	public void onBecomeOfflineFromOnline(Message message, NotificationContext context) {
	DummyProcess.sleep(_transDelay);

	logger.info("DummyStateModel.onBecomeOfflineFromOnline()");

	}

	public void onBecomeDroppedFromOffline(Message message, NotificationContext context) {
	DummyProcess.sleep(_transDelay);

	logger.info("DummyStateModel.onBecomeDroppedFromOffline()");

	}
	}

	public static class DummyLeaderStandbyStateModel extends StateModel {
	int _transDelay = 0;

	public void setDelay(int delay) {
	_transDelay = delay > 0 ? delay : 0;
	}

	public void onBecomeLeaderFromStandby(Message message, NotificationContext context) {
	String db = message.getPartitionName();
	String instanceName = context.getManager().getInstanceName();
	DummyProcess.sleep(_transDelay);
	logger.info("DummyLeaderStandbyStateModel.onBecomeLeaderFromStandby(), instance:"
	+ instanceName + ", db:" + db);
	}

	public void onBecomeStandbyFromLeader(Message message, NotificationContext context) {
	DummyProcess.sleep(_transDelay);

	logger.info("DummyLeaderStandbyStateModel.onBecomeStandbyFromLeader()");

	}

	public void onBecomeDroppedFromOffline(Message message, NotificationContext context) {
	DummyProcess.sleep(_transDelay);

	logger.info("DummyLeaderStandbyStateModel.onBecomeDroppedFromOffline()");

	}

	public void onBecomeStandbyFromOffline(Message message, NotificationContext context) {
	DummyProcess.sleep(_transDelay);

	logger.info("DummyLeaderStandbyStateModel.onBecomeStandbyFromOffline()");

	}

	public void onBecomeOfflineFromStandby(Message message, NotificationContext context) {
	DummyProcess.sleep(_transDelay);

	logger.info("DummyLeaderStandbyStateModel.onBecomeOfflineFromStandby()");

	}
	}

	// TODO hack OptionBuilder is not thread safe
	@SuppressWarnings("static-access")
	synchronized private static Options constructCommandLineOptions() {
	Option helpOption =
	OptionBuilder.withLongOpt(help).withDescription("Prints command-line options info")
	.create();

	Option clusterOption =
	OptionBuilder.withLongOpt(cluster).withDescription("Provide cluster name").create();
	clusterOption.setArgs(1);
	clusterOption.setRequired(true);
	clusterOption.setArgName("Cluster name (Required)");

	Option hostOption =
	OptionBuilder.withLongOpt(hostAddress).withDescription("Provide host name").create();
	hostOption.setArgs(1);
	hostOption.setRequired(true);
	hostOption.setArgName("Host name (Required)");

	Option portOption =
	OptionBuilder.withLongOpt(hostPort).withDescription("Provide host port").create();
	portOption.setArgs(1);
	portOption.setRequired(true);
	portOption.setArgName("Host port (Required)");

	Option cmTypeOption =
	OptionBuilder
	.withLongOpt(helixManagerType)
	.withDescription(
	"Provide cluster manager type (e.g. 'zk', 'static-file', or 'dynamic-file'")
	.create();
	cmTypeOption.setArgs(1);
	cmTypeOption.setRequired(true);
	cmTypeOption
	.setArgName("Clsuter manager type (e.g. 'zk', 'static-file', or 'dynamic-file') (Required)");

	Option zkServerOption =
	OptionBuilder.withLongOpt(zkServer).withDescription("Provide zookeeper address").create();
	zkServerOption.setArgs(1);
	zkServerOption.setRequired(true);
	zkServerOption.setArgName("ZookeeperServerAddress(Required for zk-based cluster manager)");

	// Option rootNsOption = OptionBuilder.withLongOpt(rootNamespace)
	// .withDescription("Provide root namespace for dynamic-file based cluster manager").create();
	// rootNsOption.setArgs(1);
	// rootNsOption.setRequired(true);
	// rootNsOption.setArgName("Root namespace (Required for dynamic-file based cluster manager)");

	Option transDelayOption =
	OptionBuilder.withLongOpt(transDelay).withDescription("Provide state trans delay").create();
	transDelayOption.setArgs(1);
	transDelayOption.setRequired(false);
	transDelayOption.setArgName("Delay time in state transition, in MS");

	OptionGroup optionGroup = new OptionGroup();
	optionGroup.addOption(zkServerOption);

	Options options = new Options();
	options.addOption(helpOption);
	options.addOption(clusterOption);
	options.addOption(hostOption);
	options.addOption(portOption);
	options.addOption(transDelayOption);
	options.addOption(cmTypeOption);

	options.addOptionGroup(optionGroup);

	return options;
	}

	public static void printUsage(Options cliOptions) {
	HelpFormatter helpFormatter = new HelpFormatter();
	helpFormatter.printHelp("java " + DummyProcess.class.getName(), cliOptions);
	}

	public static CommandLine processCommandLineArgs(String[] cliArgs) throws Exception {
	CommandLineParser cliParser = new GnuParser();
	Options cliOptions = constructCommandLineOptions();
	// CommandLine cmd = null;

	try {
	return cliParser.parse(cliOptions, cliArgs);
	} catch (ParseException pe) {
	System.err.println("CommandLineClient: failed to parse command-line options: "
	+ pe.toString());
	printUsage(cliOptions);
	System.exit(1);
	}
	return null;
	}

	public static void main(String[] args) throws Exception {
	String cmType = "zk";
	String zkConnectString = "localhost:2181";
	String clusterName = "testCluster";
	String instanceName = "localhost_8900";
	String cvFileStr = null;
	// String rootNs = null;
	int delay = 0;

	if (args.length > 0) {
	CommandLine cmd = processCommandLineArgs(args);
	zkConnectString = cmd.getOptionValue(zkServer);
	clusterName = cmd.getOptionValue(cluster);

	String host = cmd.getOptionValue(hostAddress);
	String portString = cmd.getOptionValue(hostPort);
	int port = Integer.parseInt(portString);
	instanceName = host + "_" + port;
	cmType = cmd.getOptionValue(helixManagerType);
	if (cmd.hasOption(transDelay)) {
	try {
	delay = Integer.parseInt(cmd.getOptionValue(transDelay));
	if (delay < 0) {
	throw new Exception("delay must be positive");
	}
	} catch (Exception e) {
	e.printStackTrace();
	delay = 0;
	}
	}
	}
	// Espresso_driver.py will consume this
	logger.info("Dummy process started, instanceName:" + instanceName);

	DummyProcess process =
	new DummyProcess(zkConnectString, clusterName, instanceName, cmType, delay);
	HelixManager manager = process.start();

	try {
	Thread.currentThread().join();
	} catch (InterruptedException e) {
	// ClusterManagerFactory.disconnectManagers(instanceName);
	logger.info("participant:" + instanceName + ", " + Thread.currentThread().getName()
	+ " interrupted");
	// if (manager != null)
	// {
	// manager.disconnect();
	// }
	} finally {
	if (manager != null) {
	manager.disconnect();
	}
	}

	}
	}