Generalizing GraphFactory classes as Abstract classes. Adding ScenarioTest for loading with Triples.

adding javadocs, license, and author tags to newest classes; deleting unused TravelGraphSchema
reverting file changes for MultiRowParallelScanInterpreter.java; moving all new classes to src/test
reverting file changes for MultiRowParallelScanInterpreter.java; moving all new classes to src/test
removing shell script
updating constructor logic and preconditions in Triple; moving HotelTriples to test resources.

Author: Addison Slabaugh

src/main/java/com/amazon/janusgraph/example/MarvelGraphFactory.java

@@ -65,9 +65,9 @@ public class MarvelGraphFactory {
     public static final String WEAPON = "weapon";
     public static final MetricRegistry REGISTRY = MetricManager.INSTANCE.getRegistry();
     public static final ConsoleReporter REPORTER = ConsoleReporter.forRegistry(REGISTRY).build();
-    private static final String TIMER_LINE = "MarvelGraphFactory.line";
-    private static final String TIMER_CREATE = "MarvelGraphFactory.create_";
-    private static final String COUNTER_GET = "MarvelGraphFactory.get_";
+    private static final String TIMER_LINE = "MarvelGraph.line";
+    private static final String TIMER_CREATE = "MarvelGraph.create_";
+    private static final String COUNTER_GET = "MarvelGraph.get_";
     private static final String[] WEAPONS = { "claws", "ring", "shield", "robotic suit", "cards", "surf board", "glider", "gun", "swords", "lasso" };
     private static final AtomicInteger COMPLETED_TASK_COUNT = new AtomicInteger(0);
     private static final int POOL_SIZE = 10;
@@ -164,7 +164,7 @@ public static void load(final JanusGraph graph, final int rowsToLoad, final bool
                 REPORTER.report();
             }
         }
-        LOG.info("MarvelGraphFactory.load complete");
+        LOG.info("MarvelGraph.load complete");
     }
 
     @RequiredArgsConstructor

src/test/java/com/amazon/janusgraph/AbstractGraph.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2014-2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+package com.amazon.janusgraph;
+
+import com.amazon.janusgraph.triple.Triple;
+import org.janusgraph.core.JanusGraph;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.URL;
+import java.util.Set;
+import java.util.concurrent.*;
+
+
+/**
+ *
+ * @author Addison Slabaugh
+ *
+ */
+public interface AbstractGraph {
+
+    int BATCH_SIZE = 10;
+    Logger LOG = LoggerFactory.getLogger(AbstractGraph.class);
+    int POOL_SIZE = 10;
+
+    ClassLoader classLoader = TravelGraph.class.getClassLoader();
+    BlockingQueue<Runnable> creationQueue = new LinkedBlockingQueue<>();
+    ExecutorService executor = Executors.newFixedThreadPool(POOL_SIZE);
+
+    static void load(final JanusGraph graph, boolean report) throws Exception {
+    }
+
+    public abstract Set<Triple> processFile(URL url);
+}

src/test/java/com/amazon/janusgraph/AbstractGraphFactory.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2014-2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+package com.amazon.janusgraph;
+
+
+/**
+ *
+ * @author Addison Slabaugh
+ *
+ */
+public interface AbstractGraphFactory {
+    public AbstractGraph createGraphFactory();
+}

src/test/java/com/amazon/janusgraph/MarvelTest.java

@@ -20,6 +20,7 @@
 import java.util.Collection;
 import java.util.Iterator;
 
+import com.amazon.janusgraph.example.MarvelGraphFactory;
 import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
 import org.apache.tinkerpop.gremlin.structure.Direction;
 import org.apache.tinkerpop.gremlin.structure.Edge;
@@ -35,7 +36,6 @@
 import org.junit.runners.Parameterized;
 
 import com.amazon.janusgraph.diskstorage.dynamodb.BackendDataModel;
-import com.amazon.janusgraph.example.MarvelGraphFactory;
 import com.amazon.janusgraph.graphdb.dynamodb.TestCombination;
 import com.amazon.janusgraph.testcategory.IsolateRemainingTestsCategory;
 import com.codahale.metrics.MetricRegistry;

src/test/java/com/amazon/janusgraph/ScenarioTests.java

@@ -90,6 +90,16 @@ public void lockingTest() throws BackendException, InterruptedException {
         createSchemaAndDemoLockExpiry(USE_STORAGE_NATIVE_LOCKING, USE_EDGESTORE_LOCKING, USE_GRAPHINDEX_LOCKING, 100);
     }
 
+    /**
+     * https://stackoverflow.com/questions/44535054/generalizing-dynamodb-janusgraph-factory-lock-and-schema-problems
+     * @throws Exception
+     */
+    @Test
+    public void tripleIngestTest(StandardJanusGraph graph) throws Exception {
+        TravelGraphFactory graphFactory = new TravelGraphFactory();
+        graphFactory.loadGraphFactory(graph, false);
+    }
+
     @Test
     public void demoNonnativeLockersWithDynamoDB() throws BackendException, InterruptedException {
         createSchemaAndDemoLockExpiry(USE_JANUSGRAPH_LOCKING, USE_EDGESTORE_LOCKING, USE_GRAPHINDEX_LOCKING, 200);

src/test/java/com/amazon/janusgraph/TravelGraph.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2014-2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+package com.amazon.janusgraph;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.net.URL;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import com.amazon.janusgraph.batch.BatchCommand;
+import com.amazon.janusgraph.creator.ObjectCreationCommand;
+import org.janusgraph.core.JanusGraph;
+import org.janusgraph.util.stats.MetricManager;
+import com.codahale.metrics.ConsoleReporter;
+import com.codahale.metrics.MetricRegistry;
+
+import com.amazon.janusgraph.triple.Triple;
+
+/**
+ *
+ * @author Addison Slabaugh
+ *
+ */
+public class TravelGraph implements AbstractGraph {
+
+    public static final MetricRegistry REGISTRY = MetricManager.INSTANCE.getRegistry();
+    public static final ConsoleReporter REPORTER = ConsoleReporter.forRegistry(REGISTRY).build();
+    private static final int POOL_SIZE = 10;
+    private static Set<Triple> triples;
+
+    /**
+     *
+     * @param graph
+     * @param report        allows client to determine whether or not to generate a report after ingestion
+     * @throws Exception
+     */
+    public void load(final JanusGraph graph, boolean report) throws Exception {
+
+        URL resource = classLoader.getResource("META-INF/HotelTriples.txt");
+
+        triples = processFile(resource);
+        for (Triple t : triples) {
+            creationQueue.add(new ObjectCreationCommand(graph, t, REGISTRY, LOG));
+        }
+
+        for (int i = 0; i < POOL_SIZE; i++) {
+            executor.execute(new BatchCommand(graph, creationQueue, LOG, BATCH_SIZE));
+        }
+        executor.shutdown();
+
+        while (!executor.awaitTermination(60, TimeUnit.MILLISECONDS)) {
+            LOG.info("Awaiting:" + creationQueue.size());
+            if(report) {
+                REPORTER.report();
+            }
+        }
+        executor.shutdown();
+
+        LOG.info("TravelGraph.load complete");
+    }
+
+    /**
+     * A method that processes a file containing left and right objects, properties, and relationships and
+     * returns a set of triples.
+     *
+     * @param url   URL that points to a resource containing left and right objects, properties
+     *              and relationships
+     * @return      a set of triples that can be ingested into a graph
+     */
+    @Override
+    public Set<Triple> processFile(URL url) {
+        Set<Triple> triples = new HashSet<>();
+        try {
+            BufferedReader bf = new BufferedReader(new InputStreamReader(url.openStream()));
+            String line = "";
+            while ((line = bf.readLine()) != null) {
+                String[] split = line.split("\t");
+                Triple t = new Triple(split);
+                System.out.println(String.valueOf(t));
+                triples.add(t);
+            }
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+        return triples;
+    }
+}

src/test/java/com/amazon/janusgraph/TravelGraphFactory.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2014-2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+package com.amazon.janusgraph;
+
+import org.janusgraph.graphdb.database.StandardJanusGraph;
+
+/**
+ *
+ * @author Addison Slabaugh
+ *
+ */
+public class TravelGraphFactory implements AbstractGraphFactory {
+    @Override
+    public AbstractGraph createGraphFactory() {
+        TravelGraph graph = new TravelGraph();
+        return graph;
+    }
+
+    /**
+     * This method creates a new TravelGraph instance and loads it. The client can specify whether or not
+     * to generate a report after ingestion.
+     *
+     * @param graph
+     * @param report        indicates whether or not to produce a report after ingestion
+     * @throws Exception
+     */
+    public static void loadGraphFactory(StandardJanusGraph graph, boolean report) throws Exception {
+        TravelGraph g = new TravelGraph();
+        g.load(graph, report);
+    }
+}

src/test/java/com/amazon/janusgraph/batch/BatchCommand.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2014-2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+package com.amazon.janusgraph.batch;
+
+import org.apache.commons.lang.exception.ExceptionUtils;
+import org.janusgraph.core.JanusGraph;
+import org.slf4j.Logger;
+
+import java.util.concurrent.BlockingQueue;
+
+/**
+ *
+ * @author Addison Slabaugh
+ *
+ */
+public class BatchCommand implements Runnable {
+
+    final JanusGraph graph;
+    final BlockingQueue<Runnable> commands;
+    public static Logger LOG;
+    public static int BATCH_SIZE;
+
+    public BatchCommand(JanusGraph graph, BlockingQueue<Runnable> commands, Logger LOG, int BATCH_SIZE) {
+        this.graph = graph;
+        this.commands = commands;
+        this.LOG = LOG;
+        this.BATCH_SIZE = BATCH_SIZE;
+    }
+
+    @Override
+    public void run() {
+        int i = 0;
+        Runnable command = null;
+        while ((command = commands.poll()) != null) {
+            try {
+                command.run();
+            } catch (Throwable e) {
+                Throwable rootCause = ExceptionUtils.getRootCause(e);
+                String rootCauseMessage = null == rootCause ? "" : rootCause.getMessage();
+                LOG.error("Error processing line {} {}", e.getMessage(), rootCauseMessage, e);
+            }
+            if (i++ % BATCH_SIZE == 0) {
+                try {
+                    graph.tx().commit();
+                } catch (Throwable e) {
+                    LOG.error("Error processing commit {} {}", e.getMessage(), ExceptionUtils.getRootCause(e).getMessage());
+                }
+            }
+        }
+
+        try {
+            graph.tx().commit();
+        } catch (Throwable e) {
+            LOG.error("Error processing commit {} {}", e.getMessage(), ExceptionUtils.getRootCause(e).getMessage());
+        }
+    }
+
+}