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

		if (!cmd.hasOption(OPTION_ZOOKEEPER) || !cmd.hasOption(OPTION_GROUP)
				|| !cmd.hasOption(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(OPTION_ZOOKEEPER);
		String groupId = cmd.getOptionValue(OPTION_GROUP);
		String topic = cmd.getOptionValue(OPTION_TOPIC);

		// not implemented
		String fetchSize = cmd.getOptionValue(OPTION_GROUP);
		String bulkInsertSize;

		map.put(OPTION_ZOOKEEPER, zkAddress);
		map.put(OPTION_GROUP, groupId);
		map.put(OPTION_TOPIC, topic);

		return map;
	}