class 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];

            // The iterator over the groups will not produce real next pointers, they are instead
            // a count meaning how many groups come after it. This encoder will set the real group ids.
            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 );

                // OK so this group is the last in this chain, which means all the groups in this chain
                // are now fully populated. We can now prepare these groups so that their potential
                // secondary units ends up very close by.
                for ( int j = groupStartIndex; j <= i; j++ )
                {
                    store.prepareForCommit( batch[j] );
                }

                groupStartIndex = i + 1;
            }
        }
        assert groupStartIndex == batch.length;

        sender.send( batch );
    }

    public  EncodeGroupsStep( StageControl control, Configuration config, RecordStore<RelationshipGroupRecord> store );

}