You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by ds...@apache.org on 2015/07/06 23:27:16 UTC
[4/5] incubator-geode git commit: GEODE-9: Imported
gemfire-spark-connector from geode-1.0.0-SNAPSHOT-2.src.tar
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/85d44f04/gemfire-spark-connector/gemfire-spark-connector/src/it/java/ittest/io/pivotal/gemfire/spark/connector/JavaApiIntegrationTest.java
----------------------------------------------------------------------
diff --git a/gemfire-spark-connector/gemfire-spark-connector/src/it/java/ittest/io/pivotal/gemfire/spark/connector/JavaApiIntegrationTest.java b/gemfire-spark-connector/gemfire-spark-connector/src/it/java/ittest/io/pivotal/gemfire/spark/connector/JavaApiIntegrationTest.java
new file mode 100644
index 0000000..8357d8f
--- /dev/null
+++ b/gemfire-spark-connector/gemfire-spark-connector/src/it/java/ittest/io/pivotal/gemfire/spark/connector/JavaApiIntegrationTest.java
@@ -0,0 +1,368 @@
+package ittest.io.pivotal.gemfire.spark.connector;
+
+import com.gemstone.gemfire.cache.Region;
+import io.pivotal.gemfire.spark.connector.GemFireConnection;
+import io.pivotal.gemfire.spark.connector.GemFireConnectionConf;
+import io.pivotal.gemfire.spark.connector.GemFireConnectionConf$;
+import io.pivotal.gemfire.spark.connector.internal.DefaultGemFireConnectionManager$;
+import io.pivotal.gemfire.spark.connector.javaapi.GemFireJavaRegionRDD;
+import ittest.io.pivotal.gemfire.spark.connector.testkit.GemFireCluster$;
+import ittest.io.pivotal.gemfire.spark.connector.testkit.IOUtils;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.api.java.function.Function;
+import org.apache.spark.api.java.function.PairFunction;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.scalatest.junit.JUnitSuite;
+import io.pivotal.gemfire.spark.connector.package$;
+import scala.Tuple2;
+import scala.Option;
+import scala.Some;
+
+import java.util.*;
+
+import static io.pivotal.gemfire.spark.connector.javaapi.GemFireJavaUtil.javaFunctions;
+import static org.junit.Assert.*;
+
+public class JavaApiIntegrationTest extends JUnitSuite {
+
+ static JavaSparkContext jsc = null;
+ static GemFireConnectionConf connConf = null;
+
+ static int numServers = 2;
+ static int numObjects = 1000;
+ static String regionPath = "pr_str_int_region";
+
+ @BeforeClass
+ public static void setUpBeforeClass() throws Exception {
+ // start gemfire cluster, and spark context
+ Properties settings = new Properties();
+ settings.setProperty("cache-xml-file", "src/it/resources/test-retrieve-regions.xml");
+ settings.setProperty("num-of-servers", Integer.toString(numServers));
+ int locatorPort = GemFireCluster$.MODULE$.start(settings);
+
+ // start spark context in local mode
+ Properties props = new Properties();
+ props.put("log4j.logger.org.apache.spark", "INFO");
+ props.put("log4j.logger.io.pivotal.gemfire.spark.connector","DEBUG");
+ IOUtils.configTestLog4j("ERROR", props);
+ SparkConf conf = new SparkConf()
+ .setAppName("RetrieveRegionIntegrationTest")
+ .setMaster("local[2]")
+ .set(package$.MODULE$.GemFireLocatorPropKey(), "localhost:"+ locatorPort);
+ // sc = new SparkContext(conf);
+ jsc = new JavaSparkContext(conf);
+ connConf = GemFireConnectionConf.apply(jsc.getConf());
+ }
+
+ @AfterClass
+ public static void tearDownAfterClass() throws Exception {
+ // stop connection, spark context, and gemfire cluster
+ DefaultGemFireConnectionManager$.MODULE$.closeConnection(GemFireConnectionConf$.MODULE$.apply(jsc.getConf()));
+ jsc.stop();
+ GemFireCluster$.MODULE$.stop();
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // utility methods
+ // --------------------------------------------------------------------------------------------
+
+ private <K,V> void matchMapAndPairList(Map<K,V> map, List<Tuple2<K,V>> list) {
+ assertTrue("size mismatch \nmap: " + map.toString() + "\nlist: " + list.toString(), map.size() == list.size());
+ for (Tuple2<K, V> p : list) {
+ assertTrue("value mismatch: k=" + p._1() + " v1=" + p._2() + " v2=" + map.get(p._1()),
+ p._2().equals(map.get(p._1())));
+ }
+ }
+
+ private Region<String, Integer> prepareStrIntRegion(String regionPath, int start, int stop) {
+ HashMap<String, Integer> entriesMap = new HashMap<>();
+ for (int i = start; i < stop; i ++) {
+ entriesMap.put("k_" + i, i);
+ }
+
+ GemFireConnection conn = connConf.getConnection();
+ Region<String, Integer> region = conn.getRegionProxy(regionPath);
+ region.removeAll(region.keySetOnServer());
+ region.putAll(entriesMap);
+ return region;
+ }
+
+ private JavaPairRDD<String, Integer> prepareStrIntJavaPairRDD(int start, int stop) {
+ List<Tuple2<String, Integer>> data = new ArrayList<>();
+ for (int i = start; i < stop; i ++) {
+ data.add(new Tuple2<>("k_" + i, i));
+ }
+ return jsc.parallelizePairs(data);
+ }
+
+ private JavaPairRDD<Integer, Integer> prepareIntIntJavaPairRDD(int start, int stop) {
+ List<Tuple2<Integer, Integer>> data = new ArrayList<>();
+ for (int i = start; i < stop; i ++) {
+ data.add(new Tuple2<>(i, i * 2));
+ }
+ return jsc.parallelizePairs(data);
+ }
+
+ private JavaRDD<Integer> prepareIntJavaRDD(int start, int stop) {
+ List<Integer> data = new ArrayList<>();
+ for (int i = start; i < stop; i ++) {
+ data.add(i);
+ }
+ return jsc.parallelize(data);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // JavaRDD.saveToGemfire
+ // --------------------------------------------------------------------------------------------
+
+ static class IntToStrIntPairFunction implements PairFunction<Integer, String, Integer> {
+ @Override public Tuple2<String, Integer> call(Integer x) throws Exception {
+ return new Tuple2<>("k_" + x, x);
+ }
+ }
+
+ @Test
+ public void testRDDSaveToGemfireWithDefaultConnConf() throws Exception {
+ verifyRDDSaveToGemfire(true);
+ }
+
+ @Test
+ public void testRDDSaveToGemfireWithConnConf() throws Exception {
+ verifyRDDSaveToGemfire(false);
+ }
+
+ public void verifyRDDSaveToGemfire(boolean useDefaultConnConf) throws Exception {
+ Region<String, Integer> region = prepareStrIntRegion(regionPath, 0, 0); // remove all entries
+ JavaRDD<Integer> rdd1 = prepareIntJavaRDD(0, numObjects);
+
+ PairFunction<Integer, String, Integer> func = new IntToStrIntPairFunction();
+ if (useDefaultConnConf)
+ javaFunctions(rdd1).saveToGemfire(regionPath, func);
+ else
+ javaFunctions(rdd1).saveToGemfire(regionPath, func, connConf);
+
+ Set<String> keys = region.keySetOnServer();
+ Map<String, Integer> map = region.getAll(keys);
+
+ List<Tuple2<String, Integer>> expectedList = new ArrayList<>();
+
+ for (int i = 0; i < numObjects; i ++) {
+ expectedList.add(new Tuple2<>("k_" + i, i));
+ }
+ matchMapAndPairList(map, expectedList);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // JavaPairRDD.saveToGemfire
+ // --------------------------------------------------------------------------------------------
+
+ @Test
+ public void testPairRDDSaveToGemfireWithDefaultConnConf() throws Exception {
+ verifyPairRDDSaveToGemfire(true);
+ }
+
+ @Test
+ public void testPairRDDSaveToGemfireWithConnConf() throws Exception {
+ verifyPairRDDSaveToGemfire(false);
+ }
+
+ public void verifyPairRDDSaveToGemfire(boolean useDefaultConnConf) throws Exception {
+ Region<String, Integer> region = prepareStrIntRegion(regionPath, 0, 0); // remove all entries
+ JavaPairRDD<String, Integer> rdd1 = prepareStrIntJavaPairRDD(0, numObjects);
+
+ if (useDefaultConnConf)
+ javaFunctions(rdd1).saveToGemfire(regionPath);
+ else
+ javaFunctions(rdd1).saveToGemfire(regionPath, connConf);
+
+ Set<String> keys = region.keySetOnServer();
+ Map<String, Integer> map = region.getAll(keys);
+
+ List<Tuple2<String, Integer>> expectedList = new ArrayList<>();
+ for (int i = 0; i < numObjects; i ++) {
+ expectedList.add(new Tuple2<>("k_" + i, i));
+ }
+ matchMapAndPairList(map, expectedList);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // JavaSparkContext.gemfireRegion and where clause
+ // --------------------------------------------------------------------------------------------
+
+ @Test
+ public void testJavaSparkContextGemfireRegion() throws Exception {
+ prepareStrIntRegion(regionPath, 0, numObjects); // remove all entries
+ Properties emptyProps = new Properties();
+ GemFireJavaRegionRDD<String, Integer> rdd1 = javaFunctions(jsc).gemfireRegion(regionPath);
+ GemFireJavaRegionRDD<String, Integer> rdd2 = javaFunctions(jsc).gemfireRegion(regionPath, emptyProps);
+ GemFireJavaRegionRDD<String, Integer> rdd3 = javaFunctions(jsc).gemfireRegion(regionPath, connConf);
+ GemFireJavaRegionRDD<String, Integer> rdd4 = javaFunctions(jsc).gemfireRegion(regionPath, connConf, emptyProps);
+ GemFireJavaRegionRDD<String, Integer> rdd5 = rdd1.where("value.intValue() < 50");
+
+ HashMap<String, Integer> expectedMap = new HashMap<>();
+ for (int i = 0; i < numObjects; i ++) {
+ expectedMap.put("k_" + i, i);
+ }
+
+ matchMapAndPairList(expectedMap, rdd1.collect());
+ matchMapAndPairList(expectedMap, rdd2.collect());
+ matchMapAndPairList(expectedMap, rdd3.collect());
+ matchMapAndPairList(expectedMap, rdd4.collect());
+
+ HashMap<String, Integer> expectedMap2 = new HashMap<>();
+ for (int i = 0; i < 50; i ++) {
+ expectedMap2.put("k_" + i, i);
+ }
+
+ matchMapAndPairList(expectedMap2, rdd5.collect());
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // JavaPairRDD.joinGemfireRegion
+ // --------------------------------------------------------------------------------------------
+
+ @Test
+ public void testPairRDDJoinWithSameKeyType() throws Exception {
+ prepareStrIntRegion(regionPath, 0, numObjects);
+ JavaPairRDD<String, Integer> rdd1 = prepareStrIntJavaPairRDD(-5, 10);
+
+ JavaPairRDD<Tuple2<String, Integer>, Integer> rdd2a = javaFunctions(rdd1).joinGemfireRegion(regionPath);
+ JavaPairRDD<Tuple2<String, Integer>, Integer> rdd2b = javaFunctions(rdd1).joinGemfireRegion(regionPath, connConf);
+ // System.out.println("=== Result RDD =======\n" + rdd2a.collect() + "\n=========================");
+
+ HashMap<Tuple2<String, Integer>, Integer> expectedMap = new HashMap<>();
+ for (int i = 0; i < 10; i ++) {
+ expectedMap.put(new Tuple2<>("k_" + i, i), i);
+ }
+ matchMapAndPairList(expectedMap, rdd2a.collect());
+ matchMapAndPairList(expectedMap, rdd2b.collect());
+ }
+
+ static class IntIntPairToStrKeyFunction implements Function<Tuple2<Integer, Integer>, String> {
+ @Override public String call(Tuple2<Integer, Integer> pair) throws Exception {
+ return "k_" + pair._1();
+ }
+ }
+
+ @Test
+ public void testPairRDDJoinWithDiffKeyType() throws Exception {
+ prepareStrIntRegion(regionPath, 0, numObjects);
+ JavaPairRDD<Integer, Integer> rdd1 = prepareIntIntJavaPairRDD(-5, 10);
+ Function<Tuple2<Integer, Integer>, String> func = new IntIntPairToStrKeyFunction();
+
+ JavaPairRDD<Tuple2<Integer, Integer>, Integer> rdd2a = javaFunctions(rdd1).joinGemfireRegion(regionPath, func);
+ JavaPairRDD<Tuple2<Integer, Integer>, Integer> rdd2b = javaFunctions(rdd1).joinGemfireRegion(regionPath, func, connConf);
+ //System.out.println("=== Result RDD =======\n" + rdd2a.collect() + "\n=========================");
+
+ HashMap<Tuple2<Integer, Integer>, Integer> expectedMap = new HashMap<>();
+ for (int i = 0; i < 10; i ++) {
+ expectedMap.put(new Tuple2<>(i, i * 2), i);
+ }
+ matchMapAndPairList(expectedMap, rdd2a.collect());
+ matchMapAndPairList(expectedMap, rdd2b.collect());
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // JavaPairRDD.outerJoinGemfireRegion
+ // --------------------------------------------------------------------------------------------
+
+ @Test
+ public void testPairRDDOuterJoinWithSameKeyType() throws Exception {
+ prepareStrIntRegion(regionPath, 0, numObjects);
+ JavaPairRDD<String, Integer> rdd1 = prepareStrIntJavaPairRDD(-5, 10);
+
+ JavaPairRDD<Tuple2<String, Integer>, Option<Integer>> rdd2a = javaFunctions(rdd1).outerJoinGemfireRegion(regionPath);
+ JavaPairRDD<Tuple2<String, Integer>, Option<Integer>> rdd2b = javaFunctions(rdd1).outerJoinGemfireRegion(regionPath, connConf);
+ //System.out.println("=== Result RDD =======\n" + rdd2a.collect() + "\n=========================");
+
+ HashMap<Tuple2<String, Integer>, Option<Integer>> expectedMap = new HashMap<>();
+ for (int i = -5; i < 10; i ++) {
+ if (i < 0)
+ expectedMap.put(new Tuple2<>("k_" + i, i), Option.apply((Integer) null));
+ else
+ expectedMap.put(new Tuple2<>("k_" + i, i), Some.apply(i));
+ }
+ matchMapAndPairList(expectedMap, rdd2a.collect());
+ matchMapAndPairList(expectedMap, rdd2b.collect());
+ }
+
+ @Test
+ public void testPairRDDOuterJoinWithDiffKeyType() throws Exception {
+ prepareStrIntRegion(regionPath, 0, numObjects);
+ JavaPairRDD<Integer, Integer> rdd1 = prepareIntIntJavaPairRDD(-5, 10);
+ Function<Tuple2<Integer, Integer>, String> func = new IntIntPairToStrKeyFunction();
+
+ JavaPairRDD<Tuple2<Integer, Integer>, Option<Integer>> rdd2a = javaFunctions(rdd1).outerJoinGemfireRegion(regionPath, func);
+ JavaPairRDD<Tuple2<Integer, Integer>, Option<Integer>> rdd2b = javaFunctions(rdd1).outerJoinGemfireRegion(regionPath, func, connConf);
+ //System.out.println("=== Result RDD =======\n" + rdd2a.collect() + "\n=========================");
+
+ HashMap<Tuple2<Integer, Integer>, Option<Integer>> expectedMap = new HashMap<>();
+ for (int i = -5; i < 10; i ++) {
+ if (i < 0)
+ expectedMap.put(new Tuple2<>(i, i * 2), Option.apply((Integer) null));
+ else
+ expectedMap.put(new Tuple2<>(i, i * 2), Some.apply(i));
+ }
+ matchMapAndPairList(expectedMap, rdd2a.collect());
+ matchMapAndPairList(expectedMap, rdd2b.collect());
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // JavaRDD.joinGemfireRegion
+ // --------------------------------------------------------------------------------------------
+
+ static class IntToStrKeyFunction implements Function<Integer, String> {
+ @Override public String call(Integer x) throws Exception {
+ return "k_" + x;
+ }
+ }
+
+ @Test
+ public void testRDDJoinWithSameKeyType() throws Exception {
+ prepareStrIntRegion(regionPath, 0, numObjects);
+ JavaRDD<Integer> rdd1 = prepareIntJavaRDD(-5, 10);
+
+ Function<Integer, String> func = new IntToStrKeyFunction();
+ JavaPairRDD<Integer, Integer> rdd2a = javaFunctions(rdd1).joinGemfireRegion(regionPath, func);
+ JavaPairRDD<Integer, Integer> rdd2b = javaFunctions(rdd1).joinGemfireRegion(regionPath, func, connConf);
+ //System.out.println("=== Result RDD =======\n" + rdd2a.collect() + "\n=========================");
+
+ HashMap<Integer, Integer> expectedMap = new HashMap<>();
+ for (int i = 0; i < 10; i ++) {
+ expectedMap.put(i, i);
+ }
+ matchMapAndPairList(expectedMap, rdd2a.collect());
+ matchMapAndPairList(expectedMap, rdd2b.collect());
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // JavaRDD.outerJoinGemfireRegion
+ // --------------------------------------------------------------------------------------------
+
+ @Test
+ public void testRDDOuterJoinWithSameKeyType() throws Exception {
+ prepareStrIntRegion(regionPath, 0, numObjects);
+ JavaRDD<Integer> rdd1 = prepareIntJavaRDD(-5, 10);
+
+ Function<Integer, String> func = new IntToStrKeyFunction();
+ JavaPairRDD<Integer, Option<Integer>> rdd2a = javaFunctions(rdd1).outerJoinGemfireRegion(regionPath, func);
+ JavaPairRDD<Integer, Option<Integer>> rdd2b = javaFunctions(rdd1).outerJoinGemfireRegion(regionPath, func, connConf);
+ //System.out.println("=== Result RDD =======\n" + rdd2a.collect() + "\n=========================");
+
+ HashMap<Integer, Option<Integer>> expectedMap = new HashMap<>();
+ for (int i = -5; i < 10; i ++) {
+ if (i < 0)
+ expectedMap.put(i, Option.apply((Integer) null));
+ else
+ expectedMap.put(i, Some.apply(i));
+ }
+ matchMapAndPairList(expectedMap, rdd2a.collect());
+ matchMapAndPairList(expectedMap, rdd2b.collect());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/85d44f04/gemfire-spark-connector/gemfire-spark-connector/src/it/java/ittest/io/pivotal/gemfire/spark/connector/Portfolio.java
----------------------------------------------------------------------
diff --git a/gemfire-spark-connector/gemfire-spark-connector/src/it/java/ittest/io/pivotal/gemfire/spark/connector/Portfolio.java b/gemfire-spark-connector/gemfire-spark-connector/src/it/java/ittest/io/pivotal/gemfire/spark/connector/Portfolio.java
new file mode 100644
index 0000000..d5e9c85
--- /dev/null
+++ b/gemfire-spark-connector/gemfire-spark-connector/src/it/java/ittest/io/pivotal/gemfire/spark/connector/Portfolio.java
@@ -0,0 +1,93 @@
+package ittest.io.pivotal.gemfire.spark.connector;
+
+import java.io.Serializable;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Properties;
+import com.gemstone.gemfire.cache.Declarable;
+
+/**
+ * A stock portfolio that consists of multiple {@link Position} objects that
+ * represent shares of stock (a "security"). Instances of
+ * <code>Portfolio</code> can be stored in a GemFire <code>Region</code> and
+ * their contents can be queried using the GemFire query service.
+ * </p>
+ * This class is <code>Serializable</code> because we want it to be distributed
+ * to multiple members of a distributed system. Because this class is
+ * <code>Declarable</code>, we can describe instances of it in a GemFire
+ * <code>cache.xml</code> file.
+ * </p>
+ *
+ */
+public class Portfolio implements Declarable, Serializable {
+
+ private static final long serialVersionUID = 9097335119586059309L;
+
+ private int id; /* id is used as the entry key and is stored in the entry */
+ private String type;
+ private Map<String,Position> positions = new LinkedHashMap<String,Position>();
+ private String status;
+
+ public Portfolio(Properties props) {
+ init(props);
+ }
+
+ @Override
+ public void init(Properties props) {
+ this.id = Integer.parseInt(props.getProperty("id"));
+ this.type = props.getProperty("type", "type1");
+ this.status = props.getProperty("status", "active");
+
+ // get the positions. These are stored in the properties object
+ // as Positions, not String, so use Hashtable protocol to get at them.
+ // the keys are named "positionN", where N is an integer.
+ for (Map.Entry<Object, Object> entry: props.entrySet()) {
+ String key = (String)entry.getKey();
+ if (key.startsWith("position")) {
+ Position pos = (Position)entry.getValue();
+ this.positions.put(pos.getSecId(), pos);
+ }
+ }
+ }
+
+ public void setType(String t) {this.type = t; }
+
+ public String getStatus(){
+ return status;
+ }
+
+ public int getId(){
+ return this.id;
+ }
+
+ public Map<String,Position> getPositions(){
+ return this.positions;
+ }
+
+ public String getType() {
+ return this.type;
+ }
+
+ public boolean isActive(){
+ return status.equals("active");
+ }
+
+ @Override
+ public String toString(){
+ StringBuilder buf = new StringBuilder();
+ buf.append("\n\tPortfolio [id=" + this.id + " status=" + this.status);
+ buf.append(" type=" + this.type);
+ boolean firstTime = true;
+ for (Map.Entry<String, Position> entry: positions.entrySet()) {
+ if (!firstTime) {
+ buf.append(", ");
+ }
+ buf.append("\n\t\t");
+ buf.append(entry.getKey() + ":" + entry.getValue());
+ firstTime = false;
+ }
+ buf.append("]");
+ return buf.toString();
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/85d44f04/gemfire-spark-connector/gemfire-spark-connector/src/it/java/ittest/io/pivotal/gemfire/spark/connector/Position.java
----------------------------------------------------------------------
diff --git a/gemfire-spark-connector/gemfire-spark-connector/src/it/java/ittest/io/pivotal/gemfire/spark/connector/Position.java b/gemfire-spark-connector/gemfire-spark-connector/src/it/java/ittest/io/pivotal/gemfire/spark/connector/Position.java
new file mode 100644
index 0000000..8940d4f
--- /dev/null
+++ b/gemfire-spark-connector/gemfire-spark-connector/src/it/java/ittest/io/pivotal/gemfire/spark/connector/Position.java
@@ -0,0 +1,57 @@
+package ittest.io.pivotal.gemfire.spark.connector;
+
+import java.io.Serializable;
+import java.util.Properties;
+import com.gemstone.gemfire.cache.Declarable;
+
+/**
+ * Represents a number of shares of a stock ("security") held in a {@link
+ * Portfolio}.
+ * </p>
+ * This class is <code>Serializable</code> because we want it to be distributed
+ * to multiple members of a distributed system. Because this class is
+ * <code>Declarable</code>, we can describe instances of it in a GemFire
+ * <code>cache.xml</code> file.
+ * </p>
+ *
+ */
+public class Position implements Declarable, Serializable {
+
+ private static final long serialVersionUID = -8229531542107983344L;
+
+ private String secId;
+ private double qty;
+ private double mktValue;
+
+ public Position(Properties props) {
+ init(props);
+ }
+
+ @Override
+ public void init(Properties props) {
+ this.secId = props.getProperty("secId");
+ this.qty = Double.parseDouble(props.getProperty("qty"));
+ this.mktValue = Double.parseDouble(props.getProperty("mktValue"));
+ }
+
+ public String getSecId(){
+ return this.secId;
+ }
+
+ public double getQty(){
+ return this.qty;
+ }
+
+ public double getMktValue() {
+ return this.mktValue;
+ }
+
+ @Override
+ public String toString(){
+ return new StringBuilder()
+ .append("Position [secId=").append(secId)
+ .append(" qty=").append(this.qty)
+ .append(" mktValue=").append(mktValue).append("]").toString();
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/85d44f04/gemfire-spark-connector/gemfire-spark-connector/src/it/resources/test-regions.xml
----------------------------------------------------------------------
diff --git a/gemfire-spark-connector/gemfire-spark-connector/src/it/resources/test-regions.xml b/gemfire-spark-connector/gemfire-spark-connector/src/it/resources/test-regions.xml
new file mode 100644
index 0000000..9c59d7d
--- /dev/null
+++ b/gemfire-spark-connector/gemfire-spark-connector/src/it/resources/test-regions.xml
@@ -0,0 +1,32 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE cache PUBLIC
+ "-//GemStone Systems, Inc.//GemFire Declarative Caching 6.5//EN"
+ "http://www.gemstone.com/dtd/cache6_5.dtd" >
+
+<cache>
+ <!-- test region for OQL test -->
+ <region name="obj_obj_region" refid="PARTITION_REDUNDANT" />
+
+ <region name="obj_obj_rep_region" refid="REPLICATE" />
+
+ <region name="str_int_region" refid="PARTITION_REDUNDANT">
+ <region-attributes>
+ <key-constraint>java.lang.String</key-constraint>
+ <value-constraint>java.lang.Integer</value-constraint>
+ </region-attributes>
+ </region>
+
+ <region name="str_str_region" refid="PARTITION_REDUNDANT">
+ <region-attributes>
+ <key-constraint>java.lang.String</key-constraint>
+ <value-constraint>java.lang.String</value-constraint>
+ </region-attributes>
+ </region>
+
+ <region name="str_str_rep_region" refid="REPLICATE">
+ <region-attributes>
+ <key-constraint>java.lang.String</key-constraint>
+ <value-constraint>java.lang.String</value-constraint>
+ </region-attributes>
+ </region>
+</cache>
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/85d44f04/gemfire-spark-connector/gemfire-spark-connector/src/it/resources/test-retrieve-regions.xml
----------------------------------------------------------------------
diff --git a/gemfire-spark-connector/gemfire-spark-connector/src/it/resources/test-retrieve-regions.xml b/gemfire-spark-connector/gemfire-spark-connector/src/it/resources/test-retrieve-regions.xml
new file mode 100644
index 0000000..1608751
--- /dev/null
+++ b/gemfire-spark-connector/gemfire-spark-connector/src/it/resources/test-retrieve-regions.xml
@@ -0,0 +1,40 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE cache PUBLIC
+ "-//GemStone Systems, Inc.//GemFire Declarative Caching 6.5//EN"
+ "http://www.gemstone.com/dtd/cache6_5.dtd" >
+
+<cache>
+ <!-- combinations of key, value types with region types -->
+ <region name="pr_r_obj_obj_region" refid="PARTITION_REDUNDANT" />
+ <region name="pr_obj_obj_region" refid="PARTITION" />
+ <region name="rr_obj_obj_region" refid="REPLICATE" />
+ <region name="rr_p_obj_obj_region" refid="REPLICATE_PERSISTENT" />
+
+ <region name="pr_r_str_int_region" refid="PARTITION_REDUNDANT">
+ <region-attributes>
+ <key-constraint>java.lang.String</key-constraint>
+ <value-constraint>java.lang.Integer</value-constraint>
+ </region-attributes>
+ </region>
+
+ <region name="pr_str_int_region" refid="PARTITION">
+ <region-attributes>
+ <key-constraint>java.lang.String</key-constraint>
+ <value-constraint>java.lang.Integer</value-constraint>
+ </region-attributes>
+ </region>
+
+ <region name="rr_str_int_region" refid="REPLICATE">
+ <region-attributes>
+ <key-constraint>java.lang.String</key-constraint>
+ <value-constraint>java.lang.Integer</value-constraint>
+ </region-attributes>
+ </region>
+
+ <region name="rr_p_str_int_region" refid="REPLICATE_PERSISTENT">
+ <region-attributes>
+ <key-constraint>java.lang.String</key-constraint>
+ <value-constraint>java.lang.Integer</value-constraint>
+ </region-attributes>
+ </region>
+</cache>
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/85d44f04/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/BasicIntegrationTest.scala
----------------------------------------------------------------------
diff --git a/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/BasicIntegrationTest.scala b/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/BasicIntegrationTest.scala
new file mode 100644
index 0000000..dfd29e1
--- /dev/null
+++ b/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/BasicIntegrationTest.scala
@@ -0,0 +1,582 @@
+package ittest.io.pivotal.gemfire.spark.connector
+
+import java.util.Properties
+import com.gemstone.gemfire.cache.query.QueryService
+import com.gemstone.gemfire.cache.query.internal.StructImpl
+import io.pivotal.gemfire.spark.connector._
+import com.gemstone.gemfire.cache.Region
+import io.pivotal.gemfire.spark.connector.internal.{RegionMetadata, DefaultGemFireConnectionManager}
+import io.pivotal.gemfire.spark.connector.internal.oql.{RDDConverter, QueryRDD}
+import ittest.io.pivotal.gemfire.spark.connector.testkit.GemFireCluster
+import ittest.io.pivotal.gemfire.spark.connector.testkit.IOUtils
+import org.apache.spark.streaming.{Seconds, StreamingContext, TestInputDStream}
+import org.apache.spark.{SparkContext, SparkConf}
+import org.scalatest.{BeforeAndAfterAll, FunSuite, Matchers}
+import scala.collection.JavaConversions
+import scala.reflect.ClassTag
+
+case class Number(str: String, len: Int)
+
+class BasicIntegrationTest extends FunSuite with Matchers with BeforeAndAfterAll with GemFireCluster {
+
+ var sc: SparkContext = null
+
+ override def beforeAll() {
+ // start gemfire cluster, and spark context
+ val settings = new Properties()
+ settings.setProperty("cache-xml-file", "src/it/resources/test-regions.xml")
+ settings.setProperty("num-of-servers", "2")
+ val locatorPort = GemFireCluster.start(settings)
+
+ // start spark context in local mode
+ IOUtils.configTestLog4j("ERROR", "log4j.logger.org.apache.spark" -> "INFO",
+ "log4j.logger.io.pivotal.gemfire.spark.connector" -> "DEBUG")
+ val conf = new SparkConf()
+ .setAppName("BasicIntegrationTest")
+ .setMaster("local[2]")
+ .set("spark.streaming.clock", "org.apache.spark.streaming.util.ManualClock")
+ .set(GemFireLocatorPropKey, s"localhost[$locatorPort]")
+ .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
+ .set("spark.kryo.registrator", "io.pivotal.gemfire.spark.connector.GemFireKryoRegistrator")
+
+ sc = new SparkContext(conf)
+ }
+
+ override def afterAll() {
+ // stop connection, spark context, and gemfire cluster
+ DefaultGemFireConnectionManager.closeConnection(GemFireConnectionConf(sc.getConf))
+ sc.stop()
+ GemFireCluster.stop()
+ }
+
+ //Convert Map[Object, Object] to java.util.Properties
+ private def map2Props(map: Map[Object, Object]): java.util.Properties =
+ (new java.util.Properties /: map) {case (props, (k,v)) => props.put(k,v); props}
+
+ // ===========================================================
+ // DefaultGemFireConnection functional tests
+ // ===========================================================
+
+ test("DefaultGemFireConnection.validateRegion()") {
+ val conn = GemFireConnectionConf(sc.getConf).getConnection
+
+ // normal exist-region
+ var regionPath: String = "str_str_region"
+ conn.validateRegion[String, String](regionPath)
+
+ // non-exist region
+ regionPath = "non_exist_region"
+ try {
+ conn.validateRegion[String, String](regionPath)
+ fail("validateRegion failed to catch non-exist region error")
+ } catch {
+ case e: RuntimeException =>
+ if (! e.getMessage.contains(s"The region named $regionPath was not found"))
+ fail("validateRegion gives wrong exception on non-exist region", e)
+ case e: Throwable =>
+ fail("validateRegion gives wrong exception on non-exist region", e)
+ }
+
+ // Note: currently, can't catch type mismatch error
+ conn.validateRegion[String, Integer]("str_str_region")
+ }
+
+ test("DefaultGemFireConnection.getRegionMetadata()") {
+ val conn = GemFireConnectionConf(sc.getConf).getConnection
+
+ // exist region
+ validateRegionMetadata(conn, "obj_obj_region", true, 113, null, null, false)
+ validateRegionMetadata(conn, "str_int_region", true, 113, "java.lang.String", "java.lang.Integer", false)
+ validateRegionMetadata(conn, "str_str_rep_region", false, 0, "java.lang.String", "java.lang.String", true)
+
+ // non-exist region
+ assert(! conn.getRegionMetadata("no_exist_region").isDefined)
+ }
+
+ def validateRegionMetadata(
+ conn: GemFireConnection, regionPath: String, partitioned: Boolean, buckets: Int,
+ keyType: String, valueType: String, emptyMap: Boolean): Unit = {
+
+ val mdOption = conn.getRegionMetadata(regionPath)
+ val md = mdOption.get
+
+ assert(md.getRegionPath == s"/$regionPath")
+ assert(md.isPartitioned == partitioned)
+ assert(md.getKeyTypeName == keyType)
+ assert(md.getValueTypeName == valueType)
+ assert(md.getTotalBuckets == buckets)
+ if (emptyMap) assert(md.getServerBucketMap == null)
+ else assert(md.getServerBucketMap != null)
+ }
+
+ test("DefaultGemFireConnection.getRegionProxy()") {
+ val conn = GemFireConnectionConf(sc.getConf).getConnection
+
+ val region1 = conn.getRegionProxy[String, String]("str_str_region")
+ region1.put("1", "One")
+ assert(region1.get("1") == "One")
+ region1.remove("1")
+ assert(region1.get("1") == null)
+
+ // getRegionProxy doesn't fail when region doesn't exist
+ val region2 = conn.getRegionProxy[String, String]("non_exist_region")
+ try {
+ region2.put("1", "One")
+ fail("getRegionProxy failed to catch non-exist region error")
+ } catch {
+ case e: Exception =>
+ if (e.getCause == null || ! e.getCause.getMessage.contains(s"Region named /non_exist_region was not found")) {
+ e.printStackTrace()
+ fail("validateRegion gives wrong exception on non-exist region", e)
+ }
+ }
+ }
+
+ // Note: DefaultGemFireConnecton.getQuery() and getRegionData() are covered by
+ // RetrieveRegionIntegrationTest.scala and following OQL tests.
+
+ // ===========================================================
+ // OQL functional tests
+ // ===========================================================
+
+ private def initRegion(regionName: String): Unit = {
+
+ //Populate some data in the region
+ val conn = GemFireConnectionConf(sc.getConf).getConnection
+ val rgn: Region[Object, Object] = conn.getRegionProxy(regionName)
+ rgn.removeAll(rgn.keySetOnServer())
+
+ //This will call the implicit conversion map2Properties in connector package object, since it is Map[String, String]
+ var position1 = new Position(Map("secId" -> "SUN", "qty" -> "34000", "mktValue" -> "24.42"))
+ var position2 = new Position(Map("secId" -> "IBM", "qty" -> "8765", "mktValue" -> "34.29"))
+ val portfolio1 = new Portfolio(map2Props(Map("id" ->"1", "type" -> "type1", "status" -> "active",
+ "position1" -> position1, "position2" -> position2)))
+ rgn.put("1", portfolio1)
+
+ position1 = new Position(Map("secId" -> "YHOO", "qty" -> "9834", "mktValue" -> "12.925"))
+ position2 = new Position(Map("secId" -> "GOOG", "qty" -> "12176", "mktValue" -> "21.972"))
+ val portfolio2 = new Portfolio(map2Props(Map("id" -> "2", "type" -> "type2", "status" -> "inactive",
+ "position1" -> position1, "position2" -> position2)))
+ rgn.put("2", portfolio2)
+
+ position1 = new Position(Map("secId" -> "MSFT", "qty" -> "98327", "mktValue" -> "23.32"))
+ position2 = new Position(Map("secId" -> "AOL", "qty" -> "978", "mktValue" -> "40.373"))
+ val portfolio3 = new Portfolio(map2Props(Map("id" -> "3", "type" -> "type3", "status" -> "active",
+ "position1" -> position1, "position2" -> position2)))
+ rgn.put("3", portfolio3)
+
+ position1 = new Position(Map("secId" -> "APPL", "qty" -> "67", "mktValue" -> "67.356572"))
+ position2 = new Position(Map("secId" -> "ORCL", "qty" -> "376", "mktValue" -> "101.34"))
+ val portfolio4 = new Portfolio(map2Props(Map("id" -> "4", "type" -> "type1", "status" -> "inactive",
+ "position1" -> position1, "position2" -> position2)))
+ rgn.put("4", portfolio4)
+
+ position1 = new Position(Map("secId" -> "SAP", "qty" -> "90", "mktValue" -> "67.356572"))
+ position2 = new Position(Map("secId" -> "DELL", "qty" -> "376", "mktValue" -> "101.34"))
+ val portfolio5 = new Portfolio(map2Props(Map("id" -> "5", "type" -> "type2", "status" -> "active",
+ "position1" -> position1, "position2" -> position2)))
+ rgn.put("5", portfolio5)
+
+ position1 = new Position(Map("secId" -> "RHAT", "qty" -> "90", "mktValue" -> "67.356572"))
+ position2 = new Position(Map("secId" -> "NOVL", "qty" -> "376", "mktValue" -> "101.34"))
+ val portfolio6 = new Portfolio(map2Props(Map("id" -> "6", "type" -> "type3", "status" -> "inactive",
+ "position1" -> position1, "position2" -> position2)))
+ rgn.put("6", portfolio6)
+
+ position1 = new Position(Map("secId" -> "MSFT", "qty" -> "98327", "mktValue" -> "23.32"))
+ position2 = new Position(Map("secId" -> "AOL", "qty" -> "978", "mktValue" -> "40.373"))
+ val portfolio7 = new Portfolio(map2Props(Map("id" -> "7", "type" -> "type4", "status" -> "active",
+ "position1" -> position1, "position2" -> position2)))
+ //Not using null, due to intermittent query failure on column containing null, likely a Spark SQL bug
+ //portfolio7.setType(null)
+ rgn.put("7", portfolio7)
+ }
+
+ private def getQueryRDD[T: ClassTag](
+ query: String, connConf: GemFireConnectionConf = GemFireConnectionConf(sc.getConf)): QueryRDD[T] =
+ new QueryRDD[T](sc, query, connConf)
+
+ test("Run GemFire OQL query and convert the returned QueryRDD to DataFrame: Partitioned Region") {
+ simpleQuery("obj_obj_region")
+ }
+
+ test("Run GemFire OQL query and convert the returned QueryRDD to DataFrame: Replicated Region") {
+ simpleQuery("obj_obj_rep_region")
+ }
+
+ private def simpleQuery(regionName: String) {
+ //Populate some data in the region
+ val connConf: GemFireConnectionConf = GemFireConnectionConf(sc.getConf)
+ val conn = connConf.getConnection
+ val rgn: Region[String, String] = conn.getRegionProxy(regionName)
+ rgn.removeAll(rgn.keySetOnServer())
+ rgn.putAll(JavaConversions.mapAsJavaMap(Map("1" -> "one", "2" -> "two", "3" -> "three")))
+
+ //Create QueryRDD using OQL
+ val OQLResult: QueryRDD[String] = getQueryRDD[String](s"select * from /$regionName")
+
+ //verify the QueryRDD
+ val oqlRS: Array[String] = OQLResult.collect()
+ oqlRS should have length 3
+ oqlRS should contain theSameElementsAs List("one", "two", "three")
+
+ //Convert QueryRDD to DataFrame
+ val sqlContext = new org.apache.spark.sql.SQLContext(sc)
+ // this is used to implicitly convert an RDD to a DataFrame.
+ import sqlContext.implicits._
+ val dataFrame = OQLResult.map(x => Number(x, x.length)).toDF()
+ //Register dataFrame as a table of two columns of type String and Int respectively
+ dataFrame.registerTempTable("numberTable")
+
+ //Issue SQL query against the table
+ val SQLResult = sqlContext.sql("SELECT * FROM numberTable")
+ //Verify the SQL query result, r(0) mean column 0
+ val sqlRS: Array[Any] = SQLResult.map(r => r(0)).collect()
+ sqlRS should have length 3
+ sqlRS should contain theSameElementsAs List("one", "two", "three")
+
+ //Convert QueryRDD to DataFrame using RDDConverter
+ val dataFrame2 = RDDConverter.queryRDDToDataFrame(OQLResult, sqlContext)
+ //Register dataFrame2 as a table of two columns of type String and Int respectively
+ dataFrame2.registerTempTable("numberTable2")
+
+ //Issue SQL query against the table
+ val SQLResult2 = sqlContext.sql("SELECT * FROM numberTable2")
+ //Verify the SQL query result, r(0) mean column 0
+ val sqlRS2: Array[Any] = SQLResult2.map(r => r(0)).collect()
+ sqlRS2 should have length 3
+ sqlRS2 should contain theSameElementsAs List("one", "two", "three")
+
+ //Remove the region entries, because other tests might use the same region as well
+ List("1", "2", "3").foreach(rgn.remove)
+ }
+
+ test("Run GemFire OQL query and directly return DataFrame: Partitioned Region") {
+ simpleQueryDataFrame("obj_obj_region")
+ }
+
+ test("Run GemFire OQL query and directly return DataFrame: Replicated Region") {
+ simpleQueryDataFrame("obj_obj_rep_region")
+ }
+
+ private def simpleQueryDataFrame(regionName: String) {
+ //Populate some data in the region
+ val conn = GemFireConnectionConf(sc.getConf).getConnection
+ val rgn: Region[String, String] = conn.getRegionProxy(regionName)
+ rgn.removeAll(rgn.keySetOnServer())
+ rgn.putAll(JavaConversions.mapAsJavaMap(Map("1" -> "one", "2" -> "two", "3" -> "three")))
+
+ //Create DataFrame using GemFire OQL
+ val sqlContext = new org.apache.spark.sql.SQLContext(sc)
+ val dataFrame = sqlContext.gemfireOQL(s"select * from /$regionName")
+ dataFrame.registerTempTable("numberTable")
+
+ //Issue SQL query against the table
+ val SQLResult = sqlContext.sql("SELECT * FROM numberTable")
+ //Verify the SQL query result, r(0) mean column 0
+ val sqlRS: Array[Any] = SQLResult.map(r => r(0)).collect()
+ sqlRS should have length 3
+ sqlRS should contain theSameElementsAs List("one", "two", "three")
+
+ //Remove the region entries, because other tests might use the same region as well
+ List("1", "2", "3").foreach(rgn.remove)
+ }
+
+ test("GemFire OQL query with UDT: Partitioned Region") {
+ queryUDT("obj_obj_region")
+ }
+
+ test("GemFire OQL query with UDT: Replicated Region") {
+ queryUDT("obj_obj_rep_region")
+ }
+
+ private def queryUDT(regionName: String) {
+
+ //Populate some data in the region
+ val conn = GemFireConnectionConf(sc.getConf).getConnection
+ val rgn: Region[Object, Object] = conn.getRegionProxy(regionName)
+ rgn.removeAll(rgn.keySetOnServer())
+ val e1: Employee = new Employee("hello", 123)
+ val e2: Employee = new Employee("world", 456)
+ rgn.putAll(JavaConversions.mapAsJavaMap(Map("1" -> e1, "2" -> e2)))
+
+ //Create QueryRDD using OQL
+ val OQLResult: QueryRDD[Object] = getQueryRDD(s"select name, age from /$regionName")
+
+ //verify the QueryRDD
+ val oqlRS: Array[Object] = OQLResult.collect()
+ oqlRS should have length 2
+ oqlRS.map(e => e.asInstanceOf[StructImpl].getFieldValues.apply(1)) should contain theSameElementsAs List(123, 456)
+
+ //Convert QueryRDD to DataFrame
+ val sqlContext = new org.apache.spark.sql.SQLContext(sc)
+
+ //Convert QueryRDD to DataFrame using RDDConverter
+ val dataFrame = RDDConverter.queryRDDToDataFrame(OQLResult, sqlContext)
+ dataFrame.registerTempTable("employee")
+ val SQLResult = sqlContext.sql("SELECT * FROM employee")
+
+ //Verify the SQL query result
+ val sqlRS = SQLResult.map(r => r(0)).collect()
+ sqlRS should have length 2
+ sqlRS should contain theSameElementsAs List("hello", "world")
+
+ List("1", "2").foreach(rgn.remove)
+ }
+
+ test("GemFire OQL query with UDT and directly return DataFrame: Partitioned Region") {
+ queryUDTDataFrame("obj_obj_region")
+ }
+
+ test("GemFire OQL query with UDT and directly return DataFrame: Replicated Region") {
+ queryUDTDataFrame("obj_obj_rep_region")
+ }
+
+ private def queryUDTDataFrame(regionName: String) {
+ //Populate some data in the region
+ val conn = GemFireConnectionConf(sc.getConf).getConnection
+ val rgn: Region[Object, Object] = conn.getRegionProxy(regionName)
+ rgn.removeAll(rgn.keySetOnServer())
+ val e1: Employee = new Employee("hello", 123)
+ val e2: Employee = new Employee("world", 456)
+ rgn.putAll(JavaConversions.mapAsJavaMap(Map("1" -> e1, "2" -> e2)))
+
+ //Create DataFrame using GemFire OQL
+ val sqlContext = new org.apache.spark.sql.SQLContext(sc)
+ val dataFrame = sqlContext.gemfireOQL(s"select name, age from /$regionName")
+
+ dataFrame.registerTempTable("employee")
+ val SQLResult = sqlContext.sql("SELECT * FROM employee")
+
+ //Verify the SQL query result
+ val sqlRS = SQLResult.map(r => r(0)).collect()
+ sqlRS should have length 2
+ sqlRS should contain theSameElementsAs List("hello", "world")
+
+ List("1", "2").foreach(rgn.remove)
+ }
+
+ test("GemFire OQL query with more complex UDT: Partitioned Region") {
+ complexUDT("obj_obj_region")
+ }
+
+ test("GemFire OQL query with more complex UDT: Replicated Region") {
+ complexUDT("obj_obj_rep_region")
+ }
+
+ private def complexUDT(regionName: String) {
+
+ initRegion(regionName)
+
+ //Create QueryRDD using OQL
+ val OQLResult: QueryRDD[Object] = getQueryRDD(s"SELECT DISTINCT * FROM /$regionName WHERE status = 'active'")
+
+ //verify the QueryRDD
+ val oqlRS: Array[Int] = OQLResult.collect().map(r => r.asInstanceOf[Portfolio].getId)
+ oqlRS should contain theSameElementsAs List(1, 3, 5, 7)
+
+ //Convert QueryRDD to DataFrame
+ val sqlContext = new org.apache.spark.sql.SQLContext(sc)
+
+ //Convert QueryRDD to DataFrame using RDDConverter
+ val dataFrame = RDDConverter.queryRDDToDataFrame(OQLResult, sqlContext)
+
+ dataFrame.registerTempTable("Portfolio")
+
+ val SQLResult = sqlContext.sql("SELECT * FROM Portfolio")
+
+ //Verify the SQL query result
+ val sqlRS = SQLResult.collect().map(r => r(0).asInstanceOf[Portfolio].getType)
+ sqlRS should contain theSameElementsAs List("type1", "type2", "type3", "type4")
+ }
+
+ test("GemFire OQL query with more complex UDT and directly return DataFrame: Partitioned Region") {
+ complexUDTDataFrame("obj_obj_region")
+ }
+
+ test("GemFire OQL query with more complex UDT and directly return DataFrame: Replicated Region") {
+ complexUDTDataFrame("obj_obj_rep_region")
+ }
+
+ private def complexUDTDataFrame(regionName: String) {
+
+ initRegion(regionName)
+
+ //Create DataFrame using GemFire OQL
+ val sqlContext = new org.apache.spark.sql.SQLContext(sc)
+ val dataFrame = sqlContext.gemfireOQL(s"SELECT DISTINCT * FROM /$regionName WHERE status = 'active'")
+ dataFrame.registerTempTable("Portfolio")
+
+ val SQLResult = sqlContext.sql("SELECT * FROM Portfolio")
+
+ //Verify the SQL query result
+ val sqlRS = SQLResult.collect().map(r => r(0).asInstanceOf[Portfolio].getType)
+ sqlRS should contain theSameElementsAs List("type1", "type2", "type3", "type4")
+ }
+
+ test("GemFire OQL query with more complex UDT with Projection: Partitioned Region") {
+ queryComplexUDTProjection("obj_obj_region")
+ }
+
+ test("GemFire OQL query with more complex UDT with Projection: Replicated Region") {
+ queryComplexUDTProjection("obj_obj_rep_region")
+ }
+
+ private def queryComplexUDTProjection(regionName: String) {
+
+ initRegion(regionName)
+
+ //Create QueryRDD using OQL
+ val OQLResult: QueryRDD[Object] = getQueryRDD[Object](s"""SELECT id, "type", positions, status FROM /$regionName WHERE status = 'active'""")
+
+ //verify the QueryRDD
+ val oqlRS: Array[Int] = OQLResult.collect().map(si => si.asInstanceOf[StructImpl].getFieldValues.apply(0).asInstanceOf[Int])
+ oqlRS should contain theSameElementsAs List(1, 3, 5, 7)
+
+ //Convert QueryRDD to DataFrame
+ val sqlContext = new org.apache.spark.sql.SQLContext(sc)
+
+ //Convert QueryRDD to DataFrame using RDDConverter
+ val dataFrame = RDDConverter.queryRDDToDataFrame(OQLResult, sqlContext)
+
+ dataFrame.registerTempTable("Portfolio")
+
+ val SQLResult = sqlContext.sql("SELECT id, type FROM Portfolio where type = 'type3'")
+
+ //Verify the SQL query result
+ val sqlRS = SQLResult.collect().map(r => r(0))
+ sqlRS should contain theSameElementsAs List(3)
+ }
+
+ test("GemFire OQL query with more complex UDT with Projection and directly return DataFrame: Partitioned Region") {
+ queryComplexUDTProjectionDataFrame("obj_obj_region")
+ }
+
+ test("GemFire OQL query with more complex UDT with Projection and directly return DataFrame: Replicated Region") {
+ queryComplexUDTProjectionDataFrame("obj_obj_rep_region")
+ }
+
+ private def queryComplexUDTProjectionDataFrame(regionName: String) {
+
+ initRegion(regionName)
+
+ //Create DataFrame using GemFire OQL
+ val sqlContext = new org.apache.spark.sql.SQLContext(sc)
+ val dataFrame = sqlContext.gemfireOQL(s"""SELECT id, "type", positions, status FROM /$regionName WHERE status = 'active'""")
+ dataFrame.registerTempTable("Portfolio")
+
+ val SQLResult = sqlContext.sql("SELECT id, type FROM Portfolio where type = 'type3'")
+
+ //Verify the SQL query result
+ val sqlRS = SQLResult.collect().map(r => r(0))
+ sqlRS should contain theSameElementsAs List(3)
+ }
+
+ test("GemFire OQL query with more complex UDT with nested Projection and directly return DataFrame: Partitioned Region") {
+ queryComplexUDTNestProjectionDataFrame("obj_obj_region")
+ }
+
+ test("GemFire OQL query with more complex UDT with nested Projection and directly return DataFrame: Replicated Region") {
+ queryComplexUDTNestProjectionDataFrame("obj_obj_rep_region")
+ }
+
+ private def queryComplexUDTNestProjectionDataFrame(regionName: String) {
+
+ initRegion(regionName)
+
+ //Create DataFrame using GemFire OQL
+ val sqlContext = new org.apache.spark.sql.SQLContext(sc)
+ val dataFrame = sqlContext.gemfireOQL(s"""SELECT r.id, r."type", r.positions, r.status FROM /$regionName r, r.positions.values f WHERE r.status = 'active' and f.secId = 'MSFT'""")
+ dataFrame.registerTempTable("Portfolio")
+
+ val SQLResult = sqlContext.sql("SELECT id, type FROM Portfolio where type = 'type3'")
+
+ //Verify the SQL query result
+ val sqlRS = SQLResult.collect().map(r => r(0))
+ sqlRS should contain theSameElementsAs List(3)
+ }
+
+ test("Undefined instance deserialization: Partitioned Region") {
+ undefinedInstanceDeserialization("obj_obj_region")
+ }
+
+ test("Undefined instance deserialization: Replicated Region") {
+ undefinedInstanceDeserialization("obj_obj_rep_region")
+ }
+
+ private def undefinedInstanceDeserialization(regionName: String) {
+
+ val conn = GemFireConnectionConf(sc.getConf).getConnection
+ val rgn: Region[Object, Object] = conn.getRegionProxy(regionName)
+ rgn.removeAll(rgn.keySetOnServer())
+
+ //Put some new data
+ rgn.put("1", "one")
+
+ //Query some non-existent columns, which should return UNDEFINED
+ val sqlContext = new org.apache.spark.sql.SQLContext(sc)
+ val dataFrame = sqlContext.gemfireOQL(s"SELECT col100, col200 FROM /$regionName")
+ val col1 = dataFrame.first().apply(0)
+ val col2 = dataFrame.first().apply(1)
+ assert(col1 == QueryService.UNDEFINED)
+ assert(col2 == QueryService.UNDEFINED)
+ //Verify that col1 and col2 refer to the same Undefined object
+ assert(col1.asInstanceOf[AnyRef] eq col2.asInstanceOf[AnyRef])
+ }
+
+ test("RDD.saveToGemFire") {
+ val regionName = "str_str_region"
+ // generate: Vector((1,11), (2,22), (3,33), (4,44), (5,55), (6,66))
+ val data = (1 to 6).map(_.toString).map(e=> (e, e*2))
+ val rdd = sc.parallelize(data)
+ rdd.saveToGemfire(regionName)
+
+ // verify
+ val connConf: GemFireConnectionConf = GemFireConnectionConf(sc.getConf)
+ val region: Region[String, String] = connConf.getConnection.getRegionProxy(regionName)
+ println("region key set on server: " + region.keySetOnServer())
+ assert((1 to 6).map(_.toString).toSet == JavaConversions.asScalaSet(region.keySetOnServer()))
+ (1 to 6).map(_.toString).foreach(e => assert(e*2 == region.get(e)))
+ }
+
+ // ===========================================================
+ // DStream.saveToGemfire() functional tests
+ // ===========================================================
+
+ test("Basic DStream test") {
+ import org.apache.spark.streaming.scheduler.{StreamingListenerBatchCompleted, StreamingListener}
+ import io.pivotal.gemfire.spark.connector.streaming._
+ import org.apache.spark.streaming.ManualClockHelper
+
+ class TestStreamListener extends StreamingListener {
+ var count = 0
+ override def onBatchCompleted(batch: StreamingListenerBatchCompleted) = count += 1
+ }
+
+ def batchDuration = Seconds(1)
+ val ssc = new StreamingContext(sc, batchDuration)
+ val input = Seq(1 to 4, 5 to 8, 9 to 12)
+ val dstream = new TestInputDStream(ssc, input, 2)
+ dstream.saveToGemfire[String, Int]("str_int_region", (e: Int) => (e.toString, e))
+ try {
+ val listener = new TestStreamListener
+ ssc.addStreamingListener(listener)
+ ssc.start()
+ ManualClockHelper.addToTime(ssc, batchDuration.milliseconds * input.length)
+ while (listener.count < input.length) ssc.awaitTerminationOrTimeout(50)
+ } catch {
+ case e: Exception => e.printStackTrace(); throw e
+// } finally {
+// ssc.stop()
+ }
+
+ val connConf: GemFireConnectionConf = GemFireConnectionConf(sc.getConf)
+ val conn = connConf.getConnection
+ val region: Region[String, Int] = conn.getRegionProxy("str_int_region")
+
+ // verify gemfire region contents
+ println("region key set on server: " + region.keySetOnServer())
+ assert((1 to 12).map(_.toString).toSet == JavaConversions.asScalaSet(region.keySetOnServer()))
+ (1 to 12).foreach(e => assert(e == region.get(e.toString)))
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/85d44f04/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/RDDJoinRegionIntegrationTest.scala
----------------------------------------------------------------------
diff --git a/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/RDDJoinRegionIntegrationTest.scala b/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/RDDJoinRegionIntegrationTest.scala
new file mode 100644
index 0000000..1e1bde3
--- /dev/null
+++ b/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/RDDJoinRegionIntegrationTest.scala
@@ -0,0 +1,284 @@
+package ittest.io.pivotal.gemfire.spark.connector
+
+import java.util.Properties
+
+import io.pivotal.gemfire.spark.connector._
+import com.gemstone.gemfire.cache.Region
+import io.pivotal.gemfire.spark.connector.internal.DefaultGemFireConnectionManager
+import ittest.io.pivotal.gemfire.spark.connector.testkit.GemFireCluster
+import ittest.io.pivotal.gemfire.spark.connector.testkit.IOUtils
+import org.apache.spark.{SparkContext, SparkConf}
+import org.scalatest.{BeforeAndAfterAll, FunSuite, Matchers}
+import java.util.{HashMap => JHashMap}
+
+class RDDJoinRegionIntegrationTest extends FunSuite with Matchers with BeforeAndAfterAll with GemFireCluster {
+
+ var sc: SparkContext = null
+ val numServers = 3
+ val numObjects = 1000
+
+ override def beforeAll() {
+ // start gemfire cluster, and spark context
+ val settings = new Properties()
+ settings.setProperty("cache-xml-file", "src/it/resources/test-retrieve-regions.xml")
+ settings.setProperty("num-of-servers", numServers.toString)
+ val locatorPort = GemFireCluster.start(settings)
+
+ // start spark context in local mode
+ IOUtils.configTestLog4j("ERROR", "log4j.logger.org.apache.spark" -> "INFO",
+ "log4j.logger.io.pivotal.gemfire.spark.connector" -> "DEBUG")
+ val conf = new SparkConf()
+ .setAppName("RDDJoinRegionIntegrationTest")
+ .setMaster("local[2]")
+ .set(GemFireLocatorPropKey, s"localhost[$locatorPort]")
+ sc = new SparkContext(conf)
+ }
+
+ override def afterAll() {
+ // stop connection, spark context, and gemfire cluster
+ DefaultGemFireConnectionManager.closeConnection(GemFireConnectionConf(sc.getConf))
+ sc.stop()
+ GemFireCluster.stop()
+ }
+
+// def matchMaps[K,V](map1:Map[K,V], map2:Map[K,V]) = {
+// assert(map1.size == map2.size)
+// map1.foreach(e => {
+// assert(map2.contains(e._1))
+// assert (e._2 == map2.get(e._1).get)
+// })
+// }
+
+ // --------------------------------------------------------------------------------------------
+ // PairRDD.joinGemfireRegion[K2 <: K, V2](regionPath, connConf): GemFireJoinRDD[(K, V), K, V2]
+ // --------------------------------------------------------------------------------------------
+
+ test("PairRDD.joinGemFireRegion: RDD[K, V] with Region[K, V2], replicated region", JoinTest) {
+ verifyPairRDDJoinRegionWithSameKeyType("rr_str_int_region")
+ }
+
+ test("PairRDD.joinGemFireRegion: RDD[K, V] with Region[K, V2], partitioned region", JoinTest) {
+ verifyPairRDDJoinRegionWithSameKeyType("pr_str_int_region")
+ }
+
+ test("PairRDD.joinGemFireRegion: RDD[K, V] with Region[K, V2], partitioned redundant region", JoinTest) {
+ verifyPairRDDJoinRegionWithSameKeyType("pr_r_str_int_region")
+ }
+
+ def verifyPairRDDJoinRegionWithSameKeyType(regionPath: String): Unit = {
+ val entriesMap: JHashMap[String, Int] = new JHashMap()
+ (0 until numObjects).map(i => entriesMap.put("k_" + i, i))
+
+ val connConf: GemFireConnectionConf = GemFireConnectionConf(sc.getConf)
+ val conn = connConf.getConnection
+ val rgn: Region[String, Int] = conn.getRegionProxy(regionPath)
+ rgn.removeAll(rgn.keySetOnServer())
+ rgn.putAll(entriesMap)
+
+ val data = (-5 until 50).map(x => ("k_" + x, x*2))
+ val rdd = sc.parallelize(data)
+
+ val rdd2 = rdd.joinGemfireRegion[String, Int](regionPath, connConf)
+ val rdd2Content = rdd2.collect()
+
+ val expectedMap = (0 until 50).map(i => ((s"k_$i", i*2), i)).toMap
+ // matchMaps[(String, Int), Int](expectedMap, rdd2Content.toMap)
+ assert(expectedMap == rdd2Content.toMap)
+ }
+
+ // ------------------------------------------------------------------------------------------------------
+ // PairRDD.joinGemfireRegion[K2, V2](regionPath, ((K, V)) => K2, connConf): GemFireJoinRDD[(K, V), K2, V2]
+ // -------------------------------------------------------------------------------------------------------
+
+ test("PairRDD.joinGemFireRegion: RDD[K, V] with Region[K2, V2], replicated region", JoinTest) {
+ verifyPairRDDJoinRegionWithDiffKeyType("rr_str_int_region")
+ }
+
+ test("PairRDD.joinGemFireRegion: RDD[K, V] with Region[K2, V2], partitioned region", JoinTest) {
+ verifyPairRDDJoinRegionWithDiffKeyType("pr_str_int_region")
+ }
+
+ test("PairRDD.joinGemFireRegion: RDD[K, V] with Region[K2, V2], partitioned redundant region", JoinTest) {
+ verifyPairRDDJoinRegionWithDiffKeyType("pr_r_str_int_region")
+ }
+
+ def verifyPairRDDJoinRegionWithDiffKeyType(regionPath: String): Unit = {
+ val entriesMap: JHashMap[String, Int] = new JHashMap()
+ (0 until numObjects).map(i => entriesMap.put("k_" + i, i))
+
+ val connConf: GemFireConnectionConf = GemFireConnectionConf(sc.getConf)
+ val conn = connConf.getConnection
+ val rgn: Region[String, Int] = conn.getRegionProxy(regionPath)
+ rgn.removeAll(rgn.keySetOnServer())
+ rgn.putAll(entriesMap)
+
+ val data = (-5 until 50).map(x => (x, x*2))
+ val rdd = sc.parallelize(data)
+
+ val func :((Int, Int)) => String = pair => s"k_${pair._1}"
+
+ val rdd2 = rdd.joinGemfireRegion[String, Int](regionPath, func /*, connConf*/)
+ val rdd2Content = rdd2.collect()
+
+ val expectedMap = (0 until 50).map(i => ((i, i*2), i)).toMap
+ // matchMaps[(Int, Int), Int](expectedMap, rdd2Content.toMap)
+ assert(expectedMap == rdd2Content.toMap)
+ }
+
+ // ------------------------------------------------------------------------------------------------
+ // PairRDD.outerJoinGemfireRegion[K2 <: K, V2](regionPath, connConf): GemFireJoinRDD[(K, V), K, V2]
+ // ------------------------------------------------------------------------------------------------
+
+ test("PairRDD.outerJoinGemFireRegion: RDD[K, V] with Region[K, V2], replicated region", OuterJoinTest) {
+ verifyPairRDDOuterJoinRegionWithSameKeyType("rr_str_int_region")
+ }
+
+ test("PairRDD.outerJoinGemFireRegion: RDD[K, V] with Region[K, V2], partitioned region", OuterJoinTest) {
+ verifyPairRDDOuterJoinRegionWithSameKeyType("pr_str_int_region")
+ }
+
+ test("PairRDD.outerJoinGemFireRegion: RDD[K, V] with Region[K, V2], partitioned redundant region", OuterJoinTest) {
+ verifyPairRDDOuterJoinRegionWithSameKeyType("pr_r_str_int_region")
+ }
+
+ def verifyPairRDDOuterJoinRegionWithSameKeyType(regionPath: String): Unit = {
+ val entriesMap: JHashMap[String, Int] = new JHashMap()
+ (0 until numObjects).map(i => entriesMap.put("k_" + i, i))
+
+ val connConf: GemFireConnectionConf = GemFireConnectionConf(sc.getConf)
+ val conn = connConf.getConnection
+ val rgn: Region[String, Int] = conn.getRegionProxy(regionPath)
+ rgn.removeAll(rgn.keySetOnServer())
+ rgn.putAll(entriesMap)
+
+ val data = (-5 until 50).map(x => ("k_" + x, x*2))
+ val rdd = sc.parallelize(data)
+
+ val rdd2 = rdd.outerJoinGemfireRegion[String, Int](regionPath /*, connConf*/)
+ val rdd2Content = rdd2.collect()
+
+ val expectedMap = (-5 until 50).map {
+ i => if (i < 0) ((s"k_$i", i * 2), None)
+ else ((s"k_$i", i*2), Some(i))}.toMap
+ // matchMaps[(String, Int), Option[Int]](expectedMap, rdd2Content.toMap)
+ assert(expectedMap == rdd2Content.toMap)
+ }
+
+ // ------------------------------------------------------------------------------------------------------
+ // PairRDD.joinGemfireRegion[K2, V2](regionPath, ((K, V)) => K2, connConf): GemFireJoinRDD[(K, V), K2, V2]
+ // -------------------------------------------------------------------------------------------------------
+
+ test("PairRDD.outerJoinGemFireRegion: RDD[K, V] with Region[K2, V2], replicated region", OuterJoinTest) {
+ verifyPairRDDOuterJoinRegionWithDiffKeyType("rr_str_int_region")
+ }
+
+ test("PairRDD.outerJoinGemFireRegion: RDD[K, V] with Region[K2, V2], partitioned region", OuterJoinTest) {
+ verifyPairRDDOuterJoinRegionWithDiffKeyType("pr_str_int_region")
+ }
+
+ test("PairRDD.outerJoinGemFireRegion: RDD[K, V] with Region[K2, V2], partitioned redundant region", OuterJoinTest) {
+ verifyPairRDDOuterJoinRegionWithDiffKeyType("pr_r_str_int_region")
+ }
+
+ def verifyPairRDDOuterJoinRegionWithDiffKeyType(regionPath: String): Unit = {
+ val entriesMap: JHashMap[String, Int] = new JHashMap()
+ (0 until numObjects).map(i => entriesMap.put("k_" + i, i))
+
+ val connConf: GemFireConnectionConf = GemFireConnectionConf(sc.getConf)
+ val conn = connConf.getConnection
+ val rgn: Region[String, Int] = conn.getRegionProxy(regionPath)
+ rgn.removeAll(rgn.keySetOnServer())
+ rgn.putAll(entriesMap)
+
+ val data = (-5 until 50).map(x => (x, x*2))
+ val rdd = sc.parallelize(data)
+
+ val func :((Int, Int)) => String = pair => s"k_${pair._1}"
+
+ val rdd2 = rdd.outerJoinGemfireRegion[String, Int](regionPath, func, connConf)
+ val rdd2Content = rdd2.collect()
+
+ val expectedMap = (-5 until 50).map {
+ i => if (i < 0) ((i, i * 2), None)
+ else ((i, i*2), Some(i))}.toMap
+ // matchMaps[(Int, Int), Option[Int]](expectedMap, rdd2Content.toMap)
+ assert(expectedMap == rdd2Content.toMap)
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // RDD.joinGemfireRegion[K, V](regionPath, T => K, connConf): GemFireJoinRDD[T, K, V]
+ // --------------------------------------------------------------------------------------------
+
+ test("RDD.joinGemFireRegion: RDD[T] with Region[K, V], replicated region", JoinTest) {
+ verifyRDDJoinRegion("rr_str_int_region")
+ }
+
+ test("RDD.joinGemFireRegion: RDD[T] with Region[K, V], partitioned region", JoinTest) {
+ verifyRDDJoinRegion("pr_str_int_region")
+ }
+
+ test("RDD.joinGemFireRegion: RDD[T] with Region[K, V], partitioned redundant region", JoinTest) {
+ verifyRDDJoinRegion("pr_r_str_int_region")
+ }
+
+ def verifyRDDJoinRegion(regionPath: String): Unit = {
+ val entriesMap: JHashMap[String, Int] = new JHashMap()
+ (0 until numObjects).map(i => entriesMap.put("k_" + i, i))
+
+ val connConf: GemFireConnectionConf = GemFireConnectionConf(sc.getConf)
+ val conn = connConf.getConnection
+ val rgn: Region[String, Int] = conn.getRegionProxy(regionPath)
+ rgn.removeAll(rgn.keySetOnServer())
+ rgn.putAll(entriesMap)
+
+ val data = (-5 until 50).map(x => s"k_$x")
+ val rdd = sc.parallelize(data)
+
+ val rdd2 = rdd.joinGemfireRegion[String, Int](regionPath, x => x, connConf)
+ val rdd2Content = rdd2.collect()
+
+ val expectedMap = (0 until 50).map(i => (s"k_$i", i)).toMap
+ // matchMaps[String, Int](expectedMap, rdd2Content.toMap)
+ assert(expectedMap == rdd2Content.toMap)
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // RDD.outerJoinGemfireRegion[K, V](regionPath, T => K, connConf): GemFireJoinRDD[T, K, V]
+ // --------------------------------------------------------------------------------------------
+
+ test("RDD.outerJoinGemFireRegion: RDD[T] with Region[K, V], replicated region", OnlyTest) {
+ verifyRDDOuterJoinRegion("rr_str_int_region")
+ }
+
+ test("RDD.outerJoinGemFireRegion: RDD[T] with Region[K, V], partitioned region", OnlyTest) {
+ verifyRDDOuterJoinRegion("pr_str_int_region")
+ }
+
+ test("RDD.outerJoinGemFireRegion: RDD[T] with Region[K, V], partitioned redundant region", OnlyTest) {
+ verifyRDDOuterJoinRegion("pr_r_str_int_region")
+ }
+
+ def verifyRDDOuterJoinRegion(regionPath: String): Unit = {
+ val entriesMap: JHashMap[String, Int] = new JHashMap()
+ (0 until numObjects).map(i => entriesMap.put("k_" + i, i))
+
+ val connConf: GemFireConnectionConf = GemFireConnectionConf(sc.getConf)
+ val conn = connConf.getConnection
+ val rgn: Region[String, Int] = conn.getRegionProxy(regionPath)
+ rgn.removeAll(rgn.keySetOnServer())
+ rgn.putAll(entriesMap)
+
+ val data = (-5 until 50).map(x => s"k_$x")
+ val rdd = sc.parallelize(data)
+
+ val rdd2 = rdd.outerJoinGemfireRegion[String, Int](regionPath, x => x /*, connConf */)
+ val rdd2Content = rdd2.collect()
+
+ val expectedMap = (-5 until 50).map {
+ i => if (i < 0) (s"k_$i", None)
+ else (s"k_$i", Some(i))}.toMap
+ // matchMaps[String, Option[Int]](expectedMap, rdd2Content.toMap)
+ assert(expectedMap == rdd2Content.toMap)
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/85d44f04/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/RetrieveRegionIntegrationTest.scala
----------------------------------------------------------------------
diff --git a/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/RetrieveRegionIntegrationTest.scala b/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/RetrieveRegionIntegrationTest.scala
new file mode 100644
index 0000000..e91221c
--- /dev/null
+++ b/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/RetrieveRegionIntegrationTest.scala
@@ -0,0 +1,237 @@
+package ittest.io.pivotal.gemfire.spark.connector
+
+import java.util.Properties
+
+import io.pivotal.gemfire.spark.connector._
+import com.gemstone.gemfire.cache.Region
+import io.pivotal.gemfire.spark.connector.internal.DefaultGemFireConnectionManager
+import ittest.io.pivotal.gemfire.spark.connector.testkit.GemFireCluster
+import ittest.io.pivotal.gemfire.spark.connector.testkit.IOUtils
+import org.apache.spark.{SparkContext, SparkConf}
+import org.scalatest.{Tag, BeforeAndAfterAll, FunSuite, Matchers}
+import java.util.{HashMap => JHashMap}
+
+
+class RetrieveRegionIntegrationTest extends FunSuite with Matchers with BeforeAndAfterAll with GemFireCluster {
+
+ var sc: SparkContext = null
+ val numServers = 4
+ val numObjects = 1000
+
+ override def beforeAll() {
+ // start gemfire cluster, and spark context
+ val settings = new Properties()
+ settings.setProperty("cache-xml-file", "src/it/resources/test-retrieve-regions.xml")
+ settings.setProperty("num-of-servers", numServers.toString)
+ val locatorPort = GemFireCluster.start(settings)
+
+ // start spark context in local mode
+ IOUtils.configTestLog4j("ERROR", "log4j.logger.org.apache.spark" -> "INFO",
+ "log4j.logger.io.pivotal.gemfire.spark.connector" -> "DEBUG")
+ val conf = new SparkConf()
+ .setAppName("RetrieveRegionIntegrationTest")
+ .setMaster("local[2]")
+ .set(GemFireLocatorPropKey, s"localhost[$locatorPort]")
+ sc = new SparkContext(conf)
+ }
+
+ override def afterAll() {
+ // stop connection, spark context, and gemfire cluster
+ DefaultGemFireConnectionManager.closeConnection(GemFireConnectionConf(sc.getConf))
+ sc.stop()
+ GemFireCluster.stop()
+ }
+
+ def executeTest[K,V](regionName:String, numObjects:Int, entriesMap:java.util.Map[K,V]) = {
+ //Populate some data in the region
+ val connConf: GemFireConnectionConf = GemFireConnectionConf(sc.getConf)
+ val conn = connConf.getConnection
+ val rgn: Region[K, V] = conn.getRegionProxy(regionName)
+ rgn.removeAll(rgn.keySetOnServer())
+ rgn.putAll(entriesMap)
+ verifyRetrieveRegion[K,V](regionName, entriesMap)
+ }
+
+ def verifyRetrieveRegion[K,V](regionName:String, entriesMap:java.util.Map[K,V]) = {
+ val rdd = sc.gemfireRegion(regionName)
+ val collectedObjs = rdd.collect()
+ collectedObjs should have length entriesMap.size
+ import scala.collection.JavaConverters._
+ matchMaps[K,V](entriesMap.asScala.toMap, collectedObjs.toMap)
+ }
+
+ def matchMaps[K,V](map1:Map[K,V], map2:Map[K,V]) = {
+ assert(map1.size == map2.size)
+ map1.foreach(e => {
+ assert(map2.contains(e._1))
+ assert (e._2 == map2.get(e._1).get)
+ }
+ )
+ }
+
+ //Retrieve region for Partitioned Region where some nodes are empty (empty iterator)
+ //This test has to run first...the rest of the tests always use the same num objects
+ test("Retrieve Region for PR where some nodes are empty (Empty Iterator)") {
+ val numObjects = numServers - 1
+ val entriesMap:JHashMap[String, Int] = new JHashMap()
+ (0 until numObjects).map(i => entriesMap.put("key_" + i, i))
+ executeTest[String, Int]("rr_str_int_region", numObjects, entriesMap)
+ }
+
+ //Test for retrieving from region containing string key and int value
+ def verifyRetrieveStringStringRegion(regionName:String) = {
+ val entriesMap:JHashMap[String, String] = new JHashMap()
+ (0 until numObjects).map(i => entriesMap.put("key_" + i, "value_" + i))
+ executeTest[String, String](regionName, numObjects, entriesMap)
+ }
+
+ test("Retrieve Region with replicate redundant string string") {
+ verifyRetrieveStringStringRegion("rr_obj_obj_region")
+ }
+
+ test("Retrieve Region with partitioned string string") {
+ verifyRetrieveStringStringRegion("pr_obj_obj_region")
+ }
+
+ test("Retrieve Region with partitioned redundant string string") {
+ verifyRetrieveStringStringRegion("pr_r_obj_obj_region")
+ }
+
+
+ //Test for retrieving from region containing string key and string value
+ def verifyRetrieveStringIntRegion(regionName:String) = {
+ val entriesMap:JHashMap[String, Int] = new JHashMap()
+ (0 until numObjects).map(i => entriesMap.put("key_" + i, i))
+ executeTest[String, Int](regionName, numObjects, entriesMap)
+ }
+
+ test("Retrieve Region with replicate string int region") {
+ verifyRetrieveStringIntRegion("rr_str_int_region")
+ }
+
+ test("Retrieve Region with partitioned string int region") {
+ verifyRetrieveStringIntRegion("pr_str_int_region")
+ }
+
+ test("Retrieve Region with partitioned redundant string int region") {
+ verifyRetrieveStringIntRegion("pr_r_str_int_region")
+ }
+
+ //Tests for retrieving from region containing string key and object value
+ def verifyRetrieveStringObjectRegion(regionName:String) = {
+ val entriesMap:JHashMap[String, Object] = new JHashMap()
+ (0 until numObjects).map(i => entriesMap.put("key_" + i, new Employee("ename" + i, i)))
+ executeTest[String, Object](regionName, numObjects, entriesMap)
+ }
+
+ test("Retrieve Region with replicate string obj") {
+ verifyRetrieveStringObjectRegion("rr_obj_obj_region")
+ }
+
+ test("Retrieve Region with partitioned string obj") {
+ verifyRetrieveStringObjectRegion("pr_obj_obj_region")
+ }
+
+ test("Retrieve Region with partitioned redundant string obj") {
+ verifyRetrieveStringObjectRegion("pr_r_obj_obj_region")
+ }
+
+ //Test for retrieving from region containing string key and map value
+ def verifyRetrieveStringMapRegion(regionName:String) = {
+ val entriesMap:JHashMap[String,JHashMap[String,String]] = new JHashMap()
+ (0 until numObjects).map(i => {
+ val hashMap:JHashMap[String, String] = new JHashMap()
+ hashMap.put("mapKey:" + i, "mapValue:" + i)
+ entriesMap.put("key_" + i, hashMap)
+ })
+ executeTest(regionName, numObjects, entriesMap)
+ }
+
+ test("Retrieve Region with replicate string map region") {
+ verifyRetrieveStringMapRegion("rr_obj_obj_region")
+ }
+
+ test("Retrieve Region with partitioned string map region") {
+ verifyRetrieveStringMapRegion("pr_obj_obj_region")
+ }
+
+ test("Retrieve Region with partitioned redundant string map region") {
+ verifyRetrieveStringMapRegion("pr_r_obj_obj_region")
+ }
+
+ //Test and helpers specific for retrieving from region containing string key and byte[] value
+ def executeTestWithByteArrayValues[K](regionName:String, numObjects:Int, entriesMap:java.util.Map[K,Array[Byte]]) = {
+ //Populate some data in the region
+ val connConf: GemFireConnectionConf = GemFireConnectionConf(sc.getConf)
+ val conn = connConf.getConnection
+ val rgn: Region[K, Array[Byte]] = conn.getRegionProxy(regionName)
+ rgn.putAll(entriesMap)
+ verifyRetrieveRegionWithByteArrayValues[K](regionName, entriesMap)
+ }
+
+ def verifyRetrieveRegionWithByteArrayValues[K](regionName:String, entriesMap:java.util.Map[K,Array[Byte]]) = {
+ val rdd = sc.gemfireRegion(regionName)
+ val collectedObjs = rdd.collect()
+ collectedObjs should have length entriesMap.size
+ import scala.collection.JavaConverters._
+ matchByteArrayMaps[K](entriesMap.asScala.toMap, collectedObjs.toMap)
+ }
+
+ def matchByteArrayMaps[K](map1:Map[K,Array[Byte]], map2:Map[K,Array[Byte]]) = {
+ map1.foreach(e => {
+ assert(map2.contains(e._1))
+ assert (java.util.Arrays.equals(e._2, map2.get(e._1).get))
+ }
+ )
+ assert(map1.size == map2.size)
+
+ }
+
+ def verifyRetrieveStringByteArrayRegion(regionName:String) = {
+ val entriesMap:JHashMap[String, Array[Byte]] = new JHashMap()
+ (0 until numObjects).map(i => entriesMap.put("key_" + i, Array[Byte](192.toByte, 168.toByte, 0, i.toByte)))
+ executeTestWithByteArrayValues[String](regionName, numObjects, entriesMap)
+ }
+
+ test("Retrieve Region with replicate region string byte[] region") {
+ verifyRetrieveStringByteArrayRegion("rr_obj_obj_region")
+ }
+
+ test("Retrieve Region with partition region string byte[] region") {
+ verifyRetrieveStringByteArrayRegion("pr_obj_obj_region")
+ }
+
+ test("Retrieve Region with partition redundant region string byte[] region") {
+ verifyRetrieveStringByteArrayRegion("pr_r_obj_obj_region")
+ }
+
+ test("Retrieve Region with where clause on partitioned redundant region", FilterTest) {
+ verifyRetrieveRegionWithWhereClause("pr_r_str_int_region")
+ }
+
+ test("Retrieve Region with where clause on partitioned region", FilterTest) {
+ verifyRetrieveRegionWithWhereClause("pr_str_int_region")
+ }
+
+ test("Retrieve Region with where clause on replicated region", FilterTest) {
+ verifyRetrieveRegionWithWhereClause("rr_str_int_region")
+ }
+
+ def verifyRetrieveRegionWithWhereClause(regionPath: String): Unit = {
+ val entriesMap: JHashMap[String, Int] = new JHashMap()
+ (0 until numObjects).map(i => entriesMap.put("key_" + i, i))
+
+ val connConf: GemFireConnectionConf = GemFireConnectionConf(sc.getConf)
+ val conn = connConf.getConnection
+ val rgn: Region[String, Int] = conn.getRegionProxy(regionPath)
+ rgn.removeAll(rgn.keySetOnServer())
+ rgn.putAll(entriesMap)
+
+ val rdd = sc.gemfireRegion(regionPath).where("value.intValue() < 50")
+ val expectedMap = (0 until 50).map(i => (s"key_$i", i)).toMap
+ val collectedObjs = rdd.collect()
+ // collectedObjs should have length expectedMap.size
+ matchMaps[String, Int](expectedMap, collectedObjs.toMap)
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/85d44f04/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/package.scala
----------------------------------------------------------------------
diff --git a/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/package.scala b/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/package.scala
new file mode 100644
index 0000000..744e2d1
--- /dev/null
+++ b/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/package.scala
@@ -0,0 +1,13 @@
+package ittest.io.pivotal.gemfire.spark
+
+import org.scalatest.Tag
+
+package object connector {
+
+ object OnlyTest extends Tag("OnlyTest")
+ object FetchDataTest extends Tag("FetchDateTest")
+ object FilterTest extends Tag("FilterTest")
+ object JoinTest extends Tag("JoinTest")
+ object OuterJoinTest extends Tag("OuterJoinTest")
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/85d44f04/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/testkit/GemFireCluster.scala
----------------------------------------------------------------------
diff --git a/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/testkit/GemFireCluster.scala b/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/testkit/GemFireCluster.scala
new file mode 100644
index 0000000..f87bc93
--- /dev/null
+++ b/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/testkit/GemFireCluster.scala
@@ -0,0 +1,31 @@
+package ittest.io.pivotal.gemfire.spark.connector.testkit
+
+import java.util.Properties
+
+trait GemFireCluster {
+ def startGemFireCluster(settings: Properties): Int = {
+ println("=== GemFireCluster start()")
+ GemFireCluster.start(settings)
+ }
+}
+
+object GemFireCluster {
+ private var gemfire: Option[GemFireRunner] = None
+
+ def start(settings: Properties): Int = {
+ gemfire.map(_.stopGemFireCluster()) // Clean up any old running GemFire instances
+ val runner = new GemFireRunner(settings)
+ gemfire = Some(runner)
+ runner.getLocatorPort
+ }
+
+ def stop(): Unit = {
+ println("=== GemFireCluster shutdown: " + gemfire.toString)
+ gemfire match {
+ case None => println("Nothing to shutdown.")
+ case Some(runner) => runner.stopGemFireCluster()
+ }
+ gemfire = None
+ println("=== GemFireCluster shutdown finished.")
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/85d44f04/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/testkit/GemFireRunner.scala
----------------------------------------------------------------------
diff --git a/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/testkit/GemFireRunner.scala b/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/testkit/GemFireRunner.scala
new file mode 100644
index 0000000..1984d6c
--- /dev/null
+++ b/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/testkit/GemFireRunner.scala
@@ -0,0 +1,136 @@
+package ittest.io.pivotal.gemfire.spark.connector.testkit
+
+import java.io.{IOException, File}
+import java.net.InetAddress
+import java.util.Properties
+import org.apache.commons.httpclient.HttpClient
+import org.apache.commons.io.FileUtils
+import org.apache.commons.io.filefilter.IOFileFilter
+
+/**
+* A class that manages GemFire locator and servers. Uses gfsh to
+* start and stop the locator and servers.
+*/
+class GemFireRunner(settings: Properties) {
+
+ private def gemfireHome: String = scala.util.Properties.envOrNone("GEMFIRE") match {
+ case Some(home) => home
+ case None => throw new RuntimeException("No GEMFIRE environment variable set in environment.")
+ }
+
+ val gfshCmd = new File(gemfireHome, "bin/gfsh").toString
+ val cacheXMLFile = settings.get("cache-xml-file")
+ val numServers: Int = settings.get("num-of-servers").asInstanceOf[String].toInt
+ val cwd = new File(".").getAbsolutePath
+ val gemfireFunctionsTargetDir = new File("../gemfire-functions/target")
+ val testroot = "target/testgemfire"
+ val classpath = new File(cwd, "target/scala-2.10/it-classes/")
+ val locatorPort = startGemFireCluster(numServers)
+
+ def getLocatorPort: Int = locatorPort
+
+ private def startGemFireCluster(numServers: Int): Int = {
+ //ports(0) for GemFire locator, the other ports are for GemFire servers
+ val ports: Seq[Int] = IOUtils.getRandomAvailableTCPPorts(2 + numServers)
+ startGemFireLocator(ports(0), ports(1))
+ startGemFireServers(ports(0), ports.drop(2))
+ registerFunctions(ports(1))
+ ports(0)
+ }
+
+ private def startGemFireLocator(locatorPort: Int, jmxHttpPort:Int) {
+ println(s"=== GemFireRunner: starting locator on port $locatorPort")
+ val locatorDir = new File(cwd, s"$testroot/locator")
+ if (locatorDir.exists())
+ FileUtils.deleteDirectory(locatorDir)
+ IOUtils.mkdir(locatorDir)
+ new ProcessBuilder()
+ .command(gfshCmd, "start", "locator",
+ "--name=locator",
+ s"--dir=$locatorDir",
+ s"--port=$locatorPort",
+ s"--J=-Dgemfire.jmx-manager-http-port=$jmxHttpPort")
+ .inheritIO()
+ .start()
+
+ // Wait 30 seconds for locator to start
+ println(s"=== GemFireRunner: waiting for locator on port $locatorPort")
+ if (!IOUtils.waitForPortOpen(InetAddress.getByName("localhost"), locatorPort, 30000))
+ throw new IOException("Failed to start GemFire locator.")
+ println(s"=== GemFireRunner: done waiting for locator on port $locatorPort")
+ }
+
+ private def startGemFireServers(locatorPort: Int, serverPorts: Seq[Int]) {
+ val procs = for (i <- 0 until serverPorts.length) yield {
+ println(s"=== GemFireRunner: starting server${i+1} with clientPort ${serverPorts(i)}")
+ val serverDir = new File(cwd, s"$testroot/server${i+1}")
+ if (serverDir.exists())
+ FileUtils.deleteDirectory(serverDir)
+ IOUtils.mkdir(serverDir)
+ new ProcessBuilder()
+ .command(gfshCmd, "start", "server",
+ s"--name=server${i+1}",
+ s"--locators=localhost[$locatorPort]",
+ s"--bind-address=localhost",
+ s"--server-port=${serverPorts(i)}",
+ s"--dir=$serverDir",
+ s"--cache-xml-file=$cacheXMLFile",
+ s"--classpath=$classpath")
+ .inheritIO()
+ .start()
+ }
+ procs.foreach(p => p.waitFor)
+ println(s"All $serverPorts.length servers have been started")
+ }
+
+ private def registerFunctions(jmxHttpPort:Int) {
+ import scala.collection.JavaConversions._
+ FileUtils.listFiles(gemfireFunctionsTargetDir, fileFilter, dirFilter).foreach{ f => registerFunction(jmxHttpPort, f)}
+ }
+
+ def fileFilter = new IOFileFilter {
+ def accept (file: File) = file.getName.endsWith(".jar") && file.getName.startsWith("gemfire-functions")
+ def accept (dir: File, name: String) = name.endsWith(".jar") && name.startsWith("gemfire-functions")
+ }
+
+ def dirFilter = new IOFileFilter {
+ def accept (file: File) = file.getName.startsWith("scala")
+ def accept (dir: File, name: String) = name.startsWith("scala")
+ }
+
+ private def registerFunction(jmxHttpPort:Int, jar:File) {
+ println("Deploying:" + jar.getName)
+ import io.pivotal.gemfire.spark.connector.GemFireFunctionDeployer
+ val deployer = new GemFireFunctionDeployer(new HttpClient())
+ deployer.deploy("localhost", jmxHttpPort, jar)
+ }
+
+ def stopGemFireCluster(): Unit = {
+ stopGemFireServers(numServers)
+ stopGemFireLocator()
+ if (!IOUtils.waitForPortClose(InetAddress.getByName("localhost"), getLocatorPort, 30000))
+ throw new IOException(s"Failed to stop GemFire locator at port $getLocatorPort.")
+ println(s"Successfully stop GemFire locator at port $getLocatorPort.")
+ }
+
+ private def stopGemFireLocator() {
+ println(s"=== GemFireRunner: stop locator")
+ val p = new ProcessBuilder()
+ .inheritIO()
+ .command(gfshCmd, "stop", "locator", s"--dir=$testroot/locator")
+ .start()
+ p.waitFor()
+ }
+
+ private def stopGemFireServers(numServers: Int) {
+ val procs = for (i <-1 to numServers) yield {
+ println(s"=== GemFireRunner: stop server $i.")
+ new ProcessBuilder()
+ .inheritIO()
+ .command(gfshCmd, "stop", "server", s"--dir=$testroot/server$i")
+ .start()
+ }
+ procs.foreach(p => p.waitFor())
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/85d44f04/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/testkit/IOUtils.scala
----------------------------------------------------------------------
diff --git a/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/testkit/IOUtils.scala b/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/testkit/IOUtils.scala
new file mode 100644
index 0000000..99f9b33
--- /dev/null
+++ b/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/ittest/io/pivotal/gemfire/spark/connector/testkit/IOUtils.scala
@@ -0,0 +1,78 @@
+package ittest.io.pivotal.gemfire.spark.connector.testkit
+
+import java.io.{File, IOException}
+import java.net.{InetAddress, Socket}
+import com.gemstone.gemfire.internal.AvailablePort
+import scala.util.Try
+import org.apache.log4j.PropertyConfigurator
+import java.util.Properties
+
+object IOUtils {
+
+ /** Makes a new directory or throws an `IOException` if it cannot be made */
+ def mkdir(dir: File): File = {
+ if (!dir.mkdirs())
+ throw new IOException(s"Could not create dir $dir")
+ dir
+ }
+
+ private def socketPortProb(host: InetAddress, port: Int) = Iterator.continually {
+ Try {
+ Thread.sleep(100)
+ new Socket(host, port).close()
+ }
+ }
+
+ /**
+ * Waits until a port at the given address is open or timeout passes.
+ * @return true if managed to connect to the port, false if timeout happened first
+ */
+ def waitForPortOpen(host: InetAddress, port: Int, timeout: Long): Boolean = {
+ val startTime = System.currentTimeMillis()
+ socketPortProb(host, port)
+ .dropWhile(p => p.isFailure && System.currentTimeMillis() - startTime < timeout)
+ .next()
+ .isSuccess
+ }
+
+ /**
+ * Waits until a port at the given address is close or timeout passes.
+ * @return true if host:port is un-connect-able, false if timeout happened first
+ */
+ def waitForPortClose(host: InetAddress, port: Int, timeout: Long): Boolean = {
+ val startTime = System.currentTimeMillis()
+ socketPortProb(host, port)
+ .dropWhile(p => p.isSuccess && System.currentTimeMillis() - startTime < timeout)
+ .next()
+ .isFailure
+ }
+
+ /**
+ * Returns array of unique randomly available tcp ports of specified count.
+ */
+ def getRandomAvailableTCPPorts(count: Int): Seq[Int] =
+ (0 until count).map(x => AvailablePort.getRandomAvailablePortKeeper(AvailablePort.SOCKET))
+ .map{x => x.release(); x.getPort}.toArray
+
+ /**
+ * config a log4j properties used for integration tests
+ */
+ def configTestLog4j(level: String, props: (String, String)*): Unit = {
+ val pro = new Properties()
+ props.foreach(p => pro.put(p._1, p._2))
+ configTestLog4j(level, pro)
+ }
+
+ def configTestLog4j(level: String, props: Properties): Unit = {
+ val pro = new Properties()
+ pro.put("log4j.rootLogger", s"$level, console")
+ pro.put("log4j.appender.console", "org.apache.log4j.ConsoleAppender")
+ pro.put("log4j.appender.console.target", "System.err")
+ pro.put("log4j.appender.console.layout", "org.apache.log4j.PatternLayout")
+ pro.put("log4j.appender.console.layout.ConversionPattern",
+ "%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n")
+ pro.putAll(props)
+ PropertyConfigurator.configure(pro)
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/85d44f04/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/org/apache/spark/streaming/ManualClockHelper.scala
----------------------------------------------------------------------
diff --git a/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/org/apache/spark/streaming/ManualClockHelper.scala b/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/org/apache/spark/streaming/ManualClockHelper.scala
new file mode 100644
index 0000000..fbcbffb
--- /dev/null
+++ b/gemfire-spark-connector/gemfire-spark-connector/src/it/scala/org/apache/spark/streaming/ManualClockHelper.scala
@@ -0,0 +1,12 @@
+package org.apache.spark.streaming
+
+import org.apache.spark.util.ManualClock
+
+object ManualClockHelper {
+
+ def addToTime(ssc: StreamingContext, timeToAdd: Long): Unit = {
+ val clock = ssc.scheduler.clock.asInstanceOf[ManualClock]
+ clock.advance(timeToAdd)
+ }
+
+}