@Test
    @FeatureRequirementSet(FeatureRequirementSet.Package.SIMPLE)
    public void shouldWriteToMultiplePartitions() {
        final PartitionStrategy partitionStrategyAA = PartitionStrategy.build()
                .partitionKey(partition).writePartition("A").readPartitions("A").create();
        final GraphTraversalSource sourceAA = g.withStrategies(partitionStrategyAA);

        final PartitionStrategy partitionStrategyBA = PartitionStrategy.build()
                .partitionKey(partition).writePartition("B").readPartitions("A").create();
        final GraphTraversalSource sourceBA = g.withStrategies(partitionStrategyBA);

        final PartitionStrategy partitionStrategyCAB = PartitionStrategy.build()
                .partitionKey(partition).writePartition("C").readPartitions("A","B").create();
        final GraphTraversalSource sourceCAB = g.withStrategies(partitionStrategyCAB);

        final PartitionStrategy partitionStrategyC = PartitionStrategy.build()
                .partitionKey(partition).writePartition("C").create();
        final GraphTraversalSource sourceC = g.withStrategies(partitionStrategyC);

        final PartitionStrategy partitionStrategyCA = PartitionStrategy.build()
                .partitionKey(partition).writePartition("C").readPartitions("A").create();
        final GraphTraversalSource sourceCA = g.withStrategies(partitionStrategyCA);

        final GraphTraversalSource sourceCABC = g.withStrategies(PartitionStrategy.create(new MapConfiguration(new HashMap<String, Object>() {{

            put(PartitionStrategy.WRITE_PARTITION, "C");
            put(PartitionStrategy.PARTITION_KEY, partition);
            put(PartitionStrategy.READ_PARTITIONS, Arrays.asList("A", "B", "C"));
        }})));
        final PartitionStrategy partitionStrategyCC = PartitionStrategy.build()
                .partitionKey(partition).writePartition("C").readPartitions("C").create();
        final GraphTraversalSource sourceCC = g.withStrategies(partitionStrategyCC);

        final PartitionStrategy partitionStrategyCBC = PartitionStrategy.build()
                .partitionKey(partition).writePartition("C").readPartitions(Arrays.asList("B", "C")).create();
        final GraphTraversalSource sourceCBC = g.withStrategies(partitionStrategyCBC);

        final Vertex vA = sourceAA.addV().property("any", "a").next();
        final Vertex vAA = sourceAA.addV().property("any", "aa").next();
        final Edge eAtoAA = sourceAA.withSideEffect("vAA", vAA).V(vA.id()).addE("aTOa").to("vAA").next();

        final Vertex vB = sourceBA.addV().property("any", "b").next();
        sourceBA.withSideEffect("vB", vB).V(vA.id()).addE("aTOb").to("vB").next();

        final Vertex vC = sourceCAB.addV().property("any", "c").next();
        final Edge eBtovC = sourceCAB.withSideEffect("vC", vC).V(vB.id()).addE("bTOc").to("vC").next();
        final Edge eAtovC = sourceCAB.withSideEffect("vC", vC).V(vA.id()).addE("aTOc").to("vC").next();

        assertEquals(0, IteratorUtils.count(sourceC.V()));
        assertEquals(0, IteratorUtils.count(sourceC.E()));

        assertEquals(new Long(2), sourceCA.V().count().next());
        assertEquals(new Long(1), sourceCA.E().count().next());
        assertEquals(new Long(1), sourceCA.V(vA.id()).outE().count().next());
        assertEquals(eAtoAA.id(), sourceCA.V(vA.id()).outE().next().id());
        assertEquals(new Long(1), sourceCA.V(vA.id()).out().count().next());
        assertEquals(vAA.id(), sourceCA.V(vA.id()).out().next().id());

        final Vertex vA1 = sourceCA.V(vA.id()).next();
        assertEquals(new Long(1), sourceCA.V(vA1).outE().count().next());
        assertEquals(eAtoAA.id(), sourceCA.V(vA1).outE().next().id());
        assertEquals(new Long(1), sourceCA.V(vA1).out().count().next());
        assertEquals(vAA.id(), sourceCA.V(vA1).out().next().id());

        assertEquals(new Long(3), sourceCAB.V().count().next());
        assertEquals(new Long(2), sourceCAB.E().count().next());

        assertEquals(new Long(4), sourceCABC.V().count().next());
        assertEquals(new Long(4), sourceCABC.E().count().next());

        assertEquals(1, IteratorUtils.count(sourceCC.V()));
        // two edges are in the "C" partition, but one each of their incident vertices are not
        assertEquals(2, IteratorUtils.count(sourceCC.E()));

        // two edges are in the "C" partition, but one each of their incident vertices are not
        assertEquals(new Long(2), sourceCC.V(vC.id()).inE().count().next());
        assertEquals(new Long(0), sourceCC.V(vC.id()).in().count().next());

        // excluded vertices; vA is not in {B,C}
        assertEquals(new Long(2), sourceCBC.V(vC.id()).inE().count().next());
        assertEquals(new Long(1), sourceCBC.V(vC.id()).in().count().next());
        assertEquals(vC.id(), sourceCBC.E(eBtovC.id()).inV().id().next());
        assertEquals(vB.id(), sourceCBC.E(eBtovC.id()).outV().id().next());
        assertEquals(vC.id(), sourceCBC.E(eAtovC.id()).inV().id().next());
        assertFalse(sourceCBC.E(eAtovC.id()).outV().hasNext());
    }
@author Stephen Mallette (http://stephen.genoprime.com)
gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/strategy/decoration/PartitionStrategyProcessTest.java