public static Map<String, String> parseCmdOptions(String[] args)
	    throws ParseException {
	Options options = OptionParser.buildOptions();
	CommandLineParser parser = new GnuParser();
	CommandLine cmd = parser.parse(options, args);

	if (!cmd.hasOption(Kafka2Es.OPTION_ZOOKEEPER)
		|| !cmd.hasOption(Kafka2Es.OPTION_GROUP)
		|| !cmd.hasOption(Kafka2Es.OPTION_TOPIC)) {
	    HelpFormatter formatter = new HelpFormatter();
	    formatter.printHelp("kafka2es [OPTION]...", options);
	    return null;
	}
	Map<String, String> map = new HashMap<String, String>();
	String zkAddress = cmd.getOptionValue(Kafka2Es.OPTION_ZOOKEEPER);
	String groupId = cmd.getOptionValue(Kafka2Es.OPTION_GROUP);
	String topic = cmd.getOptionValue(Kafka2Es.OPTION_TOPIC);
	String esAddress = cmd.getOptionValue(Kafka2Es.OPTION_ES_ADDRESS);
	String esClusterName = cmd
		.getOptionValue(Kafka2Es.OPTION_ES_CLUSTER_NAME);
	String esIndexName = cmd.getOptionValue(Kafka2Es.OPTION_ES_INDEX_NAME);
	String esTypeName = cmd.getOptionValue(Kafka2Es.OPTION_ES_TYPE_NAME);
	String esRoutingKeyName = cmd
		.getOptionValue(Kafka2Es.OPTION_ES_ROUTINGKEY);
	String esBulkSize = cmd.getOptionValue(Kafka2Es.OPTION_ES_BULKSIZE);
	String esBulkMaxIntervalSec = cmd
		.getOptionValue(Kafka2Es.OPTION_ES_BULK_INTERVAL_SEC);

	/* not implemented */
	String fetchSize = cmd.getOptionValue("fetchSize");

	map.put(Kafka2Es.OPTION_ZOOKEEPER, zkAddress);
	map.put(Kafka2Es.OPTION_GROUP, groupId);
	map.put(Kafka2Es.OPTION_TOPIC, topic);
	map.put(Kafka2Es.OPTION_ES_ADDRESS, esAddress);
	map.put(Kafka2Es.OPTION_ES_CLUSTER_NAME, esClusterName);
	map.put(Kafka2Es.OPTION_ES_INDEX_NAME, esIndexName);
	map.put(Kafka2Es.OPTION_ES_TYPE_NAME, esTypeName);
	map.put(Kafka2Es.OPTION_ES_ROUTINGKEY, esRoutingKeyName);
	map.put(Kafka2Es.OPTION_ES_BULKSIZE, esBulkSize);
	map.put(Kafka2Es.OPTION_ES_BULK_INTERVAL_SEC, esBulkMaxIntervalSec);

	if (cmd.hasOption(Kafka2Es.OPTION_ENABLE_ES_INSERT)
		&& esAddress != null && esClusterName != null) {
	    map.put(Kafka2Es.OPTION_ENABLE_ES_INSERT, "true");
	    map.put(Kafka2Es.OPTION_ES_ADDRESS, esAddress);
	    map.put(Kafka2Es.OPTION_ES_CLUSTER_NAME, esClusterName);
	} else if (cmd.hasOption(Kafka2Es.OPTION_ENABLE_ES_INSERT)) {
	    HelpFormatter formatter = new HelpFormatter();
	    formatter
		    .printHelp(
			    "kafka2es --enable --elasticsearch=<urls> -- clustername=<name> ...",
			    options);
	    return null;
	} else {
	    map.put(Kafka2Es.OPTION_ENABLE_ES_INSERT, "false");
	}

	System.out.println("paring cli options");
	for (Entry<String, String> entry : map.entrySet()) {
	    System.out.println(entry.getKey() + "=" + entry.getValue());
	}

	return map;
    }