KeyedMessage<K, V> keyedMessage = new KeyedMessage<>(topicName, key, message)

public class CuratorAppTest {
	private TestingServer server;
	
	@BeforeClass
	public void setUp() throws Exception {
		server = new TestingServer();
		server.start();
	}
	@AfterClass
	public void tearDown() throws IOException {
		server.stop();
	}
	
	@Test
	public void testSetAndGetData() {		
		CuratorApp app = new CuratorApp();
		String payload = System.currentTimeMillis() + "";
		String result = app.setAndGetData(server.getConnectString(), payload);
		assertEquals(result, payload);
	}
	
	@Test
	public void testWatch() throws Exception {		
		CuratorApp app = new CuratorApp();
		app.watch(server.getConnectString());
	}
}

Properties startupProperties = ...

QuorumPeerConfig quorumConfiguration = new QuorumPeerConfig();
try {
    quorumConfiguration.parseProperties(startupProperties);
} catch(Exception e) {
    throw new RuntimeException(e);
}

zooKeeperServer = new ZooKeeperServerMain();
final ServerConfig configuration = new ServerConfig();
configuration.readFrom(quorumConfiguration);

new Thread() {
    public void run() {
        try {
            zooKeeperServer.runFromConfig(configuration);
        } catch (IOException e) {
            log.error("ZooKeeper Failed", e);
        }
    }
}.start();

@BeforeClass
public void setup() {
	......
}
@AfterClass
public void teardown() {
	......
}
@Test
public void testReceive() {
	Properties consumerProps = TestUtils.createConsumerProperties(zkServer.connectString(), "group_1", "consumer_id", 1000);
	consumerProps.putAll(kafkaProps);
	ConsumerConfig consumerConfig = new ConsumerConfig(consumerProps);
	ConsumerConnector consumerConnector = Consumer.createJavaConsumerConnector(consumerConfig);

	Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
	topicCountMap.put(topic, new Integer(1));
	Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumerConnector.createMessageStreams(topicCountMap);
	List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);

	// now launch all the threads
	ExecutorService executor = Executors.newFixedThreadPool(2);

	// now create an object to consume the messages
	int threadNumber = 0;
	for (final KafkaStream<byte[], byte[]> stream : streams) {
		executor.submit(new ConsumerThread(stream, threadNumber));
		threadNumber++;
	}
	// setup producer
	Properties properties = TestUtils.getProducerConfig("localhost:" + port);
	properties.put("serializer.class", StringEncoder.class.getCanonicalName());

	ProducerConfig pConfig = new ProducerConfig(properties);
	Producer<Integer, String> producer = new Producer<>(pConfig);

	// send message
	for (int i = 0; i < 10; i++) {
		KeyedMessage<Integer, String> data = new KeyedMessage<>(topic, "test-message-" + i);

		List<KeyedMessage<Integer, String>> messages = new ArrayList<KeyedMessage<Integer, String>>();
		messages.add(data);
		// producer.send(scala.collection.JavaConversions.asScalaBuffer(messages));
		producer.send(scala.collection.JavaConversions.asScalaBuffer(messages));
	}

	try {
		Thread.sleep(20000);
	} catch (InterruptedException e) {
		e.printStackTrace();
	}

	executor.shutdownNow();
	producer.close();
}

public final class JavaSparkPi {
	public static void main(String[] args) throws Exception {
		double pi = calculatePi(args);		
		System.out.println("Pi is roughly " + pi);
		
	}

	public static double calculatePi(String[] args) {
		SparkConf sparkConf = new SparkConf().setAppName("JavaSparkPi");
		JavaSparkContext jsc = new JavaSparkContext(sparkConf);
		int slices = (args.length == 1) ? Integer.parseInt(args[0]) : 2;
		int n = 100000 * slices;
		List<Integer> l = new ArrayList<Integer>(n);
		for (int i = 0; i < n; i++) {
			l.add(i);
		}
		JavaRDD<Integer> dataSet = jsc.parallelize(l, slices);
		int count = dataSet.map(new Function<Integer, Integer>() {
			@Override
			public Integer call(Integer integer) {
				double x = Math.random() * 2 - 1;
				double y = Math.random() * 2 - 1;
				return (x * x + y * y < 1) ? 1 : 0;
			}
		}).reduce(new Function2<Integer, Integer, Integer>() {
			@Override
			public Integer call(Integer integer, Integer integer2) {
				return integer + integer2;
			}
		});
		double pi = 4.0 * count / n;
		jsc.stop();
		return pi;
	}
}

public class JavaSparkPiTest {

	@Test
	public void testPi() {
		Properties props = System.getProperties();		
		props.setProperty("spark.master", "local[4]");
		props.setProperty("spark.rdd.compress", "true");
		props.setProperty("spark.executor.memory", "1g");

		try {
			double pi = JavaSparkPi.calculatePi(new String[]{"1"});
			assertTrue((pi -3.14) < 1);
		} catch (Exception e) {
			fail(e.getMessage(),e);
		}
	}
}