public static void main(String[] args) throws ParseException {
		//read options 
		Map<String, String> argMap = OptionParser.parseCmdOptions(args);
		if (argMap == null || argMap.isEmpty()) return;
		String zooKeeper = argMap.get(OPTION_ZOOKEEPER);
		String groupId = argMap.get(OPTION_GROUP);
		String topic = argMap.get(OPTION_TOPIC);
		boolean enableES = Boolean.parseBoolean(argMap
				.get(OPTION_ENABLE_ES_INSERT));
		String esAddress = argMap.get(OPTION_ES_ADDRESS);
		String clusterName = argMap.get(OPTION_ES_CLUSTER_NAME);
		String indexName = argMap.get(OPTION_ES_INDEX_NAME);
		String typeName = argMap.get(OPTION_ES_TYPE_NAME);
		String routingKey = argMap.get(OPTION_ES_ROUTINGKEY);
		
		int esBulkSize = 0;
		int esBulkMaxInterval = 0;
		try {
		    esBulkSize = Integer.valueOf(argMap.get(OPTION_ES_BULKSIZE));
		    esBulkMaxInterval = Integer.valueOf(argMap
				.get(OPTION_ES_BULK_INTERVAL_SEC));
		} catch (Exception e){
		   
		}
		System.out.println("!!");
		
		KafkaStreamJob job = null;

		//check requeired options
		
		//create job instance for data handling
		if (enableES) {
			try {
				job = new KafkaStreamJobImplES(esAddress, clusterName,
						indexName, typeName, routingKey, esBulkSize,
						esBulkMaxInterval);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		} else {
			job = new KafkaStreamJobImplStdout();
		}

		//create kafka consumer group & make consumer stream 
		KafkaGroupedConsumer consumerGroup = new KafkaGroupedConsumer(
				zooKeeper, groupId, 1);
		List<KafkaStream<byte[], byte[]>> streams = consumerGroup
				.getMessageStreams(topic);
		ExecutorService executor = Executors.newFixedThreadPool(1);

		try {
			executor.submit(new KafkaStreamHandler(streams.get(0), 0, job))
					.get();
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			executor.shutdown();
			consumerGroup.close();
		}
	}