You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by to...@apache.org on 2014/08/05 00:15:01 UTC
[08/11] git commit: Tests passing, need to create more tests
Tests passing, need to create more tests
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/f3b7ead7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/f3b7ead7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/f3b7ead7
Branch: refs/heads/USERGRID-188
Commit: f3b7ead7736385f85b214c87807e995de9275eae
Parents: 69a3faa
Author: Todd Nine <tn...@apigee.com>
Authored: Thu Jul 17 16:54:56 2014 -0600
Committer: Todd Nine <tn...@apigee.com>
Committed: Thu Jul 17 16:54:56 2014 -0600
----------------------------------------------------------------------
.../shard/impl/NodeShardAllocationImpl.java | 102 ++-
.../impl/shard/NodeShardAllocationTest.java | 830 ++++++++++---------
2 files changed, 522 insertions(+), 410 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f3b7ead7/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/NodeShardAllocationImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/NodeShardAllocationImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/NodeShardAllocationImpl.java
index 20cb921..b808f2f 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/NodeShardAllocationImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/NodeShardAllocationImpl.java
@@ -26,18 +26,28 @@ import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import org.apache.commons.collections4.iterators.PushbackIterator;
import org.apache.usergrid.persistence.core.consistency.TimeService;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
+import org.apache.usergrid.persistence.graph.Edge;
import org.apache.usergrid.persistence.graph.GraphFig;
+import org.apache.usergrid.persistence.graph.MarkedEdge;
import org.apache.usergrid.persistence.graph.exception.GraphRuntimeException;
+import org.apache.usergrid.persistence.graph.impl.SimpleSearchByEdgeType;
+import org.apache.usergrid.persistence.graph.impl.SimpleSearchByIdType;
import org.apache.usergrid.persistence.graph.serialization.EdgeSerialization;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.EdgeColumnFamilies;
import org.apache.usergrid.persistence.graph.serialization.impl.shard.EdgeShardSerialization;
import org.apache.usergrid.persistence.graph.serialization.impl.shard.NodeShardAllocation;
import org.apache.usergrid.persistence.graph.serialization.impl.shard.NodeShardApproximation;
import org.apache.usergrid.persistence.graph.serialization.impl.shard.NodeType;
import org.apache.usergrid.persistence.graph.serialization.impl.shard.Shard;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.ShardEntries;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.ShardedEdgeSerialization;
import org.apache.usergrid.persistence.model.entity.Id;
import com.google.common.base.Optional;
@@ -53,11 +63,13 @@ import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
public class NodeShardAllocationImpl implements NodeShardAllocation {
+ private static final Logger LOG = LoggerFactory.getLogger(NodeShardAllocationImpl.class);
+
private static final MinShardTimeComparator MIN_SHARD_TIME_COMPARATOR = new MinShardTimeComparator();
private final EdgeShardSerialization edgeShardSerialization;
- private final EdgeSerialization edgeSerialization;
- // private final NodeShardCounterSerialization edgeShardCounterSerialization;
+ private final EdgeColumnFamilies edgeColumnFamilies;
+ private final ShardedEdgeSerialization shardedEdgeSerialization;
private final NodeShardApproximation nodeShardApproximation;
private final TimeService timeService;
private final GraphFig graphFig;
@@ -66,11 +78,13 @@ public class NodeShardAllocationImpl implements NodeShardAllocation {
@Inject
public NodeShardAllocationImpl( final EdgeShardSerialization edgeShardSerialization,
- final EdgeSerialization edgeSerialization,
+ final EdgeColumnFamilies edgeColumnFamilies,
+ final ShardedEdgeSerialization shardedEdgeSerialization,
final NodeShardApproximation nodeShardApproximation, final TimeService timeService,
final GraphFig graphFig, final Keyspace keyspace ) {
this.edgeShardSerialization = edgeShardSerialization;
- this.edgeSerialization = edgeSerialization;
+ this.edgeColumnFamilies = edgeColumnFamilies;
+ this.shardedEdgeSerialization = shardedEdgeSerialization;
this.nodeShardApproximation = nodeShardApproximation;
this.timeService = timeService;
this.graphFig = graphFig;
@@ -159,7 +173,8 @@ public class NodeShardAllocationImpl implements NodeShardAllocation {
public boolean auditMaxShard( final ApplicationScope scope, final Id nodeId, final NodeType nodeType,
final String... edgeType ) {
- final Iterator<Shard> maxShards = getShards( scope, nodeId, nodeType, Optional.<Shard>absent(), edgeType );
+ final Iterator<Shard> maxShards = edgeShardSerialization.getEdgeMetaData( scope, nodeId, nodeType, Optional.<Shard>absent(), edgeType );
+
//if the first shard has already been allocated, do nothing.
@@ -171,11 +186,20 @@ public class NodeShardAllocationImpl implements NodeShardAllocation {
final Shard maxShard = maxShards.next();
+
+
/**
- * Check out if we have a count for our shard allocation
+ * Nothing to do, it's been created very recently, we don't create a new one
*/
+ if(maxShard.getCreatedTime() >= getMinTime()){
+ return false;
+ }
+ /**
+ * Check out if we have a count for our shard allocation
+ */
+
final long count =
nodeShardApproximation.getCount( scope, nodeId, nodeType, maxShard.getShardIndex(), edgeType );
@@ -185,23 +209,63 @@ public class NodeShardAllocationImpl implements NodeShardAllocation {
/**
- * TODO, use the EdgeShardStrategy and ShardEdgeSerialization to audit this shard
+ * Allocate the shard
+ */
+
+ Iterator<MarkedEdge> edges;
+
+ /**
+ * This is fugly, I think our allocation interface needs to get more declarative
*/
+ if(nodeType == NodeType.SOURCE){
+
+ if(edgeType.length == 1){
+ edges = shardedEdgeSerialization.getEdgesFromSource(edgeColumnFamilies, scope, new SimpleSearchByEdgeType( nodeId, edgeType[0], Long.MAX_VALUE, null), Collections.singleton( new ShardEntries(Collections.singleton( maxShard ))).iterator()) ;
+ }
- //get the max edge, in this shard, and write it.
+ else if(edgeType.length == 2){
+ edges = shardedEdgeSerialization.getEdgesFromSourceByTargetType(edgeColumnFamilies, scope, new SimpleSearchByIdType( nodeId, edgeType[0], Long.MAX_VALUE, edgeType[1], null), Collections.singleton( new ShardEntries(Collections.singleton( maxShard ))).iterator());
+ }
+ else{
+ throw new UnsupportedOperationException( "More than 2 edge types aren't supported" );
+ }
- //try to get a lock here, and fail if one isn't present
+ } else{
+
+ if(edgeType.length == 1){
+ edges = shardedEdgeSerialization.getEdgesToTarget(edgeColumnFamilies, scope, new SimpleSearchByEdgeType( nodeId, edgeType[0], Long.MAX_VALUE, null), Collections.singleton( new ShardEntries(Collections.singleton( maxShard ))).iterator()) ;
+ }
+
+ else if(edgeType.length == 2){
+ edges = shardedEdgeSerialization.getEdgesToTargetBySourceType(edgeColumnFamilies, scope, new SimpleSearchByIdType( nodeId, edgeType[0], Long.MAX_VALUE, edgeType[1], null), Collections.singleton( new ShardEntries(Collections.singleton( maxShard ))).iterator());
+ }
+
+ else{
+ throw new UnsupportedOperationException( "More than 2 edge types aren't supported" );
+ }
+ }
+
+
+ if(!edges.hasNext()){
+ LOG.warn( "Tried to allocate a new shard for node id {} with edge types {}, but no max value could be found in that row", nodeId, edgeType );
+ return false;
+ }
+
+ //we have a next, allocate it based on the max
+
+ MarkedEdge marked = edges.next();
+
+ final long createTimestamp = timeService.getCurrentTime();
+
+
+ try {
+ this.edgeShardSerialization.writeEdgeMeta( scope, nodeId, nodeType, marked.getTimestamp(), createTimestamp, edgeType ).execute();
+ }
+ catch ( ConnectionException e ) {
+ throw new GraphRuntimeException( "Unable to write the new edge metadata" );
+ }
- // final long newShardTime = timeService.getCurrentTime() + graphFig.getShardCacheTimeout() * 2;
- //
- //
- // try {
- // this.edgeShardSerialization.writeEdgeMeta( scope, nodeId, newShardTime, edgeType ).execute();
- // }
- // catch ( ConnectionException e ) {
- // throw new GraphRuntimeException( "Unable to write the new edge metadata" );
- // }
return true;
@@ -230,4 +294,6 @@ public class NodeShardAllocationImpl implements NodeShardAllocation {
return result;
}
}
+
+
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f3b7ead7/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/NodeShardAllocationTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/NodeShardAllocationTest.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/NodeShardAllocationTest.java
index e582455..42b3dc2 100644
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/NodeShardAllocationTest.java
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/NodeShardAllocationTest.java
@@ -31,6 +31,9 @@ import org.mockito.ArgumentCaptor;
import org.apache.usergrid.persistence.core.consistency.TimeService;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.graph.GraphFig;
+import org.apache.usergrid.persistence.graph.MarkedEdge;
+import org.apache.usergrid.persistence.graph.SearchByIdType;
+import org.apache.usergrid.persistence.graph.impl.SimpleMarkedEdge;
import org.apache.usergrid.persistence.graph.serialization.impl.shard.impl.NodeShardAllocationImpl;
import org.apache.usergrid.persistence.model.entity.Id;
import org.apache.usergrid.persistence.model.util.UUIDGenerator;
@@ -76,397 +79,440 @@ public class NodeShardAllocationTest {
when( graphFig.getShardSize() ).thenReturn( 20000l );
when( graphFig.getShardCacheTimeout()).thenReturn( 30000l );
}
-//
-//
-//
-// @Test
-// public void minTime() {
-// final EdgeShardSerialization edgeShardSerialization = mock( EdgeShardSerialization.class );
-//
-// final NodeShardApproximation nodeShardCounterSerialization =
-// mock( NodeShardApproximation.class );
-//
-//
-// final TimeService timeService = mock( TimeService.class );
-//
-// final Keyspace keyspace = mock( Keyspace.class );
-//
-//
-// NodeShardAllocation approximation =
-// new NodeShardAllocationImpl( edgeShardSerialization, edgeSerialization, nodeShardCounterSerialization, timeService,
-// graphFig, keyspace );
-//
-//
-// final long timeservicetime = System.currentTimeMillis();
-//
-// when( timeService.getCurrentTime() ).thenReturn( timeservicetime );
-//
-// final long expected = timeservicetime - 2 * graphFig.getShardCacheTimeout();
-//
-// final long returned = approximation.getMinTime();
-//
-// assertEquals("Correct time was returned", expected, returned);
-// }
-//
-//
-// @Test
-// public void noShards() {
-// final EdgeShardSerialization edgeShardSerialization = mock( EdgeShardSerialization.class );
-//
-// final NodeShardApproximation nodeShardCounterSerialization =
-// mock( NodeShardApproximation.class );
-//
-//
-// final TimeService timeService = mock( TimeService.class );
-//
-// final Keyspace keyspace = mock( Keyspace.class );
-//
-// final MutationBatch batch = mock( MutationBatch.class );
-//
-// when( keyspace.prepareMutationBatch() ).thenReturn( batch );
-//
-// NodeShardAllocation approximation =
-// new NodeShardAllocationImpl( edgeShardSerialization, edgeSerialization, nodeShardCounterSerialization, timeService,
-// graphFig, keyspace );
-//
-// final Id nodeId = createId( "test" );
-// final String type = "type";
-// final String subType = "subType";
-//
-// /**
-// * Mock up returning an empty iterator, our audit shouldn't create a new shard
-// */
-// when( edgeShardSerialization
-// .getEdgeMetaData( same( scope ), same( nodeId ), eq( NodeType.SOURCE ), any( Optional.class ), same( type ),
-// same( subType ) ) ).thenReturn( Collections.<Shard>emptyList().iterator() );
-//
-// final boolean result = approximation.auditMaxShard( scope, nodeId, NodeType.SOURCE, type, subType );
-//
-// assertFalse( "No shard allocated", result );
-// }
-//
-//
-//
-//
-//
-// @Test
-// public void existingFutureShardSameTime() {
-// final EdgeShardSerialization edgeShardSerialization = mock( EdgeShardSerialization.class );
-//
-// final NodeShardApproximation nodeShardCounterSerialization =
-// mock( NodeShardApproximation.class );
-//
-//
-// final TimeService timeService = mock( TimeService.class );
-//
-// final Keyspace keyspace = mock( Keyspace.class );
-//
-//
-// final MutationBatch batch = mock( MutationBatch.class );
-//
-// when( keyspace.prepareMutationBatch() ).thenReturn( batch );
-//
-//
-// NodeShardAllocation approximation =
-// new NodeShardAllocationImpl( edgeShardSerialization, edgeSerialization, nodeShardCounterSerialization, timeService,
-// graphFig, keyspace );
-//
-// final Id nodeId = createId( "test" );
-// final String type = "type";
-// final String subType = "subType";
-//
-//
-// final long timeservicetime = System.currentTimeMillis();
-//
-// when( timeService.getCurrentTime() ).thenReturn( timeservicetime );
-//
-// final Shard futureShard = new Shard(10000l, timeservicetime) ;
-//
-// /**
-// * Mock up returning a min shard, and a future shard
-// */
-// when( edgeShardSerialization
-// .getEdgeMetaData( same( scope ), same( nodeId ), eq( NodeType.TARGET), any( Optional.class ), same( type ),
-// same( subType ) ) ).thenReturn( Arrays.asList( futureShard ).iterator() );
-//
-// final boolean result = approximation.auditMaxShard( scope, nodeId, NodeType.TARGET, type, subType );
-//
-// assertFalse( "No shard allocated", result );
-// }
-//
-//
-// @Test
-// public void lowCountFutureShard() {
-// final EdgeShardSerialization edgeShardSerialization = mock( EdgeShardSerialization.class );
-//
-// final NodeShardApproximation nodeShardApproximation =
-// mock( NodeShardApproximation.class );
-//
-//
-// final TimeService timeService = mock( TimeService.class );
-//
-// final Keyspace keyspace = mock( Keyspace.class );
-//
-// final MutationBatch batch = mock( MutationBatch.class );
-//
-// when( keyspace.prepareMutationBatch() ).thenReturn( batch );
-//
-//
-// NodeShardAllocation approximation =
-// new NodeShardAllocationImpl( edgeShardSerialization, edgeSerialization, nodeShardApproximation, timeService,
-// graphFig, keyspace );
-//
-// final Id nodeId = createId( "test" );
-// final String type = "type";
-// final String subType = "subType";
-//
-//
-// final long timeservicetime = System.currentTimeMillis();
-//
-// when( timeService.getCurrentTime() ).thenReturn( timeservicetime );
-//
-//
-// /**
-// * Mock up returning a min shard, and a future shard
-// */
-// when( edgeShardSerialization
-// .getEdgeMetaData( same( scope ), same( nodeId ), eq(NodeType.TARGET), any( Optional.class ), same( type ),
-// same( subType ) ) ).thenReturn( Arrays.asList( new Shard(0l, 0l) ).iterator() );
-//
-//
-// //return a shard size < our max by 1
-//
-// final long count = graphFig.getShardSize() - 1;
-//
-// when( nodeShardApproximation.getCount(scope, nodeId, NodeType.TARGET, 0l, type, subType )).thenReturn( count );
-//
-// final boolean result = approximation.auditMaxShard( scope, nodeId, NodeType.TARGET, type, subType );
-//
-// assertFalse( "Shard allocated", result );
-// }
-//
-//
-// @Test
-// public void equalCountFutureShard() {
-// final EdgeShardSerialization edgeShardSerialization = mock( EdgeShardSerialization.class );
-//
-// final NodeShardApproximation nodeShardApproximation =
-// mock( NodeShardApproximation.class );
-//
-//
-// final TimeService timeService = mock( TimeService.class );
-//
-// final Keyspace keyspace = mock( Keyspace.class );
-//
-// final MutationBatch batch = mock(MutationBatch.class);
-//
-// when(keyspace.prepareMutationBatch()).thenReturn( batch );
-//
-//
-// NodeShardAllocation approximation =
-// new NodeShardAllocationImpl( edgeShardSerialization, edgeSerialization, nodeShardApproximation, timeService,
-// graphFig, keyspace );
-//
-// final Id nodeId = createId( "test" );
-// final String type = "type";
-// final String subType = "subType";
-//
-//
-// final long timeservicetime = System.currentTimeMillis();
-//
-// when( timeService.getCurrentTime() ).thenReturn( timeservicetime );
-//
-//
-// /**
-// * Mock up returning a min shard
-// */
-// when( edgeShardSerialization
-// .getEdgeMetaData( same( scope ), same( nodeId ), eq(NodeType.SOURCE), any( Optional.class ), same( type ),
-// same( subType ) ) ).thenReturn( Arrays.asList( new Shard( 0l, 0l ) ).iterator() );
-//
-//
-// final long shardCount = graphFig.getShardSize();
-//
-// //return a shard size equal to our max
-// when( nodeShardApproximation
-// .getCount( scope , nodeId, NodeType.SOURCE, 0l,type , subType ))
-// .thenReturn( shardCount );
-//
-// ArgumentCaptor<Long> shardValue = ArgumentCaptor.forClass( Long.class );
-// ArgumentCaptor<Long> timestampValue = ArgumentCaptor.forClass( Long.class );
-//
-//
-// //mock up our mutation
-// when( edgeShardSerialization
-// .writeEdgeMeta( same( scope ), same( nodeId ), eq(NodeType.SOURCE), shardValue.capture(), timestampValue.capture(), same( type ), same( subType ) ) )
-// .thenReturn( mock( MutationBatch.class ) );
-//
-//
-// final boolean result = approximation.auditMaxShard( scope, nodeId, NodeType.SOURCE, type, subType );
-//
-// assertTrue( "Shard allocated", result );
-//
-// //check our new allocated UUID
-//
-//
-// final long savedTimestamp = timestampValue.getValue();
-//
-//
-//
-//
-//
-// assertEquals( "Expected time service time", timeservicetime, savedTimestamp );
-//
-// //now check our max value was set
-// }
-//
-//
-//
-//
-// @Test
-// public void futureCountShardCleanup() {
-// final EdgeShardSerialization edgeShardSerialization = mock( EdgeShardSerialization.class );
-//
-// final NodeShardApproximation nodeShardApproximation =
-// mock( NodeShardApproximation.class );
-//
-//
-// final TimeService timeService = mock( TimeService.class );
-//
-// final Keyspace keyspace = mock( Keyspace.class );
-//
-// final MutationBatch batch = mock(MutationBatch.class);
-//
-// when(keyspace.prepareMutationBatch()).thenReturn( batch );
-//
-//
-// NodeShardAllocation approximation =
-// new NodeShardAllocationImpl( edgeShardSerialization, edgeSerialization, nodeShardApproximation, timeService,
-// graphFig, keyspace );
-//
-// final Id nodeId = createId( "test" );
-// final String type = "type";
-// final String subType = "subType";
-//
-//
-// /**
-// * Use the time service to generate UUIDS
-// */
-// final long timeservicetime = System.currentTimeMillis();
-//
-//
-// when( timeService.getCurrentTime() ).thenReturn( timeservicetime );
-//
-// assertTrue("Shard cache mocked", graphFig.getShardCacheTimeout() > 0);
-//
-//
-// /**
-// * Simulates clock drift when 2 nodes create future shards near one another
-// */
-// final long futureTime = timeService.getCurrentTime() + 2 * graphFig.getShardCacheTimeout();
-//
-//
-// final Shard minShard = new Shard(0l, 0l);
-//
-// /**
-// * Simulate slow node
-// */
-//
-// //our second shard is the "oldest", and hence should be returned in the iterator. Future shard 1 and 3 should be removed
-// final Shard futureShard1 = new Shard(futureTime - 1, timeservicetime+1000);
-//
-// final Shard futureShard2 = new Shard(futureTime + 10000, timeservicetime);
-//
-// final Shard futureShard3 = new Shard(futureShard2.getShardIndex() + 10000, timeservicetime+2000);
-//
-// /**
-// * Mock up returning a min shard
-// */
-// when( edgeShardSerialization
-// .getEdgeMetaData( same( scope ), same( nodeId ), eq(NodeType.TARGET), any( Optional.class ), same( type ),
-// same( subType ) ) ).thenReturn(
-// Arrays.asList( futureShard3, futureShard2, futureShard1, minShard ).iterator() );
-//
-//
-//
-// ArgumentCaptor<Long> newLongValue = ArgumentCaptor.forClass( Long.class );
-//
-//
-//
-//
-// //mock up our mutation
-// when( edgeShardSerialization
-// .removeEdgeMeta( same( scope ), same( nodeId ), eq(NodeType.TARGET), newLongValue.capture(), same( type ), same( subType ) ) )
-// .thenReturn( mock( MutationBatch.class ) );
-//
-//
-// final Iterator<Shard>
-// result = approximation.getShards( scope, nodeId, NodeType.TARGET, Optional.<Shard>absent(), type, subType );
-//
-//
-// assertTrue( "Shards present", result.hasNext() );
-//
-// assertEquals("Only single next shard returned", futureShard2, result.next());
-//
-// assertTrue("Shards present", result.hasNext());
-//
-// assertEquals("Previous shard present", 0l, result.next().getShardIndex());
-//
-// assertFalse("No shards left", result.hasNext());
-//
-// /**
-// * Now we need to verify that both our mutations have been added
-// */
-//
-// List<Long> values = newLongValue.getAllValues();
-//
-// assertEquals("2 values removed", 2, values.size());
-//
-// assertEquals("Deleted Max Future", futureShard1.getShardIndex(), values.get( 0 ).longValue());
-// assertEquals("Deleted Next Future", futureShard3.getShardIndex(), values.get( 1 ).longValue());
-//
-// }
-//
-//
-//
-//
-// @Test
-// public void noShardsReturns() {
-// final EdgeShardSerialization edgeShardSerialization = mock( EdgeShardSerialization.class );
-//
-// final NodeShardApproximation nodeShardApproximation =
-// mock( NodeShardApproximation.class );
-//
-//
-// final TimeService timeService = mock( TimeService.class );
-//
-// final Keyspace keyspace = mock( Keyspace.class );
-//
-// final MutationBatch batch = mock( MutationBatch.class );
-//
-// when( keyspace.prepareMutationBatch() ).thenReturn( batch );
-//
-// NodeShardAllocation approximation =
-// new NodeShardAllocationImpl( edgeShardSerialization, edgeSerialization, nodeShardApproximation, timeService,
-// graphFig, keyspace );
-//
-// final Id nodeId = createId( "test" );
-// final String type = "type";
-// final String subType = "subType";
-//
-// /**
-// * Mock up returning an empty iterator, our audit shouldn't create a new shard
-// */
-// when( edgeShardSerialization
-// .getEdgeMetaData( same( scope ), same( nodeId ), eq(NodeType.TARGET), any( Optional.class ), same( type ),
-// same( subType ) ) ).thenReturn( Collections.<Shard>emptyList().iterator() );
-//
-// final Iterator<Shard> result = approximation.getShards( scope, nodeId, NodeType.TARGET, Optional.<Shard>absent(), type,
-// subType );
-//
-// assertEquals("0 shard allocated", 0l, result.next().getShardIndex());
-//
-// assertFalse( "No shard allocated", result.hasNext() );
-// }
+
+
+
+ @Test
+ public void minTime() {
+ final EdgeShardSerialization edgeShardSerialization = mock( EdgeShardSerialization.class );
+
+ final EdgeColumnFamilies edgeColumnFamilies = mock(EdgeColumnFamilies.class);
+
+ final ShardedEdgeSerialization shardedEdgeSerialization = mock(ShardedEdgeSerialization.class);
+
+ final NodeShardApproximation nodeShardCounterSerialization =
+ mock( NodeShardApproximation.class );
+
+
+ final TimeService timeService = mock( TimeService.class );
+
+ final Keyspace keyspace = mock( Keyspace.class );
+
+
+ NodeShardAllocation approximation =
+ new NodeShardAllocationImpl( edgeShardSerialization, edgeColumnFamilies, shardedEdgeSerialization, nodeShardCounterSerialization, timeService,
+ graphFig, keyspace );
+
+
+ final long timeservicetime = System.currentTimeMillis();
+
+ when( timeService.getCurrentTime() ).thenReturn( timeservicetime );
+
+ final long expected = timeservicetime - 2 * graphFig.getShardCacheTimeout();
+
+ final long returned = approximation.getMinTime();
+
+ assertEquals("Correct time was returned", expected, returned);
+ }
+
+
+ @Test
+ public void noShards() {
+ final EdgeShardSerialization edgeShardSerialization = mock( EdgeShardSerialization.class );
+
+ final EdgeColumnFamilies edgeColumnFamilies = mock(EdgeColumnFamilies.class);
+
+ final ShardedEdgeSerialization shardedEdgeSerialization = mock(ShardedEdgeSerialization.class);
+
+
+ final NodeShardApproximation nodeShardCounterSerialization =
+ mock( NodeShardApproximation.class );
+
+
+ final TimeService timeService = mock( TimeService.class );
+
+ final Keyspace keyspace = mock( Keyspace.class );
+
+ final MutationBatch batch = mock( MutationBatch.class );
+
+ when( keyspace.prepareMutationBatch() ).thenReturn( batch );
+
+ NodeShardAllocation approximation =
+ new NodeShardAllocationImpl( edgeShardSerialization, edgeColumnFamilies, shardedEdgeSerialization, nodeShardCounterSerialization, timeService,
+ graphFig, keyspace );
+
+ final Id nodeId = createId( "test" );
+ final String type = "type";
+ final String subType = "subType";
+
+ /**
+ * Mock up returning an empty iterator, our audit shouldn't create a new shard
+ */
+ when( edgeShardSerialization
+ .getEdgeMetaData( same( scope ), same( nodeId ), eq( NodeType.SOURCE ), any( Optional.class ), same( type ),
+ same( subType ) ) ).thenReturn( Collections.<Shard>emptyList().iterator() );
+
+ final boolean result = approximation.auditMaxShard( scope, nodeId, NodeType.SOURCE, type, subType );
+
+ assertFalse( "No shard allocated", result );
+ }
+
+
+
+
+
+ @Test
+ public void existingFutureShardSameTime() {
+ final EdgeShardSerialization edgeShardSerialization = mock( EdgeShardSerialization.class );
+
+ final EdgeColumnFamilies edgeColumnFamilies = mock(EdgeColumnFamilies.class);
+
+ final ShardedEdgeSerialization shardedEdgeSerialization = mock(ShardedEdgeSerialization.class);
+
+ final NodeShardApproximation nodeShardCounterSerialization =
+ mock( NodeShardApproximation.class );
+
+
+ final TimeService timeService = mock( TimeService.class );
+
+ final Keyspace keyspace = mock( Keyspace.class );
+
+
+ final MutationBatch batch = mock( MutationBatch.class );
+
+ when( keyspace.prepareMutationBatch() ).thenReturn( batch );
+
+
+ NodeShardAllocation approximation =
+ new NodeShardAllocationImpl( edgeShardSerialization, edgeColumnFamilies, shardedEdgeSerialization, nodeShardCounterSerialization, timeService,
+ graphFig, keyspace );
+
+ final Id nodeId = createId( "test" );
+ final String type = "type";
+ final String subType = "subType";
+
+
+ final long timeservicetime = System.currentTimeMillis();
+
+ when( timeService.getCurrentTime() ).thenReturn( timeservicetime );
+
+ final Shard futureShard = new Shard(10000l, timeservicetime) ;
+
+ /**
+ * Mock up returning a min shard, and a future shard
+ */
+ when( edgeShardSerialization
+ .getEdgeMetaData( same( scope ), same( nodeId ), eq( NodeType.TARGET), any( Optional.class ), same( type ),
+ same( subType ) ) ).thenReturn( Arrays.asList( futureShard ).iterator() );
+
+ final boolean result = approximation.auditMaxShard( scope, nodeId, NodeType.TARGET, type, subType );
+
+ assertFalse( "No shard allocated", result );
+ }
+
+
+ @Test
+ public void lowCountFutureShard() {
+ final EdgeShardSerialization edgeShardSerialization = mock( EdgeShardSerialization.class );
+
+ final EdgeColumnFamilies edgeColumnFamilies = mock(EdgeColumnFamilies.class);
+
+ final ShardedEdgeSerialization shardedEdgeSerialization = mock(ShardedEdgeSerialization.class);
+
+ final NodeShardApproximation nodeShardApproximation =
+ mock( NodeShardApproximation.class );
+
+
+ final TimeService timeService = mock( TimeService.class );
+
+ final Keyspace keyspace = mock( Keyspace.class );
+
+ final MutationBatch batch = mock( MutationBatch.class );
+
+ when( keyspace.prepareMutationBatch() ).thenReturn( batch );
+
+
+ NodeShardAllocation approximation =
+ new NodeShardAllocationImpl( edgeShardSerialization, edgeColumnFamilies, shardedEdgeSerialization, nodeShardApproximation, timeService,
+ graphFig, keyspace );
+
+ final Id nodeId = createId( "test" );
+ final String type = "type";
+ final String subType = "subType";
+
+
+ final long timeservicetime = System.currentTimeMillis();
+
+ when( timeService.getCurrentTime() ).thenReturn( timeservicetime );
+
+
+ /**
+ * Mock up returning a min shard, and a future shard
+ */
+ when( edgeShardSerialization
+ .getEdgeMetaData( same( scope ), same( nodeId ), eq(NodeType.TARGET), any( Optional.class ), same( type ),
+ same( subType ) ) ).thenReturn( Arrays.asList( new Shard(0l, 0l) ).iterator() );
+
+
+ //return a shard size < our max by 1
+
+ final long count = graphFig.getShardSize() - 1;
+
+ when( nodeShardApproximation.getCount(scope, nodeId, NodeType.TARGET, 0l, type, subType )).thenReturn( count );
+
+ final boolean result = approximation.auditMaxShard( scope, nodeId, NodeType.TARGET, type, subType );
+
+ assertFalse( "Shard allocated", result );
+ }
+
+
+ @Test
+ public void equalCountFutureShard() {
+ final EdgeShardSerialization edgeShardSerialization = mock( EdgeShardSerialization.class );
+
+ final EdgeColumnFamilies edgeColumnFamilies = mock(EdgeColumnFamilies.class);
+
+ final ShardedEdgeSerialization shardedEdgeSerialization = mock(ShardedEdgeSerialization.class);
+
+ final NodeShardApproximation nodeShardApproximation =
+ mock( NodeShardApproximation.class );
+
+
+ final TimeService timeService = mock( TimeService.class );
+
+ final Keyspace keyspace = mock( Keyspace.class );
+
+ final MutationBatch batch = mock(MutationBatch.class);
+
+ when(keyspace.prepareMutationBatch()).thenReturn( batch );
+
+
+ NodeShardAllocation approximation =
+ new NodeShardAllocationImpl( edgeShardSerialization, edgeColumnFamilies, shardedEdgeSerialization, nodeShardApproximation, timeService,
+ graphFig, keyspace );
+
+ final Id nodeId = createId( "test" );
+ final String type = "type";
+ final String subType = "subType";
+
+
+ final long timeservicetime = System.currentTimeMillis();
+
+ when( timeService.getCurrentTime() ).thenReturn( timeservicetime );
+
+
+ /**
+ * Mock up returning a min shard
+ */
+ when( edgeShardSerialization
+ .getEdgeMetaData( same( scope ), same( nodeId ), eq(NodeType.SOURCE), any( Optional.class ), same( type ),
+ same( subType ) ) ).thenReturn( Arrays.asList( new Shard( 0l, 0l ) ).iterator() );
+
+
+ final long shardCount = graphFig.getShardSize();
+
+ //return a shard size equal to our max
+ when( nodeShardApproximation
+ .getCount( scope , nodeId, NodeType.SOURCE, 0l,type , subType ))
+ .thenReturn( shardCount );
+
+ ArgumentCaptor<Long> shardValue = ArgumentCaptor.forClass( Long.class );
+ ArgumentCaptor<Long> timestampValue = ArgumentCaptor.forClass( Long.class );
+
+
+
+
+ //mock up our mutation
+ when( edgeShardSerialization
+ .writeEdgeMeta( same( scope ), same( nodeId ), eq(NodeType.SOURCE), shardValue.capture(), timestampValue.capture(), same( type ), same( subType ) ) )
+ .thenReturn( mock( MutationBatch.class ) );
+
+
+ final SimpleMarkedEdge returnedEdge = new SimpleMarkedEdge( nodeId, type, createId("subType"), 10005l, false );
+ final Iterator<MarkedEdge> edgeIterator = Collections.singleton( (MarkedEdge)returnedEdge ).iterator();
+
+ //mock up returning the value
+ when( shardedEdgeSerialization.getEdgesFromSourceByTargetType( same( edgeColumnFamilies ), same( scope ),
+ any( SearchByIdType.class ), any(Iterator.class) )).thenReturn( edgeIterator );
+
+
+
+ final boolean result = approximation.auditMaxShard( scope, nodeId, NodeType.SOURCE, type, subType );
+
+ assertTrue( "Shard allocated", result );
+
+ //check our new allocated UUID
+
+
+ final long savedTimestamp = timestampValue.getValue();
+
+
+ assertEquals( "Expected time service time", timeservicetime, savedTimestamp );
+
+
+
+ //now check our max value was set
+
+ final long savedShardPivot = shardValue.getValue();
+
+ assertEquals("Expected max value to be the same", returnedEdge.getTimestamp(), savedShardPivot);
+ }
+
+
+
+
+ @Test
+ public void futureCountShardCleanup() {
+ final EdgeShardSerialization edgeShardSerialization = mock( EdgeShardSerialization.class );
+
+ final EdgeColumnFamilies edgeColumnFamilies = mock(EdgeColumnFamilies.class);
+
+ final ShardedEdgeSerialization shardedEdgeSerialization = mock(ShardedEdgeSerialization.class);
+
+ final NodeShardApproximation nodeShardApproximation =
+ mock( NodeShardApproximation.class );
+
+
+ final TimeService timeService = mock( TimeService.class );
+
+ final Keyspace keyspace = mock( Keyspace.class );
+
+ final MutationBatch batch = mock(MutationBatch.class);
+
+ when(keyspace.prepareMutationBatch()).thenReturn( batch );
+
+
+ NodeShardAllocation approximation =
+ new NodeShardAllocationImpl(edgeShardSerialization, edgeColumnFamilies, shardedEdgeSerialization, nodeShardApproximation, timeService,
+ graphFig, keyspace );
+
+ final Id nodeId = createId( "test" );
+ final String type = "type";
+ final String subType = "subType";
+
+
+ /**
+ * Use the time service to generate UUIDS
+ */
+ final long timeservicetime = System.currentTimeMillis();
+
+
+ when( timeService.getCurrentTime() ).thenReturn( timeservicetime );
+
+ assertTrue("Shard cache mocked", graphFig.getShardCacheTimeout() > 0);
+
+
+ /**
+ * Simulates clock drift when 2 nodes create future shards near one another
+ */
+ final long futureTime = timeService.getCurrentTime() + 2 * graphFig.getShardCacheTimeout();
+
+
+ final Shard minShard = new Shard(0l, 0l);
+
+ /**
+ * Simulate slow node
+ */
+
+ //our second shard is the "oldest", and hence should be returned in the iterator. Future shard 1 and 3 should be removed
+ final Shard futureShard1 = new Shard(futureTime - 1, timeservicetime+1000);
+
+ final Shard futureShard2 = new Shard(futureTime + 10000, timeservicetime);
+
+ final Shard futureShard3 = new Shard(futureShard2.getShardIndex() + 10000, timeservicetime+2000);
+
+ /**
+ * Mock up returning a min shard
+ */
+ when( edgeShardSerialization
+ .getEdgeMetaData( same( scope ), same( nodeId ), eq(NodeType.TARGET), any( Optional.class ), same( type ),
+ same( subType ) ) ).thenReturn(
+ Arrays.asList( futureShard3, futureShard2, futureShard1, minShard ).iterator() );
+
+
+
+ ArgumentCaptor<Long> newLongValue = ArgumentCaptor.forClass( Long.class );
+
+
+
+
+ //mock up our mutation
+ when( edgeShardSerialization
+ .removeEdgeMeta( same( scope ), same( nodeId ), eq(NodeType.TARGET), newLongValue.capture(), same( type ), same( subType ) ) )
+ .thenReturn( mock( MutationBatch.class ) );
+
+
+ final Iterator<Shard>
+ result = approximation.getShards( scope, nodeId, NodeType.TARGET, Optional.<Shard>absent(), type, subType );
+
+
+ assertTrue( "Shards present", result.hasNext() );
+
+ assertEquals("Only single next shard returned", futureShard2, result.next());
+
+ assertTrue("Shards present", result.hasNext());
+
+ assertEquals("Previous shard present", 0l, result.next().getShardIndex());
+
+ assertFalse("No shards left", result.hasNext());
+
+ /**
+ * Now we need to verify that both our mutations have been added
+ */
+
+ List<Long> values = newLongValue.getAllValues();
+
+ assertEquals("2 values removed", 2, values.size());
+
+ assertEquals("Deleted Max Future", futureShard1.getShardIndex(), values.get( 0 ).longValue());
+ assertEquals("Deleted Next Future", futureShard3.getShardIndex(), values.get( 1 ).longValue());
+
+ }
+
+
+
+
+ @Test
+ public void noShardsReturns() {
+ final EdgeShardSerialization edgeShardSerialization = mock( EdgeShardSerialization.class );
+
+ final EdgeColumnFamilies edgeColumnFamilies = mock(EdgeColumnFamilies.class);
+
+ final ShardedEdgeSerialization shardedEdgeSerialization = mock(ShardedEdgeSerialization.class);
+
+ final NodeShardApproximation nodeShardApproximation =
+ mock( NodeShardApproximation.class );
+
+
+ final TimeService timeService = mock( TimeService.class );
+
+ final Keyspace keyspace = mock( Keyspace.class );
+
+ final MutationBatch batch = mock( MutationBatch.class );
+
+ when( keyspace.prepareMutationBatch() ).thenReturn( batch );
+
+ NodeShardAllocation approximation =
+ new NodeShardAllocationImpl( edgeShardSerialization, edgeColumnFamilies, shardedEdgeSerialization, nodeShardApproximation, timeService,
+ graphFig, keyspace );
+
+ final Id nodeId = createId( "test" );
+ final String type = "type";
+ final String subType = "subType";
+
+ /**
+ * Mock up returning an empty iterator, our audit shouldn't create a new shard
+ */
+ when( edgeShardSerialization
+ .getEdgeMetaData( same( scope ), same( nodeId ), eq(NodeType.TARGET), any( Optional.class ), same( type ),
+ same( subType ) ) ).thenReturn( Collections.<Shard>emptyList().iterator() );
+
+ final Iterator<Shard> result = approximation.getShards( scope, nodeId, NodeType.TARGET, Optional.<Shard>absent(), type,
+ subType );
+
+ assertEquals("0 shard allocated", 0l, result.next().getShardIndex());
+
+ assertFalse( "No shard allocated", result.hasNext() );
+ }
}