@Override
    public Map<String, Object> getBaseConfiguration(final String graphName, final Class<?> test, final String testMethodName, final LoadGraphWith.GraphData loadGraphWith) {
        this.graphSONInput = RANDOM.nextBoolean();
        if (this.getClass().equals(SparkHadoopGraphProvider.class) && !SparkHadoopGraphProvider.class.getCanonicalName().equals(System.getProperty(PREVIOUS_SPARK_PROVIDER, null))) {
            Spark.close();
            HadoopPools.close();
            KryoShimServiceLoader.close();
            System.setProperty(PREVIOUS_SPARK_PROVIDER, SparkHadoopGraphProvider.class.getCanonicalName());
        }

        final Map<String,Object> config = new HashMap<String, Object>() {{
            put(Graph.GRAPH, HadoopGraph.class.getName());
            put(Constants.GREMLIN_HADOOP_GRAPH_READER, graphSONInput ? GraphSONInputFormat.class.getCanonicalName() : GryoInputFormat.class.getCanonicalName());
            put(Constants.GREMLIN_HADOOP_GRAPH_WRITER, GryoOutputFormat.class.getCanonicalName());
            // clear the output location if it is not empty
            String outputLocation = getWorkingDirectory();
            FileSystemStorage fileSystemStorage = FileSystemStorage.open();
            if (fileSystemStorage.ls(outputLocation).size() > 0) {
                fileSystemStorage.rm(outputLocation);
                outputLocation = getWorkingDirectory();
            }
            put(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, outputLocation);
            put(Constants.GREMLIN_HADOOP_JARS_IN_DISTRIBUTED_CACHE, false);

            put(Constants.GREMLIN_SPARK_PERSIST_CONTEXT, true);  // this makes the test suite go really fast
        }};

        // toy graph inputRDD does not have corresponding outputRDD so where jobs chain, it fails (failing makes sense)
        if (null != loadGraphWith &&
                !test.equals(ProgramTest.Traversals.class) &&
                !test.equals(PageRankTest.Traversals.class) &&
                !test.equals(ConnectedComponentTest.Traversals.class) &&
                !test.equals(ShortestPathTest.Traversals.class) &&
                !test.equals(PeerPressureTest.Traversals.class) &&
                !test.equals(FileSystemStorageCheck.class) &&
                !testMethodName.equals("shouldSupportJobChaining") &&  // GraphComputerTest.shouldSupportJobChaining
                RANDOM.nextBoolean()) {
            config.put(Constants.GREMLIN_HADOOP_GRAPH_READER, ToyGraphInputRDD.class.getCanonicalName());
        }

        // tests persisted RDDs
        if (test.equals(SparkContextStorageCheck.class)) {
            config.put(Constants.GREMLIN_HADOOP_GRAPH_READER, ToyGraphInputRDD.class.getCanonicalName());
            config.put(Constants.GREMLIN_HADOOP_GRAPH_WRITER, PersistedOutputRDD.class.getCanonicalName());
        }

        config.put(Constants.GREMLIN_HADOOP_DEFAULT_GRAPH_COMPUTER, SparkGraphComputer.class.getCanonicalName());
        config.put(SparkLauncher.SPARK_MASTER, "local[" + AVAILABLE_PROCESSORS + "]");
        config.put(Constants.SPARK_SERIALIZER, KryoSerializer.class.getCanonicalName());
        config.put(Constants.SPARK_KRYO_REGISTRATOR, GryoRegistrator.class.getCanonicalName());
        config.put(Constants.SPARK_KRYO_REGISTRATION_REQUIRED, true);
        return config;
    }
@author Marko A. Rodriguez (http://markorodriguez.com)
@author Stephen Mallette (http://stephen.genoprime.com)
spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkHadoopGraphProvider.java