Сортировка большого файла без занесения в оперативную память

Мне надо написать свою реализацию сортировки файла большого объема. По условию я немогу целиком записать файл в оперативную память. Файл представлят собой бинарное представление 16-битных чисел. Саму сортировку и обработку файлов я сделал, но столкнулся с проблемой. Если выставить xmx размером с файл, то моя программа валится с ООМ. Не понимаю в чем дело. Ведь я читаю из потока, а не целиком файл.

Вот код класса с сортировкой:

public class FileSorter {

    private final String directory;
    private final String fileName;
    private final Sorter sorter;
    private final ExecutorService pool = new ForkJoinPool(Runtime.getRuntime().availableProcessors() / 2);
    private AtomicInteger fileCounter = new AtomicInteger(0);

    public FileSorter(String directory, String fileName, Sorter sorter) {
        this.directory = directory;
        this.fileName = fileName;
        this.sorter = sorter;
    }

    public String sort(int chunkSize) throws IOException {
        int files = splitFile(chunkSize);
        ArrayList<String> fileNames = new FactorList<>();
        for (int i = 0; i < files; i++) {
            fileNames.add(String.valueOf(i));
        }
        return mergeAllFiles(fileNames);
    }


    public void clearOutputDirectory() throws IOException {
        try {
            Path path = Paths.get(directory + File.separator + "output");
            Files.walk(path)
                .map(Path::toFile)
                .forEach(File::delete);
            Files.delete(path);
        } catch (NoSuchFileException e) {
            System.out.println("This directory is not exists");
        }
    }

    private String mergeAllFiles(ArrayList<String> fileNames) {
        pool.submit(() -> System.out.println("Handling " + fileNames.size() + " files"));
        if (fileNames.size() == 1) {
            return fileNames.get(0);
        }
        ArrayList<String> sortedFiles = new FactorList<>();
        int middle = fileNames.size() / 2 + fileNames.size() % 2;
        for (int i = 0; i < middle; i++) {
            String first = fileNames.get(i);
            String second = fileNames.get(i + middle);

            String file = mergeFiles(first, second);
            sortedFiles.add(file);

            asyncDeleteFile(file, first);
            asyncDeleteFile(file, second);
        }
        return mergeAllFiles(sortedFiles);
    }

    private void asyncDeleteFile(String newFile, String oldFile) {
        if (oldFile != null) {
            pool.submit(() -> {
                try {
                    if (!newFile.equals(oldFile)) {
                        Files.delete(Paths.get(getOutputFilePath(oldFile)));
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });
        }
    }

    private String mergeFiles(String firstFile, String secondFile) {
        if (firstFile == null) {
            return secondFile;
        } else if (secondFile == null) {
            return firstFile;
        }

        String resultFile = String.valueOf(fileCounter.getAndIncrement());
        try (File16BitReader firstReader = new File16BitReader(getOutputFilePath(firstFile));
             File16BitReader secondReader = new File16BitReader(getOutputFilePath(secondFile));
             File16BitWriter writer = new File16BitWriter(getOutputFilePath(resultFile))) {

            LessValueResolver resolver = new LessValueResolver(firstReader, secondReader);
            while (resolver.isAvailable()) {
                writer.write(resolver.getLessValue());
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
        return resultFile;
    }

    private String getOutputFilePath(String fileName) {
        return directory + File.separator + "output" + File.separator + fileName;
    }

    private int splitFile(int chunkSize) throws IOException {
        String path = directory + File.separator + fileName;
        Files.createDirectory(Paths.get(directory + File.separator + "output"));
        try (File16BitReader reader = new File16BitReader(path)) {
            while (reader.isAvailable()) {
                int[] chunk = readChunk(reader, chunkSize);
                String path1 = directory + File.separator + "output" + File.separator + fileCounter.getAndIncrement();
                sortAndWrite(chunk, path1);
            }
        }
        return fileCounter.get();
    }

    private void sortAndWrite(int[] chunk, String path) {
        sorter.sort(chunk);
        try (File16BitWriter writer = new File16BitWriter(path)) {
            for (int element : chunk) {
                writer.write(element);
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private int[] readChunk(File16BitReader reader, int chunkSize) throws IOException {
        int[] chunk = new int[chunkSize];
        int numberCounter = 0;
        while (numberCounter < chunkSize) {
            int number = reader.read();
            if (number >= 0) {
                chunk[numberCounter] = number;
                numberCounter++;
            } else {
                break;
            }
        }
        if (numberCounter < chunkSize - 1) {
            int[] resizedChunk = new int[numberCounter];
            System.arraycopy(chunk, 0, resizedChunk, 0, numberCounter);
            chunk = resizedChunk;
        }

        return chunk;
    }
}

Класс для быстрой записи в файл:

public class File16BitWriter implements Closeable {

    private final int bufferSize = 8 * 1024;
    private final AsynchronousFileChannel channel;

    private long position = 0;
    private ByteBuffer buffer = ByteBuffer.allocate(bufferSize);
    private Future<Integer> futureWriting = new FutureTask<Integer>(() -> 0) {{ run(); }}; // stub (start point)

    public File16BitWriter(String filePath) throws IOException {
        Path path = Paths.get(filePath);
        if (Files.exists(path)) {
            Files.delete(path);
        }
        Files.createFile(path);
        channel = AsynchronousFileChannel.open(path, StandardOpenOption.WRITE);
    }

    public void write(int i) {
        byte right = (byte) (0xFF & i);
        byte left = (byte) (i >> 8);
        buffer.put(left);
        buffer.put(right);

        if (buffer.position() == bufferSize) {
            writeBufferToFile();
            buffer = ByteBuffer.allocate(bufferSize);
        }
    }

    private void writeBufferToFile() {
        int pos = buffer.position();
        try {
            futureWriting.get();
        } catch (ExecutionException | InterruptedException e) {
            e.printStackTrace();
        }
        buffer.flip();
        futureWriting = channel.write(buffer, position);
        position += pos;
    }

    @Override
    public void close() throws IOException {
        writeBufferToFile();
        try {
            futureWriting.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
        channel.close();
    }
}

Класс быстрого чтения из класса:

public class File16BitReader implements Closeable {

    private final int bufferSize = 8 * 1024;
    private AsynchronousFileChannel channel;
    private final long fileSize;
    private long totalReadCounter;

    private int readBytesCounter = 0;
    private byte[] alreadyReadBytes = new byte[0];
    private ByteBuffer buffer = ByteBuffer.allocate(bufferSize);
    private Future<Integer> futureWriting;

    private Path path;

    public File16BitReader(String filePath) throws IOException {
        path = Paths.get(filePath);
        fileSize = Files.size(Paths.get(filePath));
        channel = AsynchronousFileChannel.open(Paths.get(filePath), StandardOpenOption.READ);
        futureWriting = channel.read(buffer, 0);
    }

    public int read() {
        if (!isAvailable()) {
            return -1;
        }

        if (readBytesCounter == alreadyReadBytes.length) {
            updateAlreadyReadBytes();
        }
        byte left = getByte();
        byte right = getByte();

        return (((left & 0xFF) << 8) | (right & 0xFF));
    }

    public boolean isAvailable() {
        return totalReadCounter < fileSize;
    }

    private byte getByte() {
        totalReadCounter++;
        return alreadyReadBytes[readBytesCounter++];
    }

    private void updateAlreadyReadBytes() {
        try {
            futureWriting.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

        buffer.flip();
        alreadyReadBytes = new byte[buffer.limit()];
        buffer.get(alreadyReadBytes);
        buffer = ByteBuffer.allocate(bufferSize);
        futureWriting = channel.read(buffer, totalReadCounter + bufferSize);
        readBytesCounter = 0;
    }

    @Override
    public void close() throws IOException {
        channel.close();
    }
}

P.S. FactorList - это мой аналог ArrayList'а.