### Question:
EncodeGroupsStep extends ProcessorStep<RelationshipGroupRecord[]> { @Override protected void process( RelationshipGroupRecord[] batch, BatchSender sender ) { int groupStartIndex = 0; for ( int i = 0; i < batch.length; i++ ) { RelationshipGroupRecord group = batch[i]; long count = group.getNext(); boolean lastInChain = count == 0; group.setId( nextId == -1 ? nextId = store.nextId() : nextId ); if ( !lastInChain ) { group.setNext( nextId = store.nextId() ); } else { group.setNext( nextId = -1 ); for ( int j = groupStartIndex; j <= i; j++ ) { store.prepareForCommit( batch[j] ); } groupStartIndex = i + 1; } } assert groupStartIndex == batch.length; sender.send( batch ); } EncodeGroupsStep( StageControl control, Configuration config, RecordStore<RelationshipGroupRecord> store );   }### Answer:
@SuppressWarnings( "unchecked" ) @Test public void shouldEncodeGroupChains() throws Throwable { StageControl control = mock( StageControl.class ); final AtomicLong nextId = new AtomicLong(); RecordStore<RelationshipGroupRecord> store = mock( RecordStore.class ); when( store.nextId() ).thenAnswer( invocation -> nextId.incrementAndGet() ); doAnswer( invocation -> { invocation.<RelationshipGroupRecord>getArgument( 0 ).setFirstOut( 1 ); return null; } ).when( store ).prepareForCommit( any( RelationshipGroupRecord.class ) ); Configuration config = Configuration.withBatchSize( DEFAULT, 10 ); EncodeGroupsStep encoder = new EncodeGroupsStep( control, config, store ); encoder.start( Step.ORDER_SEND_DOWNSTREAM ); Catcher catcher = new Catcher(); encoder.process( batch( new Group( 1, 3 ), new Group( 2, 3 ), new Group( 3, 4 ) ), catcher ); encoder.process( batch( new Group( 4, 2 ), new Group( 5, 10 ) ), catcher ); encoder.process( batch( new Group( 6, 35 ) ), catcher ); encoder.process( batch( new Group( 7, 2 ) ), catcher ); encoder.endOfUpstream(); encoder.awaitCompleted(); encoder.close(); assertEquals( 4, catcher.batches.size() ); long lastOwningNodeLastBatch = -1; for ( RelationshipGroupRecord[] batch : catcher.batches ) { assertBatch( batch, lastOwningNodeLastBatch ); lastOwningNodeLastBatch = batch[batch.length - 1].getOwningNode(); } }