class BlockStorage implements Closeable {

    public void merge( int mergeFactor, Cancellation cancellation ) throws IOException
    {
        monitor.mergeStarted( entryCount, calculateNumberOfEntriesWrittenDuringMerges( entryCount, numberOfBlocksInCurrentFile, mergeFactor ) );
        File sourceFile = blockFile;
        File tempFile = new File( blockFile.getParent(), blockFile.getName() + ".b" );
        try ( Allocator mergeBufferAllocator = bufferFactory.newLocalAllocator() )
        {
            File targetFile = tempFile;

            // Allocate all buffers that will be used and reused for all merge iterations
            ByteBuffer writeBuffer = mergeBufferAllocator.allocate( bufferFactory.bufferSize() );
            ByteBuffer[] readBuffers = new ByteBuffer[mergeFactor];
            for ( int i = 0; i < readBuffers.length; i++ )
            {
                readBuffers[i] = mergeBufferAllocator.allocate( bufferFactory.bufferSize() );
            }

            while ( numberOfBlocksInCurrentFile > 1 )
            {
                // Perform one complete merge iteration, merging all blocks from source into target.
                // After this step, target will contain fewer blocks than source, but may need another merge iteration.
                try ( BlockReader<KEY,VALUE> reader = reader( sourceFile );
                      StoreChannel targetChannel = fs.open( targetFile, OpenMode.READ_WRITE ) )
                {
                    long blocksMergedSoFar = 0;
                    long blocksInMergedFile = 0;
                    while ( !cancellation.cancelled() && blocksMergedSoFar < numberOfBlocksInCurrentFile )
                    {
                        blocksMergedSoFar += performSingleMerge( mergeFactor, reader, targetChannel, cancellation, readBuffers, writeBuffer );
                        blocksInMergedFile++;
                    }
                    numberOfBlocksInCurrentFile = blocksInMergedFile;
                    monitor.mergeIterationFinished( blocksMergedSoFar, blocksInMergedFile );
                }

                // Flip and restore the channels
                File tmpSourceFile = sourceFile;
                sourceFile = targetFile;
                targetFile = tmpSourceFile;
            }
        }
        finally
        {
            if ( sourceFile == blockFile )
            {
                fs.deleteFile( tempFile );
            }
            else
            {
                fs.deleteFile( blockFile );
                fs.renameFile( tempFile, blockFile );
            }
        }
    }

      BlockStorage( Layout<KEY,VALUE> layout, ByteBufferFactory bufferFactory, FileSystemAbstraction fs, File blockFile, Monitor monitor );

}