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:04 UTC
[11/11] git commit: Work in progress. Added testing
Work in progress. Added testing
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/cd043b14
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/cd043b14
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/cd043b14
Branch: refs/heads/USERGRID-188
Commit: cd043b143f5f467679befb4af0659448a5a17abb
Parents: 3ecd603
Author: Todd Nine <tn...@apigee.com>
Authored: Wed Jul 23 10:39:35 2014 -0600
Committer: Todd Nine <tn...@apigee.com>
Committed: Wed Jul 23 10:39:35 2014 -0600
----------------------------------------------------------------------
.../impl/shard/EdgeShardSerialization.java | 12 +-
.../graph/serialization/impl/shard/Shard.java | 27 +-
.../impl/shard/ShardEntryGroup.java | 199 ++--
.../impl/shard/impl/EdgeSearcher.java | 2 +-
.../shard/impl/EdgeShardSerializationImpl.java | 35 +-
.../shard/impl/NodeShardAllocationImpl.java | 8 +-
.../impl/shard/impl/NodeShardCacheImpl.java | 2 +-
.../shard/impl/ShardEntryGroupIterator.java | 2 -
.../impl/shard/EdgeShardSerializationTest.java | 119 +-
.../impl/shard/NodeShardAllocationTest.java | 1104 +++++++++---------
.../impl/shard/ShardEntryGroupTest.java | 129 +-
11 files changed, 891 insertions(+), 748 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/cd043b14/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/EdgeShardSerialization.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/EdgeShardSerialization.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/EdgeShardSerialization.java
index e91dda3..1f15107 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/EdgeShardSerialization.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/EdgeShardSerialization.java
@@ -40,11 +40,11 @@ public interface EdgeShardSerialization extends Migration{
* @param scope The scope to write
* @param nodeId The id in the edge
* @param nodeType Is the node a source or target node
- * @param shard The next time to write
- * @param timestamp the timestamp to set on the shard
+ * @param shard The shard to write
* @param types The types to write to. Can be edge type, or edgeType+id type
*/
- public MutationBatch writeEdgeMeta(ApplicationScope scope, Id nodeId, NodeType nodeType, long shard, long timestamp, String... types);
+ public MutationBatch writeShardMeta( ApplicationScope scope, Id nodeId, NodeType nodeType, Shard shard,
+ String... types );
/**
* Get an iterator of all meta data and types. Returns a range from High to low
@@ -55,7 +55,8 @@ public interface EdgeShardSerialization extends Migration{
* @param types The types to use
* @return
*/
- public Iterator<Shard> getEdgeMetaData(ApplicationScope scope, Id nodeId, NodeType nodeType, Optional<Shard> start, String... types);
+ public Iterator<Shard> getShardMetaData( ApplicationScope scope, Id nodeId, NodeType nodeType,
+ Optional<Shard> start, String... types );
/**
* Remove the shard from the edge meta data from the types.
@@ -67,6 +68,7 @@ public interface EdgeShardSerialization extends Migration{
* @param types
* @return
*/
- public MutationBatch removeEdgeMeta(ApplicationScope scope, Id nodeId, NodeType nodeType, long shard, String... types);
+ public MutationBatch removeShardMeta( ApplicationScope scope, Id nodeId, NodeType nodeType, Shard shard,
+ String... types );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/cd043b14/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/Shard.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/Shard.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/Shard.java
index f95699b..4b58224 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/Shard.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/Shard.java
@@ -23,11 +23,13 @@ public class Shard implements Comparable<Shard> {
private final long shardIndex;
private final long createdTime;
+ private final boolean compacted;
- public Shard( final long shardIndex, final long createdTime ) {
+ public Shard( final long shardIndex, final long createdTime, final boolean compacted ) {
this.shardIndex = shardIndex;
this.createdTime = createdTime;
+ this.compacted = compacted;
}
@@ -48,9 +50,15 @@ public class Shard implements Comparable<Shard> {
/**
+ * Return true if this shard has been compacted
+ */
+ public boolean isCompacted() {
+ return compacted;
+ }
+
+
+ /**
* Compare the shards based on the timestamp first, then the created time second
- * @param o
- * @return
*/
@Override
public int compareTo( final Shard o ) {
@@ -70,6 +78,19 @@ public class Shard implements Comparable<Shard> {
return -1;
}
+ else {
+
+ //kind of arbitrary compacted takes precedence
+ if ( compacted && !o.compacted ) {
+ return 1;
+ }
+
+ else if ( !compacted && o.compacted ){
+ return -1;
+ }
+
+
+ }
return 0;
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/cd043b14/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardEntryGroup.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardEntryGroup.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardEntryGroup.java
index 6bd1421..ebb037e 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardEntryGroup.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardEntryGroup.java
@@ -21,192 +21,173 @@ package org.apache.usergrid.persistence.graph.serialization.impl.shard;
import java.util.Collection;
import java.util.Collections;
-import java.util.Comparator;
-import java.util.TreeMap;
+import java.util.TreeSet;
/**
- * There are cases where we need to read or write to more than 1 shard. This object encapsulates
- * a set of shards that should be written to and read from. All reads should combine the data sets from
- * all shards in the group, and writes should be written to each shard. Once the shard can safely be compacted
- * a background process should be triggered to remove additional shards and make seeks faster. This multiread/write
- * should only occur during the time period of the delta (in milliseconds), after which the next read will asynchronously compact the
- * shards into a single shard.
+ * There are cases where we need to read or write to more than 1 shard. This object encapsulates a set of shards that
+ * should be written to and read from. All reads should combine the data sets from all shards in the group, and writes
+ * should be written to each shard. Once the shard can safely be compacted a background process should be triggered to
+ * remove additional shards and make seeks faster. This multiread/write should only occur during the time period of the
+ * delta (in milliseconds), after which the next read will asynchronously compact the shards into a single shard.
*/
public class ShardEntryGroup {
- private TreeMap<Long, Shard> shards;
-
- private Shard minShardByIndex;
+ private TreeSet<Shard> shards;
private final long delta;
- private Shard neighbor;
+ private Shard compactionTarget;
+
+
+ private long maxCreatedTime;
/**
* The max delta we accept in milliseconds for create time to be considered a member of this group
- * @param delta
*/
public ShardEntryGroup( final long delta ) {
this.delta = delta;
- this.shards = new TreeMap<>( ShardTimeComparator.INSTANCE );
+ this.shards = new TreeSet<>();
+ this.maxCreatedTime = 0;
}
/**
- * Only add a shard if the created timestamp is within the delta of one of the entries
- * @param shard
- * @return
+ * Only add a shard if it is within the rules require to meet a group. The rules are outlined below.
+ *
+ * Case 1) First shard in the group, always added
+ *
+ * Case 2) Shard is unmerged, it should be included with it's peers since other nodes may not have it yet
+ *
+ * Case 3) The list contains only non compacted shards, and this is last and and merged. It is considered a lower
+ * bound
*/
- public boolean addShard(final Shard shard){
-
- //compare the time and see if it falls withing any of the elements based on their timestamp
- final long shardCreateTime = shard.getCreatedTime();
-
- final Long lessThanKey = shards.floorKey( shardCreateTime );
-
- final Long greaterThanKey = shards.ceilingKey( shardCreateTime );
-
+ public boolean addShard( final Shard shard ) {
- final long lessThanDelta = shardCreateTime - lessThanKey;
+ //shards can be ar
- final long greaterThanDelta = greaterThanKey - shardCreateTime;
-
- if(lessThanDelta < delta || greaterThanDelta < delta ){
- this.shards.put( shardCreateTime, shard );
+ //compare the time and see if it falls withing any of the elements based on their timestamp
+ // final long shardCreateTime = shard.getCreatedTime();
+
+ // final Long lessThanKey = shards.floorKey( shardCreateTime );
+ //
+ // final Long greaterThanKey = shards.ceilingKey( shardCreateTime );
+ //
+ // //first into the set
+ // if ( lessThanKey == null && greaterThanKey == null ) {
+ // addShardInternal( shard );
+ // return true;
+ // }
+ //
+ // if ( lessThanKey != null && shardCreateTime - lessThanKey < delta ) {
+ // addShardInternal( shard );
+ // return true;
+ // }
+ //
+ //
+ // if ( greaterThanKey != null && greaterThanKey - shardCreateTime < delta ) {
+ // addShardInternal( shard );
+ //
+ // return true;
+ // }
+
+ if ( shards.size() == 0 ) {
+ addShardInternal( shard );
+ return true;
+ }
- if(shard.compareTo( minShardByIndex ) < 0){
- minShardByIndex = shard;
- }
+ //shard is not compacted, or it's predecessor isn't, we should include it in this group
+ if ( !shard.isCompacted() || !shards.last().isCompacted() ) {
+ addShardInternal( shard );
return true;
}
+
return false;
}
/**
- * Add the n-1 shard to the set. This is required, because nodes that have not yet updated their
- * shard caches can be writing reading to the n-1 node only
- *
- * @param shard The shard to possibly add as a neighbor
- * @return True if this shard as added as a neighbor, false otherwise
+ * Add the shard and set the min created time
*/
- public boolean setNeighbor( final Shard shard ){
+ private void addShardInternal( final Shard shard ) {
+ shards.add( shard );
- //not in the transition state don't set the neighbor, it will slow seeks down
- if(!isRolling()){
- return false;
- }
+ maxCreatedTime = Math.max( maxCreatedTime, shard.getCreatedTime() );
+ //it's not a compacted shard, so it's a candidate to be the compaction target
+ if ( !shard.isCompacted() && ( compactionTarget == null || shard.compareTo( compactionTarget ) < 0 ) ) {
+ compactionTarget = shard;
+ }
+ }
- neighbor = shard;
- this.shards.put( shard.getCreatedTime(), shard );
- return true;
+ /**
+ * Get the entries that we should read from.
+ */
+ public Collection<Shard> getReadShards() {
+ return shards;
}
/**
- * Get the entries that we should read from.
- *
- * @return
+ * Get the entries, with the max shard time being first. We write to all shards until they're migrated
*/
- public Collection<Shard> getReadShards(final long currentTime) {
+ public Collection<Shard> getWriteShards( long currentTime ) {
/**
- * The shards are still rolling (I.E can't be compacted)
+ * The shards in this set can be combined, we should only write to the compaction target to avoid
+ * adding data to other shards
*/
- if(needsCompaction( currentTime )){
- return shards.values();
+ if ( shouldCompact( currentTime ) ) {
+ return Collections.singleton( compactionTarget );
}
- return Collections.singleton(minShardByIndex);
- }
-
- /**
- * Get the entries, with the max shard time being first. We write to all shards until they're migrated
- *
- * @return
- */
- public Collection<Shard> getWriteShards() {
- return shards.values();
+ return shards;
}
/**
* Get the shard all compactions should write to
- * @return
*/
- public Shard getMergeTarget(){
- return minShardByIndex;
+ public Shard getCompactionTarget() {
+ return compactionTarget;
}
/**
* Returns true if the newest created shard is path the currentTime - delta
+ *
* @param currentTime The current system time in milliseconds
+ *
* @return True if these shards can safely be combined into a single shard, false otherwise
*/
- public boolean needsCompaction(final long currentTime){
+ public boolean shouldCompact( final long currentTime ) {
/**
* We don't have enough shards to compact, ignore
*/
- if(shards.size() < 2){
+ if ( shards.size() < 2 ) {
return false;
}
-
- final long maxTimestamp = shards.lastKey();
-
-
- return currentTime - delta > maxTimestamp;
- }
-
-
- /**
- * Return true if the shard is rolling. If this is the case, we want to include the n-1 entry, since everyone
- * may not yet have it until compaction is safe to perform
- * @return
- */
- private boolean isRolling(){
- return shards.size() > 1;
+ return currentTime - delta > maxCreatedTime;
}
/**
* Return true if this shard can be deleted AFTER all of the data in it has been moved
- * @param shard
- * @return
*/
- public boolean canBeDeleted(final Shard shard){
+ public boolean canBeDeleted( final Shard shard ) {
//if we're a neighbor shard (n-1) or the target compaction shard, we can't be deleted
- //we purposefully use .equals here, since 2 shards might have the same index with different timestamps (unlikely but could happen)
- if(shard == neighbor || getMergeTarget().equals( shard )){
- return false;
- }
-
- return true;
- }
-
- /**
- * Compares 2 shards based on create time. Does not handle nulls intentionally
- */
- private static final class ShardTimeComparator implements Comparator<Long> {
-
- public static final ShardTimeComparator INSTANCE = new ShardTimeComparator();
-
-
- @Override
- public int compare( final Long o1, final Long o2 ) {
- return o1.compareTo( o2 );
- }
+ //we purposefully use shard index comparison over .equals here, since 2 shards might have the same index with
+ // different timestamps
+ // (unlikely but could happen)
+ return !shard.isCompacted() && ( compactionTarget != null && compactionTarget.getShardIndex() != shard
+ .getShardIndex() );
}
-
-
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/cd043b14/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/EdgeSearcher.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/EdgeSearcher.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/EdgeSearcher.java
index d0e642c..8b39115 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/EdgeSearcher.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/EdgeSearcher.java
@@ -52,7 +52,7 @@ public abstract class EdgeSearcher<R, C, T> implements ColumnParser<C, T>, Itera
* TODO Shard fix this
*/
return ScopedRowKey
- .fromKey( scope, generateRowKey( shards.next().getMergeTarget().getShardIndex() ) );
+ .fromKey( scope, generateRowKey( shards.next().getCompactionTarget().getShardIndex() ) );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/cd043b14/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/EdgeShardSerializationImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/EdgeShardSerializationImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/EdgeShardSerializationImpl.java
index 745e02a..b6f65f9 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/EdgeShardSerializationImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/EdgeShardSerializationImpl.java
@@ -66,12 +66,9 @@ public class EdgeShardSerializationImpl implements EdgeShardSerialization {
*/
private static final MultiTennantColumnFamily<ApplicationScope, DirectedRowKey, Long> EDGE_SHARDS =
new MultiTennantColumnFamily<>( "Edge_Shards",
- new OrganizationScopedRowKeySerializer<>( new DirectedEdgeRowKeySerializer() ),
- LongSerializer.get() );
+ new OrganizationScopedRowKeySerializer<>( new EdgeShardRowKeySerializer() ), LongSerializer.get() );
- private static final byte HOLDER = 0x00;
-
private static final ShardColumnParser COLUMN_PARSER = new ShardColumnParser();
@@ -90,13 +87,14 @@ public class EdgeShardSerializationImpl implements EdgeShardSerialization {
@Override
- public MutationBatch writeEdgeMeta( final ApplicationScope scope, final Id nodeId, final NodeType nodeType,
- final long shard, final long timestamp, final String... types ) {
-
+ public MutationBatch writeShardMeta( final ApplicationScope scope, final Id nodeId, final NodeType nodeType,
+ final Shard shard, final String... types ) {
ValidationUtils.validateApplicationScope( scope );
ValidationUtils.verifyIdentity( nodeId );
- Preconditions.checkArgument( shard > -1, "shardId must be greater than -1" );
+ Preconditions.checkNotNull( shard );
+ Preconditions.checkArgument( shard.getShardIndex() > -1, "shardid must be greater than -1" );
+ Preconditions.checkArgument( shard.getCreatedTime() > -1, "createdTime must be greater than -1" );
Preconditions.checkNotNull( types );
final DirectedRowKey key = new DirectedRowKey( nodeId, nodeType, types );
@@ -105,15 +103,16 @@ public class EdgeShardSerializationImpl implements EdgeShardSerialization {
final MutationBatch batch = keyspace.prepareMutationBatch();
- batch.withTimestamp( timestamp ).withRow( EDGE_SHARDS, rowKey ).putColumn( shard, HOLDER );
+ batch.withTimestamp( shard.getCreatedTime() ).withRow( EDGE_SHARDS, rowKey )
+ .putColumn( shard.getShardIndex(), shard.isCompacted() );
return batch;
}
@Override
- public Iterator<Shard> getEdgeMetaData( final ApplicationScope scope, final Id nodeId, final NodeType nodeType,
- final Optional<Shard> start, final String... types ) {
+ public Iterator<Shard> getShardMetaData( final ApplicationScope scope, final Id nodeId, final NodeType nodeType,
+ final Optional<Shard> start, final String... types ) {
/**
* If the edge is present, we need to being seeking from this
*/
@@ -139,12 +138,14 @@ public class EdgeShardSerializationImpl implements EdgeShardSerialization {
@Override
- public MutationBatch removeEdgeMeta( final ApplicationScope scope, final Id nodeId, final NodeType nodeType,
- final long shard, final String... types ) {
+ public MutationBatch removeShardMeta( final ApplicationScope scope, final Id nodeId, final NodeType nodeType,
+ final Shard shard, final String... types ) {
ValidationUtils.validateApplicationScope( scope );
ValidationUtils.verifyIdentity( nodeId );
- Preconditions.checkArgument( shard > -1, "shard must be greater than -1" );
+ Preconditions.checkNotNull( shard );
+ Preconditions.checkArgument( shard.getShardIndex() > -1, "shardid must be greater than -1" );
+ Preconditions.checkArgument( shard.getCreatedTime() > -1, "createdTime must be greater than -1" );
Preconditions.checkNotNull( types );
final DirectedRowKey key = new DirectedRowKey( nodeId, nodeType, types );
@@ -153,7 +154,7 @@ public class EdgeShardSerializationImpl implements EdgeShardSerialization {
final MutationBatch batch = keyspace.prepareMutationBatch();
- batch.withRow( EDGE_SHARDS, rowKey ).deleteColumn( shard );
+ batch.withRow( EDGE_SHARDS, rowKey ).deleteColumn( shard.getShardIndex() );
return batch;
}
@@ -186,7 +187,7 @@ public class EdgeShardSerializationImpl implements EdgeShardSerialization {
}
- private static class DirectedEdgeRowKeySerializer implements CompositeFieldSerializer<DirectedRowKey> {
+ private static class EdgeShardRowKeySerializer implements CompositeFieldSerializer<DirectedRowKey> {
private static final IdRowCompositeSerializer ID_SER = IdRowCompositeSerializer.get();
@@ -247,7 +248,7 @@ public class EdgeShardSerializationImpl implements EdgeShardSerialization {
@Override
public Shard parseColumn( final Column<Long> column ) {
- return new Shard( column.getName(), column.getTimestamp() );
+ return new Shard( column.getName(), column.getTimestamp(), column.getBooleanValue() );
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/cd043b14/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 dbf8382..6accf91 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
@@ -88,7 +88,7 @@ public class NodeShardAllocationImpl implements NodeShardAllocation {
final Optional<Shard> maxShardId, final String... edgeTypes ) {
final Iterator<Shard> existingShards =
- edgeShardSerialization.getEdgeMetaData( scope, nodeId, nodeType, maxShardId, edgeTypes );
+ edgeShardSerialization.getShardMetaData( scope, nodeId, nodeType, maxShardId, edgeTypes );
return new ShardEntryGroupIterator( existingShards, graphFig.getShardMinDelta() );
}
@@ -104,7 +104,7 @@ public class NodeShardAllocationImpl implements NodeShardAllocation {
* latest
*/
final Iterator<Shard> maxShards =
- edgeShardSerialization.getEdgeMetaData( scope, nodeId, nodeType, Optional.<Shard>absent(), edgeType );
+ edgeShardSerialization.getShardMetaData( scope, nodeId, nodeType, Optional.<Shard>absent(), edgeType );
//if the first shard has already been allocated, do nothing.
@@ -204,10 +204,12 @@ public class NodeShardAllocationImpl implements NodeShardAllocation {
final long createTimestamp = timeService.getCurrentTime();
+ final Shard shard = new Shard(marked.getTimestamp(), createTimestamp, false);
+
try {
this.edgeShardSerialization
- .writeEdgeMeta( scope, nodeId, nodeType, marked.getTimestamp(), createTimestamp, edgeType )
+ .writeShardMeta( scope, nodeId, nodeType, shard, edgeType )
.execute();
}
catch ( ConnectionException e ) {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/cd043b14/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/NodeShardCacheImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/NodeShardCacheImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/NodeShardCacheImpl.java
index ad30dc6..6ac83cf 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/NodeShardCacheImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/NodeShardCacheImpl.java
@@ -249,7 +249,7 @@ public class NodeShardCacheImpl implements NodeShardCache {
this.shards = new TreeMap<>(ShardEntriesComparator.INSTANCE);
for ( ShardEntryGroup shard : IterableUtil.wrap( shards ) ) {
- this.shards.put(shard.getMergeTarget().getShardIndex() , shard );
+ this.shards.put(shard.getCompactionTarget().getShardIndex() , shard );
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/cd043b14/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/ShardEntryGroupIterator.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/ShardEntryGroupIterator.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/ShardEntryGroupIterator.java
index 39e198e..8e69be4 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/ShardEntryGroupIterator.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/ShardEntryGroupIterator.java
@@ -91,8 +91,6 @@ public class ShardEntryGroupIterator implements Iterator<ShardEntryGroup> {
//we can't add this shard to the current group. Add the group and return.
next = shardEntryGroup;
- shardEntryGroup.setNeighbor( shard );
-
sourceIterator.pushback( shard );
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/cd043b14/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/EdgeShardSerializationTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/EdgeShardSerializationTest.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/EdgeShardSerializationTest.java
index 937a51a..00e0164 100644
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/EdgeShardSerializationTest.java
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/EdgeShardSerializationTest.java
@@ -29,10 +29,10 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
+import org.apache.usergrid.persistence.collection.guice.MigrationManagerRule;
+import org.apache.usergrid.persistence.core.cassandra.CassandraRule;
import org.apache.usergrid.persistence.core.cassandra.ITRunner;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
-import org.apache.usergrid.persistence.core.cassandra.CassandraRule;
-import org.apache.usergrid.persistence.collection.guice.MigrationManagerRule;
import org.apache.usergrid.persistence.graph.guice.TestGraphModule;
import org.apache.usergrid.persistence.model.entity.Id;
import org.apache.usergrid.persistence.model.util.UUIDGenerator;
@@ -49,8 +49,8 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
-@RunWith( ITRunner.class )
-@UseModules( { TestGraphModule.class } )
+@RunWith(ITRunner.class)
+@UseModules({ TestGraphModule.class })
public class EdgeShardSerializationTest {
@ClassRule
@@ -86,67 +86,87 @@ public class EdgeShardSerializationTest {
final Id now = createId( "test" );
- final long slice1 = 1000l;
+ final long timestamp = 10000l;
- final long slice2 = slice1 * 2;
+ final Shard shard1 = new Shard( 1000l, timestamp, false );
- final long slice3 = slice2 * 2;
+ final Shard shard2 = new Shard( shard1.getShardIndex() * 2, timestamp, true );
+
+ final Shard shard3 = new Shard( shard2.getShardIndex() * 2, timestamp, false );
- final long timestamp = 10000l;
String[] types = { "edgeType", "subType" };
- MutationBatch batch = edgeShardSerialization.writeEdgeMeta( scope, now, NodeType.SOURCE, slice1, timestamp, types );
+ MutationBatch batch = edgeShardSerialization.writeShardMeta( scope, now, NodeType.SOURCE, shard1, types );
- batch.mergeShallow( edgeShardSerialization.writeEdgeMeta( scope, now, NodeType.SOURCE, slice2, timestamp, types ) );
+ batch.mergeShallow( edgeShardSerialization.writeShardMeta( scope, now, NodeType.SOURCE, shard2, types ) );
- batch.mergeShallow( edgeShardSerialization.writeEdgeMeta( scope, now, NodeType.SOURCE, slice3, timestamp, types ) );
+ batch.mergeShallow( edgeShardSerialization.writeShardMeta( scope, now, NodeType.SOURCE, shard3, types ) );
batch.execute();
- Iterator<Shard> results = edgeShardSerialization.getEdgeMetaData( scope, now, NodeType.SOURCE, Optional.<Shard>absent(), types );
+ Iterator<Shard> results =
+ edgeShardSerialization.getShardMetaData( scope, now, NodeType.SOURCE, Optional.<Shard>absent(), types );
Shard next = results.next();
- assertEquals( slice3, next.getShardIndex() );
+ assertEquals( shard3, next.getShardIndex() );
+
+ assertEquals( shard3.getCreatedTime(), next.getCreatedTime() );
+
+ assertEquals( shard3.isCompacted(), next.isCompacted() );
- assertEquals( timestamp, next.getCreatedTime());
next = results.next();
- assertEquals( slice2, next.getShardIndex() );
+ assertEquals( shard2, next.getShardIndex() );
+
+ assertEquals( shard2.getCreatedTime(), next.getCreatedTime() );
+
+ assertEquals( shard2.isCompacted(), next.isCompacted() );
- assertEquals( timestamp, next.getCreatedTime());
next = results.next();
- assertEquals( slice1, next.getShardIndex() );
+ assertEquals( shard1, next.getShardIndex() );
+
+ assertEquals( shard1.getCreatedTime(), next.getCreatedTime() );
- assertEquals( timestamp, next.getCreatedTime());
+ assertEquals( shard1.isCompacted(), next.isCompacted() );
assertFalse( results.hasNext() );
//test we get nothing with the other node type
- results = edgeShardSerialization.getEdgeMetaData( scope, now, NodeType.TARGET, Optional.<Shard>absent(), types );
+ results =
+ edgeShardSerialization.getShardMetaData( scope, now, NodeType.TARGET, Optional.<Shard>absent(), types );
- assertFalse(results.hasNext());
+ assertFalse( results.hasNext() );
//test paging and size
- results = edgeShardSerialization.getEdgeMetaData( scope, now, NodeType.SOURCE, Optional.of( new Shard( slice2, 0l ) ), types );
+ results = edgeShardSerialization.getShardMetaData( scope, now, NodeType.SOURCE, Optional.of( shard2 ), types );
next = results.next();
- assertEquals( slice2, next.getShardIndex() );
- assertEquals( timestamp, next.getCreatedTime());
+
+ assertEquals( shard2, next.getShardIndex() );
+
+ assertEquals( shard2.getCreatedTime(), next.getCreatedTime() );
+
+ assertEquals( shard2.isCompacted(), next.isCompacted() );
+
next = results.next();
- assertEquals( slice1, next.getShardIndex() );
- assertEquals( timestamp, next.getCreatedTime());
+
+ assertEquals( shard1, next.getShardIndex() );
+
+ assertEquals( shard1.getCreatedTime(), next.getCreatedTime() );
+
+ assertEquals( shard1.isCompacted(), next.isCompacted() );
assertFalse( results.hasNext() );
@@ -158,57 +178,66 @@ public class EdgeShardSerializationTest {
final Id now = createId( "test" );
- final long slice1 = 1000l;
- final long slice2 = slice1 * 2;
+ final long timestamp = 10000l;
- final long slice3 = slice2 * 2;
+ final Shard shard1 = new Shard( 1000l, timestamp, false );
- final long timestamp = 10000l;
+ final Shard shard2 = new Shard( shard1.getShardIndex() * 2, timestamp, true );
+
+ final Shard shard3 = new Shard( shard2.getShardIndex() * 2, timestamp, false );
String[] types = { "edgeType", "subType" };
- MutationBatch batch = edgeShardSerialization.writeEdgeMeta( scope, now, NodeType.SOURCE, slice1, timestamp, types );
+ MutationBatch batch =
+ edgeShardSerialization.writeShardMeta( scope, now, NodeType.SOURCE, shard1, types );
- batch.mergeShallow( edgeShardSerialization.writeEdgeMeta( scope, now, NodeType.SOURCE, slice2, timestamp,types ) );
+ batch.mergeShallow(
+ edgeShardSerialization.writeShardMeta( scope, now, NodeType.SOURCE, shard2, types ) );
- batch.mergeShallow( edgeShardSerialization.writeEdgeMeta( scope, now, NodeType.SOURCE, slice3, timestamp, types ) );
+ batch.mergeShallow(
+ edgeShardSerialization.writeShardMeta( scope, now, NodeType.SOURCE, shard3, types ) );
batch.execute();
- Iterator<Shard> results = edgeShardSerialization.getEdgeMetaData( scope, now, NodeType.SOURCE, Optional.<Shard>absent(), types );
+ Iterator<Shard> results =
+ edgeShardSerialization.getShardMetaData( scope, now, NodeType.SOURCE, Optional.<Shard>absent(), types );
- assertEquals( slice3, results.next().getShardIndex() );
+ assertEquals( shard3, results.next() );
- assertEquals( slice2, results.next().getShardIndex() );
+ assertEquals( shard2, results.next() );
- assertEquals( slice1, results.next().getShardIndex() );
+ assertEquals( shard1, results.next() );
assertFalse( results.hasNext() );
//test nothing with other type
- results = edgeShardSerialization.getEdgeMetaData( scope, now, NodeType.TARGET, Optional.<Shard>absent(), types );
+ results =
+ edgeShardSerialization.getShardMetaData( scope, now, NodeType.TARGET, Optional.<Shard>absent(), types );
+
+ assertFalse( results.hasNext() );
- assertFalse(results.hasNext());
//test paging and size
- edgeShardSerialization.removeEdgeMeta( scope, now, NodeType.SOURCE, slice1, types ).execute();
+ edgeShardSerialization.removeShardMeta( scope, now, NodeType.SOURCE, shard1, types ).execute();
- results = edgeShardSerialization.getEdgeMetaData( scope, now, NodeType.SOURCE, Optional.<Shard>absent(), types );
+ results =
+ edgeShardSerialization.getShardMetaData( scope, now, NodeType.SOURCE, Optional.<Shard>absent(), types );
- assertEquals( slice3, results.next().getShardIndex() );
+ assertEquals( shard3, results.next().getShardIndex() );
- assertEquals( slice2, results.next().getShardIndex() );
+ assertEquals( shard2, results.next().getShardIndex() );
assertFalse( results.hasNext() );
- edgeShardSerialization.removeEdgeMeta( scope, now, NodeType.SOURCE, slice2, types ).execute();
+ edgeShardSerialization.removeShardMeta( scope, now, NodeType.SOURCE, shard2, types ).execute();
- edgeShardSerialization.removeEdgeMeta( scope, now, NodeType.SOURCE, slice3, types ).execute();
+ edgeShardSerialization.removeShardMeta( scope, now, NodeType.SOURCE, shard3, types ).execute();
- results = edgeShardSerialization.getEdgeMetaData( scope, now, NodeType.SOURCE, Optional.<Shard>absent(), types );
+ results =
+ edgeShardSerialization.getShardMetaData( scope, now, NodeType.SOURCE, Optional.<Shard>absent(), types );
assertFalse( results.hasNext() );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/cd043b14/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 8c60e2c..0ac7f78 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
@@ -86,556 +86,556 @@ public class NodeShardAllocationTest {
}
- @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 timestamps
- */
- final long timeservicetime = 10000;
-
-
- 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 minDelta = graphFig.getShardMinDelta();
-
-
- final Shard minShard = new Shard( 0l, 0l );
-
- //a shard that isn't our minimum, but exists after compaction
- final Shard compactedShard = new Shard( 5000, 1000 );
-
- /**
- * Simulate different node time allocation
- */
-
- final long minTime = 10000;
- //our second shard is the "oldest", and hence should be returned in the iterator. Future shard 1 and 3
- // should be removed
-
- //this should get dropped, It's allocated after future shard2 even though the time is less
- final Shard futureShard1 = new Shard( 10000, minTime + minDelta );
-
- //should get kept.
- final Shard futureShard2 = new Shard( 10005, minTime );
-
- //should be removed
- final Shard futureShard3 = new Shard( 10010, minTime + minDelta / 2 );
-
- /**
- * 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, compactedShard, 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<ShardEntryGroup> result =
- approximation.getShards( scope, nodeId, NodeType.TARGET, Optional.<Shard>absent(), type, subType );
-
-
- assertTrue( "Shards present", result.hasNext() );
-
-
- ShardEntryGroup shardEntryGroup = result.next();
-
- assertEquals( "Future shard returned", futureShard1, shardEntryGroup.getMergeTarget() );
-
-
- //now verify all 4 are in this group. This is because the first shard (0,0) (n-1_ may be the only shard other
- //nodes see while we're rolling our state. This means it should be read and merged from as well
-
- Collection<Shard> writeShards = shardEntryGroup.getWriteShards();
-
- assertEquals( "Shard size as expected", 4, writeShards.size() );
-
- assertTrue( writeShards.contains( futureShard1 ) );
- assertTrue( writeShards.contains( futureShard2 ) );
- assertTrue( writeShards.contains( futureShard3 ) );
- assertTrue( writeShards.contains( compactedShard ) );
-
-
- Collection<Shard> readShards = shardEntryGroup.getReadShards( minTime + minDelta );
-
- assertEquals( "Shard size as expected", 4, readShards.size() );
-
- assertTrue( readShards.contains( futureShard1 ) );
- assertTrue( readShards.contains( futureShard2 ) );
- assertTrue( readShards.contains( futureShard3 ) );
- assertTrue( readShards.contains( compactedShard ) );
-
-
- assertTrue( "Shards present", result.hasNext() );
-
- shardEntryGroup = result.next();
-
- writeShards = shardEntryGroup.getWriteShards();
-
-
- assertTrue( "Previous shard present", writeShards.contains( minShard ) );
-
-
- writeShards = shardEntryGroup.getReadShards( minTime + minDelta );
-
-
- assertTrue( "Previous shard present", writeShards.contains( minShard ) );
-
-
- assertFalse( "No shards left", result.hasNext() );
- }
-
-
- @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 );
-
- when( timeService.getCurrentTime() ).thenReturn( 10000l );
-
- 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<ShardEntryGroup> result =
- approximation.getShards( scope, nodeId, NodeType.TARGET, Optional.<Shard>absent(), type, subType );
-
-
- ShardEntryGroup shardEntryGroup = result.next();
-
- final Shard expected = new Shard( 0, 0 );
-
- assertEquals( "Future shard returned", expected, shardEntryGroup.getMergeTarget() );
-
-
- //now verify all 4 are in this group. This is because the first shard (0,0) (n-1_ may be the only shard other
- //nodes see while we're rolling our state. This means it should be read and merged from as well
-
- Collection<Shard> writeShards = shardEntryGroup.getWriteShards();
-
- Collection<Shard> readShards = shardEntryGroup.getReadShards( 10000l );
-
-
- assertTrue( "0 shard allocated", writeShards.contains( expected ) );
-
- assertTrue( "0 shard allocated", readShards.contains( expected ) );
-
-
- assertFalse( "No shard allocated", result.hasNext() );
- }
-
-
- @Test
- public void invalidConfiguration() {
-
- final GraphFig graphFig = mock( GraphFig.class );
-
- final EdgeShardSerialization edgeShardSerialization = mock( EdgeShardSerialization.class );
-
- final EdgeColumnFamilies edgeColumnFamilies = mock( EdgeColumnFamilies.class );
-
- final ShardedEdgeSerialization shardedEdgeSerialization = mock( ShardedEdgeSerialization.class );
-
- final NodeShardApproximation nodeShardApproximation = mock( NodeShardApproximation.class );
-
-
- /**
- * Return 100000 milliseconds
- */
- final TimeService timeService = mock( TimeService.class );
-
- final long time = 100000l;
-
- when( timeService.getCurrentTime() ).thenReturn( time );
-
-
- final long cacheTimeout = 30000l;
-
- when( graphFig.getShardCacheTimeout() ).thenReturn( 30000l );
-
-
- final long tooSmallDelta = ( long ) ( ( cacheTimeout * 2 ) * .99 );
-
- when( graphFig.getShardMinDelta() ).thenReturn( tooSmallDelta );
-
-
- 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 );
-
-
- /**
- * Should throw an exception
- */
- try {
- approximation.getMinTime();
- fail( "Should have thrown a GraphRuntimeException" );
- }
- catch ( GraphRuntimeException gre ) {
- //swallow
- }
-
- //now test something that passes.
-
- final long minDelta = cacheTimeout * 2;
-
- when( graphFig.getShardMinDelta() ).thenReturn( minDelta );
-
- long returned = approximation.getMinTime();
-
- long expectedReturned = time - minDelta;
-
- assertEquals( expectedReturned, returned );
-
- final long delta = cacheTimeout * 4;
-
- when( graphFig.getShardMinDelta() ).thenReturn( delta );
-
- returned = approximation.getMinTime();
-
- expectedReturned = time - delta;
-
- assertEquals( expectedReturned, returned );
- }
+// @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
+// .getShardMetaData( 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, compacted );
+//
+// /**
+// * Mock up returning a min shard, and a future shard
+// */
+// when( edgeShardSerialization
+// .getShardMetaData( 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
+// .getShardMetaData( same( scope ), same( nodeId ), eq( NodeType.TARGET ), any( Optional.class ),
+// same( type ), same( subType ) ) ).thenReturn( Arrays.asList( new Shard( 0l, 0l, compacted ) ).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
+// .getShardMetaData( same( scope ), same( nodeId ), eq( NodeType.SOURCE ), any( Optional.class ),
+// same( type ), same( subType ) ) ).thenReturn( Arrays.asList( new Shard( 0l, 0l, compacted ) ).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
+// .writeShardMeta( 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 timestamps
+// */
+// final long timeservicetime = 10000;
+//
+//
+// 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 minDelta = graphFig.getShardMinDelta();
+//
+//
+// final Shard minShard = new Shard( 0l, 0l, compacted );
+//
+// //a shard that isn't our minimum, but exists after compaction
+// final Shard compactedShard = new Shard( 5000, 1000, compacted );
+//
+// /**
+// * Simulate different node time allocation
+// */
+//
+// final long minTime = 10000;
+// //our second shard is the "oldest", and hence should be returned in the iterator. Future shard 1 and 3
+// // should be removed
+//
+// //this should get dropped, It's allocated after future shard2 even though the time is less
+// final Shard futureShard1 = new Shard( 10000, minTime + minDelta, compacted );
+//
+// //should get kept.
+// final Shard futureShard2 = new Shard( 10005, minTime, compacted );
+//
+// //should be removed
+// final Shard futureShard3 = new Shard( 10010, minTime + minDelta / 2, compacted );
+//
+// /**
+// * Mock up returning a min shard
+// */
+// when( edgeShardSerialization
+// .getShardMetaData( same( scope ), same( nodeId ), eq( NodeType.TARGET ), any( Optional.class ),
+// same( type ), same( subType ) ) ).thenReturn(
+// Arrays.asList( futureShard3, futureShard2, futureShard1, compactedShard, minShard ).iterator() );
+//
+//
+// ArgumentCaptor<Long> newLongValue = ArgumentCaptor.forClass( Long.class );
+//
+//
+// //mock up our mutation
+// when( edgeShardSerialization
+// .removeShardMeta( same( scope ), same( nodeId ), eq( NodeType.TARGET ), newLongValue.capture(),
+// same( type ), same( subType ) ) ).thenReturn( mock( MutationBatch.class ) );
+//
+//
+// final Iterator<ShardEntryGroup> result =
+// approximation.getShards( scope, nodeId, NodeType.TARGET, Optional.<Shard>absent(), type, subType );
+//
+//
+// assertTrue( "Shards present", result.hasNext() );
+//
+//
+// ShardEntryGroup shardEntryGroup = result.next();
+//
+// assertEquals( "Future shard returned", futureShard1, shardEntryGroup.getCompactionTarget() );
+//
+//
+// //now verify all 4 are in this group. This is because the first shard (0,0) (n-1_ may be the only shard other
+// //nodes see while we're rolling our state. This means it should be read and merged from as well
+//
+// Collection<Shard> writeShards = shardEntryGroup.getWriteShards();
+//
+// assertEquals( "Shard size as expected", 4, writeShards.size() );
+//
+// assertTrue( writeShards.contains( futureShard1 ) );
+// assertTrue( writeShards.contains( futureShard2 ) );
+// assertTrue( writeShards.contains( futureShard3 ) );
+// assertTrue( writeShards.contains( compactedShard ) );
+//
+//
+// Collection<Shard> readShards = shardEntryGroup.getReadShards( minTime + minDelta );
+//
+// assertEquals( "Shard size as expected", 4, readShards.size() );
+//
+// assertTrue( readShards.contains( futureShard1 ) );
+// assertTrue( readShards.contains( futureShard2 ) );
+// assertTrue( readShards.contains( futureShard3 ) );
+// assertTrue( readShards.contains( compactedShard ) );
+//
+//
+// assertTrue( "Shards present", result.hasNext() );
+//
+// shardEntryGroup = result.next();
+//
+// writeShards = shardEntryGroup.getWriteShards();
+//
+//
+// assertTrue( "Previous shard present", writeShards.contains( minShard ) );
+//
+//
+// writeShards = shardEntryGroup.getReadShards( minTime + minDelta );
+//
+//
+// assertTrue( "Previous shard present", writeShards.contains( minShard ) );
+//
+//
+// assertFalse( "No shards left", result.hasNext() );
+// }
+//
+//
+// @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 );
+//
+// when( timeService.getCurrentTime() ).thenReturn( 10000l );
+//
+// 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
+// .getShardMetaData( same( scope ), same( nodeId ), eq( NodeType.TARGET ), any( Optional.class ),
+// same( type ), same( subType ) ) ).thenReturn( Collections.<Shard>emptyList().iterator() );
+//
+// final Iterator<ShardEntryGroup> result =
+// approximation.getShards( scope, nodeId, NodeType.TARGET, Optional.<Shard>absent(), type, subType );
+//
+//
+// ShardEntryGroup shardEntryGroup = result.next();
+//
+// final Shard expected = new Shard( 0, 0, compacted );
+//
+// assertEquals( "Future shard returned", expected, shardEntryGroup.getCompactionTarget() );
+//
+//
+// //now verify all 4 are in this group. This is because the first shard (0,0) (n-1_ may be the only shard other
+// //nodes see while we're rolling our state. This means it should be read and merged from as well
+//
+// Collection<Shard> writeShards = shardEntryGroup.getWriteShards();
+//
+// Collection<Shard> readShards = shardEntryGroup.getReadShards( 10000l );
+//
+//
+// assertTrue( "0 shard allocated", writeShards.contains( expected ) );
+//
+// assertTrue( "0 shard allocated", readShards.contains( expected ) );
+//
+//
+// assertFalse( "No shard allocated", result.hasNext() );
+// }
+//
+//
+// @Test
+// public void invalidConfiguration() {
+//
+// final GraphFig graphFig = mock( GraphFig.class );
+//
+// final EdgeShardSerialization edgeShardSerialization = mock( EdgeShardSerialization.class );
+//
+// final EdgeColumnFamilies edgeColumnFamilies = mock( EdgeColumnFamilies.class );
+//
+// final ShardedEdgeSerialization shardedEdgeSerialization = mock( ShardedEdgeSerialization.class );
+//
+// final NodeShardApproximation nodeShardApproximation = mock( NodeShardApproximation.class );
+//
+//
+// /**
+// * Return 100000 milliseconds
+// */
+// final TimeService timeService = mock( TimeService.class );
+//
+// final long time = 100000l;
+//
+// when( timeService.getCurrentTime() ).thenReturn( time );
+//
+//
+// final long cacheTimeout = 30000l;
+//
+// when( graphFig.getShardCacheTimeout() ).thenReturn( 30000l );
+//
+//
+// final long tooSmallDelta = ( long ) ( ( cacheTimeout * 2 ) * .99 );
+//
+// when( graphFig.getShardMinDelta() ).thenReturn( tooSmallDelta );
+//
+//
+// 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 );
+//
+//
+// /**
+// * Should throw an exception
+// */
+// try {
+// approximation.getMinTime();
+// fail( "Should have thrown a GraphRuntimeException" );
+// }
+// catch ( GraphRuntimeException gre ) {
+// //swallow
+// }
+//
+// //now test something that passes.
+//
+// final long minDelta = cacheTimeout * 2;
+//
+// when( graphFig.getShardMinDelta() ).thenReturn( minDelta );
+//
+// long returned = approximation.getMinTime();
+//
+// long expectedReturned = time - minDelta;
+//
+// assertEquals( expectedReturned, returned );
+//
+// final long delta = cacheTimeout * 4;
+//
+// when( graphFig.getShardMinDelta() ).thenReturn( delta );
+//
+// returned = approximation.getMinTime();
+//
+// expectedReturned = time - delta;
+//
+// assertEquals( expectedReturned, returned );
+// }
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/cd043b14/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardEntryGroupTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardEntryGroupTest.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardEntryGroupTest.java
index 4e23d83..4df84f5 100644
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardEntryGroupTest.java
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardEntryGroupTest.java
@@ -21,8 +21,9 @@ package org.apache.usergrid.persistence.graph.serialization.impl.shard;
import org.junit.Test;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
@@ -36,7 +37,7 @@ public class ShardEntryGroupTest {
final long delta = 10000;
- Shard rootShard = new Shard( 0, 0 );
+ Shard rootShard = new Shard( 0, 0, false );
ShardEntryGroup shardEntryGroup = new ShardEntryGroup( delta );
@@ -46,9 +47,9 @@ public class ShardEntryGroupTest {
assertFalse( "Single shard cannot be deleted", shardEntryGroup.canBeDeleted( rootShard ) );
- assertSame( "Same shard for merge target", rootShard, shardEntryGroup.getMergeTarget() );
+ assertNull( "No merge target found", shardEntryGroup.getCompactionTarget() );
- assertFalse( "Merge cannot be run with a single shard", shardEntryGroup.needsCompaction( 0 ) );
+ assertFalse( "Merge cannot be run with a single shard", shardEntryGroup.shouldCompact( Long.MAX_VALUE ) );
}
@@ -57,22 +58,130 @@ public class ShardEntryGroupTest {
final long delta = 10000;
- Shard firstShard = new Shard( 1000, 1000 );
+ Shard firstShard = new Shard( 1000, 1000, false );
- Shard secondShard = new Shard( 1000, 1000 );
+ Shard secondShard = new Shard( 1000, 1001, false );
ShardEntryGroup shardEntryGroup = new ShardEntryGroup( delta );
- final boolean result = shardEntryGroup.addShard( rootShard );
+ boolean result = shardEntryGroup.addShard( firstShard );
assertTrue( "Shard added", result );
- assertFalse( "Single shard cannot be deleted", shardEntryGroup.canBeDeleted( rootShard ) );
+ result = shardEntryGroup.addShard( secondShard );
+
+ assertTrue( " Shard added", result );
+
+
+ assertFalse( "Root shard cannot be deleted", shardEntryGroup.canBeDeleted( firstShard ) );
+
+ assertFalse( "Root shard cannot be deleted", shardEntryGroup.canBeDeleted( secondShard ) );
+
+ assertFalse( "Duplicate shard id cannot be deleted", shardEntryGroup.canBeDeleted( secondShard ) );
+
+ assertEquals( "Same shard for merge target", firstShard, shardEntryGroup.getCompactionTarget() );
+
+ //shouldn't return true, since we haven't passed delta time in the second shard
+ assertFalse( "Merge cannot be run within min time",
+ shardEntryGroup.shouldCompact( firstShard.getCreatedTime() + delta ) );
+
+ //shouldn't return true, since we haven't passed delta time in the second shard
+ assertFalse( "Merge cannot be run within min time",
+ shardEntryGroup.shouldCompact( secondShard.getCreatedTime() + delta ) );
+
+ assertTrue( "Merge should be run with after min time",
+ shardEntryGroup.shouldCompact( secondShard.getCreatedTime() + delta + 1 ) );
+ }
+
+
+ @Test
+ public void multipleShardGroups() {
+
+ final long delta = 10000;
+
+ Shard firstShard = new Shard( 1000, 10000, false );
+
+ Shard secondShard = new Shard( 999, 9000, false );
+
+ Shard compactedShard1 = new Shard( 900, 8000, true );
+
+ Shard compactedShard2 = new Shard( 800, 7000, true );
+
+
+ ShardEntryGroup shardEntryGroup = new ShardEntryGroup( delta );
+
+ boolean result = shardEntryGroup.addShard( firstShard );
+
+ assertTrue( "Shard added", result );
+
+ result = shardEntryGroup.addShard( secondShard );
+
+ assertTrue( " Shard added", result );
+
+ result = shardEntryGroup.addShard( compactedShard1 );
+
+ assertTrue( "Shard added", result );
+
+ result = shardEntryGroup.addShard( compactedShard2 );
+
+ assertFalse( "Shouldn't add since it's compacted", result );
+
+ ShardEntryGroup secondGroup = new ShardEntryGroup( delta );
+
+ result = secondGroup.addShard( compactedShard2 );
+
+ assertTrue( "Added successfully", result );
+ }
+
+
+ @Test
+ public void boundShardGroup() {
+
+ final long delta = 10000;
+
+ Shard firstShard = new Shard( 1000, 10000, false );
+
+ Shard secondShard = new Shard( 999, 9000, false );
+
+ Shard compactedShard1 = new Shard( 900, 8000, true );
+
+
+ ShardEntryGroup shardEntryGroup = new ShardEntryGroup( delta );
+
+ boolean result = shardEntryGroup.addShard( firstShard );
+
+ assertTrue( "Shard added", result );
+
+ result = shardEntryGroup.addShard( secondShard );
+
+ assertTrue( " Shard added", result );
+
+ result = shardEntryGroup.addShard( compactedShard1 );
+
+ assertTrue( "Shard added", result );
+
+
+
+ assertFalse( "Shard cannot be deleted", shardEntryGroup.canBeDeleted( firstShard ) );
+
+ assertFalse( "Root shard cannot be deleted", shardEntryGroup.canBeDeleted( secondShard ) );
+
+ assertFalse( "Duplicate shard id cannot be deleted", shardEntryGroup.canBeDeleted( secondShard ) );
+
+ assertEquals( "Same shard for merge target", firstShard, shardEntryGroup.getCompactionTarget() );
+
+ //shouldn't return true, since we haven't passed delta time in the second shard
+ assertFalse( "Merge cannot be run within min time",
+ shardEntryGroup.shouldCompact( firstShard.getCreatedTime() + delta ) );
+
+ //shouldn't return true, since we haven't passed delta time in the second shard
+ assertFalse( "Merge cannot be run within min time",
+ shardEntryGroup.shouldCompact( secondShard.getCreatedTime() + delta ) );
- assertSame( "Same shard for merge target", rootShard, shardEntryGroup.getMergeTarget() );
+ assertTrue( "Merge should be run with after min time",
+ shardEntryGroup.shouldCompact( secondShard.getCreatedTime() + delta + 1 ) );
- assertFalse( "Merge cannot be run with a single shard", shardEntryGroup.needsCompaction( 0 ) );
}
}