// Path: src/main/java/com/walmartlabs/mupd8/application/Mupd8DataPair.java
// public class Mupd8DataPair {
// 	public String _key;
// 	public byte   _value[];
// }


import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import junit.framework.TestCase;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.server.KafkaConfig;
import kafka.server.KafkaServerStartable;
import org.apache.curator.test.TestingServer;
import com.google.common.io.Files;
import com.walmartlabs.mupd8.application.Mupd8DataPair;
    private File logDir;
    private String topic = "mupd8-test-topic";
    private String key = "Kf1";
    private String val = "data1";
    private String mesg = "{ \"" + key + "\" : \"" + val + "\" }";

    @Override
    public void setUp() throws Exception {
        zkServer = new TestingServer();
        logDir = Files.createTempDir();
        kafkaServer = startKafkaServer(zkServer.getConnectString(), logDir.getAbsolutePath());
        produceJsonMessage();
        kafkaSource = getKafkaSource();
    }

    @Override
    public void tearDown() throws Exception {
        kafkaSource.closeSource();
        kafkaServer.shutdown();
        zkServer.stop();
        logDir.delete();
    }

    public void testHasNext()  {
        assertTrue("HasNext should return true when data", kafkaSource.hasNext());
        assertTrue("HasNext should not consume data",kafkaSource.hasNext());
    }

    public void testNext() {
        assertTrue("HasNext should return true when data", kafkaSource.hasNext());
// Path: src/main/java/com/walmartlabs/mupd8/application/Mupd8DataPair.java
// public class Mupd8DataPair {
// 	public String _key;
// 	public byte   _value[];
// }


// Path: src/test/java/com/walmartlabs/mupd8/KafkaSourceTest.java
import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import junit.framework.TestCase;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.server.KafkaConfig;
import kafka.server.KafkaServerStartable;
import org.apache.curator.test.TestingServer;
import com.google.common.io.Files;
import com.walmartlabs.mupd8.application.Mupd8DataPair;

    private File logDir;
    private String topic = "mupd8-test-topic";
    private String key = "Kf1";
    private String val = "data1";
    private String mesg = "{ \"" + key + "\" : \"" + val + "\" }";

    @Override
    public void setUp() throws Exception {
        zkServer = new TestingServer();
        logDir = Files.createTempDir();
        kafkaServer = startKafkaServer(zkServer.getConnectString(), logDir.getAbsolutePath());
        produceJsonMessage();
        kafkaSource = getKafkaSource();
    }

    @Override
    public void tearDown() throws Exception {
        kafkaSource.closeSource();
        kafkaServer.shutdown();
        zkServer.stop();
        logDir.delete();
    }

    public void testHasNext()  {
        assertTrue("HasNext should return true when data", kafkaSource.hasNext());
        assertTrue("HasNext should not consume data",kafkaSource.hasNext());
    }

    public void testNext() {
        assertTrue("HasNext should return true when data", kafkaSource.hasNext());
        Mupd8DataPair mupd8DataPair = kafkaSource.getNextDataPair();