@Test public void testDeletion() throws ExecutionException, InterruptedException, ConnectionException { final long createTime = 10000; final long currentTime = createTime * 2; final Shard shard0 = new Shard( 1000, createTime, true ); final ShardEntryGroup group = new ShardEntryGroup( 1 ); group.addShard( shard0 ); assertFalse( "this should return false for our test to succeed", group.isCompactionPending() ); assertFalse( "this should return false for our test to succeed", group.isNew( currentTime ) ); final DirectedEdgeMeta directedEdgeMeta = getDirectedEdgeMeta(); final EdgeShardSerialization edgeShardSerialization = mock( EdgeShardSerialization.class ); final MutationBatch batch = mock( MutationBatch.class ); when( edgeShardSerialization.removeShardMeta( same( scope ), same( shard0 ), same( directedEdgeMeta ) ) ) .thenReturn( batch ); final TimeService timeService = mock( TimeService.class ); when( timeService.getCurrentTime() ).thenReturn( currentTime ); initExecutor( 1, 1 ); final ShardGroupDeletionImpl shardGroupDeletion = new ShardGroupDeletionImpl( asyncTaskExecutor, edgeShardSerialization, timeService ); final ListenableFuture<ShardGroupDeletion.DeleteResult> future = shardGroupDeletion.maybeDeleteShard( this.scope, directedEdgeMeta, group, Collections.emptyIterator() ); final ShardGroupDeletion.DeleteResult result = future.get(); assertEquals( "should delete", ShardGroupDeletion.DeleteResult.DELETED, result ); verify(batch).execute(); }
@Override public ListenableFuture<DeleteResult> maybeDeleteShard( final ApplicationScope applicationScope, final DirectedEdgeMeta directedEdgeMeta, final ShardEntryGroup shardEntryGroup, final Iterator<MarkedEdge> edgeIterator ) { final ListenableFuture<DeleteResult> future; try { future = asyncTaskExecutor .submit( new ShardDeleteTask( applicationScope, directedEdgeMeta, shardEntryGroup, edgeIterator ) ); } catch ( RejectedExecutionException ree ) { logger.info( "Rejected shard delete check for group {}", edgeIterator ); return Futures.immediateFuture( DeleteResult.NOT_CHECKED ); } Futures.addCallback( future, new FutureCallback<DeleteResult>() { @Override public void onSuccess( @Nullable final ShardGroupDeletion.DeleteResult result ) { if (logger.isTraceEnabled()) logger.trace( "Successfully completed delete of task {}", result ); } @Override public void onFailure( final Throwable t ) { logger.error( "Unable to perform shard delete audit. Exception is ", t ); } } ); return future; }
ShardGroupDeletionImpl implements ShardGroupDeletion { @Override public ListenableFuture<DeleteResult> maybeDeleteShard( final ApplicationScope applicationScope, final DirectedEdgeMeta directedEdgeMeta, final ShardEntryGroup shardEntryGroup, final Iterator<MarkedEdge> edgeIterator ) { final ListenableFuture<DeleteResult> future; try { future = asyncTaskExecutor .submit( new ShardDeleteTask( applicationScope, directedEdgeMeta, shardEntryGroup, edgeIterator ) ); } catch ( RejectedExecutionException ree ) { logger.info( "Rejected shard delete check for group {}", edgeIterator ); return Futures.immediateFuture( DeleteResult.NOT_CHECKED ); } Futures.addCallback( future, new FutureCallback<DeleteResult>() { @Override public void onSuccess( @Nullable final ShardGroupDeletion.DeleteResult result ) { if (logger.isTraceEnabled()) logger.trace( "Successfully completed delete of task {}", result ); } @Override public void onFailure( final Throwable t ) { logger.error( "Unable to perform shard delete audit. Exception is ", t ); } } ); return future; } }
ShardGroupDeletionImpl implements ShardGroupDeletion { @Override public ListenableFuture<DeleteResult> maybeDeleteShard( final ApplicationScope applicationScope, final DirectedEdgeMeta directedEdgeMeta, final ShardEntryGroup shardEntryGroup, final Iterator<MarkedEdge> edgeIterator ) { final ListenableFuture<DeleteResult> future; try { future = asyncTaskExecutor .submit( new ShardDeleteTask( applicationScope, directedEdgeMeta, shardEntryGroup, edgeIterator ) ); } catch ( RejectedExecutionException ree ) { logger.info( "Rejected shard delete check for group {}", edgeIterator ); return Futures.immediateFuture( DeleteResult.NOT_CHECKED ); } Futures.addCallback( future, new FutureCallback<DeleteResult>() { @Override public void onSuccess( @Nullable final ShardGroupDeletion.DeleteResult result ) { if (logger.isTraceEnabled()) logger.trace( "Successfully completed delete of task {}", result ); } @Override public void onFailure( final Throwable t ) { logger.error( "Unable to perform shard delete audit. Exception is ", t ); } } ); return future; } @Inject  ShardGroupDeletionImpl( final AsyncTaskExecutor asyncTaskExecutor,
                                   final EdgeShardSerialization edgeShardSerialization,
                                   final TimeService timeService ); }
ShardGroupDeletionImpl implements ShardGroupDeletion { @Override public ListenableFuture<DeleteResult> maybeDeleteShard( final ApplicationScope applicationScope, final DirectedEdgeMeta directedEdgeMeta, final ShardEntryGroup shardEntryGroup, final Iterator<MarkedEdge> edgeIterator ) { final ListenableFuture<DeleteResult> future; try { future = asyncTaskExecutor .submit( new ShardDeleteTask( applicationScope, directedEdgeMeta, shardEntryGroup, edgeIterator ) ); } catch ( RejectedExecutionException ree ) { logger.info( "Rejected shard delete check for group {}", edgeIterator ); return Futures.immediateFuture( DeleteResult.NOT_CHECKED ); } Futures.addCallback( future, new FutureCallback<DeleteResult>() { @Override public void onSuccess( @Nullable final ShardGroupDeletion.DeleteResult result ) { if (logger.isTraceEnabled()) logger.trace( "Successfully completed delete of task {}", result ); } @Override public void onFailure( final Throwable t ) { logger.error( "Unable to perform shard delete audit. Exception is ", t ); } } ); return future; } @Inject  ShardGroupDeletionImpl( final AsyncTaskExecutor asyncTaskExecutor,
                                   final EdgeShardSerialization edgeShardSerialization,
                                   final TimeService timeService ); @Override ListenableFuture<DeleteResult> maybeDeleteShard( final ApplicationScope applicationScope,
                                                            final DirectedEdgeMeta directedEdgeMeta,
                                                            final ShardEntryGroup shardEntryGroup,
                                                            final Iterator<MarkedEdge> edgeIterator ); }
ShardGroupDeletionImpl implements ShardGroupDeletion { @Override public ListenableFuture<DeleteResult> maybeDeleteShard( final ApplicationScope applicationScope, final DirectedEdgeMeta directedEdgeMeta, final ShardEntryGroup shardEntryGroup, final Iterator<MarkedEdge> edgeIterator ) { final ListenableFuture<DeleteResult> future; try { future = asyncTaskExecutor .submit( new ShardDeleteTask( applicationScope, directedEdgeMeta, shardEntryGroup, edgeIterator ) ); } catch ( RejectedExecutionException ree ) { logger.info( "Rejected shard delete check for group {}", edgeIterator ); return Futures.immediateFuture( DeleteResult.NOT_CHECKED ); } Futures.addCallback( future, new FutureCallback<DeleteResult>() { @Override public void onSuccess( @Nullable final ShardGroupDeletion.DeleteResult result ) { if (logger.isTraceEnabled()) logger.trace( "Successfully completed delete of task {}", result ); } @Override public void onFailure( final Throwable t ) { logger.error( "Unable to perform shard delete audit. Exception is ", t ); } } ); return future; } @Inject  ShardGroupDeletionImpl( final AsyncTaskExecutor asyncTaskExecutor,
                                   final EdgeShardSerialization edgeShardSerialization,
                                   final TimeService timeService ); @Override ListenableFuture<DeleteResult> maybeDeleteShard( final ApplicationScope applicationScope,
                                                            final DirectedEdgeMeta directedEdgeMeta,
                                                            final ShardEntryGroup shardEntryGroup,
                                                            final Iterator<MarkedEdge> edgeIterator );  }