// Path: src/main/java/com/walmartlabs/mupd8/application/Mupd8DataPair.java
// public class Mupd8DataPair {
// 	public String _key;
// 	public byte   _value[];
// }
// 
// Path: src/main/java/com/walmartlabs/mupd8/application/Mupd8Source.java
// public interface Mupd8Source {
//   /**
//    * Returns true if there is a next item ready to be read by getNextDataPair().
//    * @return true if there is a next item ready to be read by getNextDataPair(), false otherwise
//    */
// 	boolean         hasNext();
// 
//   /**
//    * Returns a Mupd8DataPair parsed from the next item
//    * @return a Mupd8DataPair parsed from the next item
//    * @throws NoSuchElementException if there is no next item to read from
//    */
// 	Mupd8DataPair   getNextDataPair();
// }


import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.walmartlabs.mupd8.application.Mupd8DataPair;
import com.walmartlabs.mupd8.application.Mupd8Source;
            key = args.get(3);
        }
        initialize();
        objMapper=new ObjectMapper();
    }

    @Override
    public boolean hasNext() {
        boolean done = false;
        boolean ret = false;
        int retry = 0;
        while(!done){
            if(retry>MAX_RETRY){
                initialize();
                retry = 0;
            }
            try {
                ret = consumerIterator.hasNext();
                done = true;
            } catch (Exception e) {
                String mesg = "Consumer iterator hasNext throws exception";
                LOG.error(mesg, e);
                threadSleep(retry++,mesg);
                done = false;
            }
        }
        return ret;
    }

    @Override
// Path: src/main/java/com/walmartlabs/mupd8/application/Mupd8DataPair.java
// public class Mupd8DataPair {
// 	public String _key;
// 	public byte   _value[];
// }
// 
// Path: src/main/java/com/walmartlabs/mupd8/application/Mupd8Source.java
// public interface Mupd8Source {
//   /**
//    * Returns true if there is a next item ready to be read by getNextDataPair().
//    * @return true if there is a next item ready to be read by getNextDataPair(), false otherwise
//    */
// 	boolean         hasNext();
// 
//   /**
//    * Returns a Mupd8DataPair parsed from the next item
//    * @return a Mupd8DataPair parsed from the next item
//    * @throws NoSuchElementException if there is no next item to read from
//    */
// 	Mupd8DataPair   getNextDataPair();
// }


// Path: src/main/java/com/walmartlabs/mupd8/KafkaSource.java
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.walmartlabs.mupd8.application.Mupd8DataPair;
import com.walmartlabs.mupd8.application.Mupd8Source;

            key = args.get(3);
        }
        initialize();
        objMapper=new ObjectMapper();
    }

    @Override
    public boolean hasNext() {
        boolean done = false;
        boolean ret = false;
        int retry = 0;
        while(!done){
            if(retry>MAX_RETRY){
                initialize();
                retry = 0;
            }
            try {
                ret = consumerIterator.hasNext();
                done = true;
            } catch (Exception e) {
                String mesg = "Consumer iterator hasNext throws exception";
                LOG.error(mesg, e);
                threadSleep(retry++,mesg);
                done = false;
            }
        }
        return ret;
    }

    @Override
    public Mupd8DataPair getNextDataPair() {