@Override
    public E next(E reuse) throws IOException {
        /* There are three ways to handle object reuse:
         * 1) reuse and return the given object
         * 2) ignore the given object and return a new object
         * 3) exchange the given object for an existing object
         *
         * The first option is not available here as the return value has
         * already been deserialized from the heap's top iterator. The second
         * option avoids object reuse. The third option is implemented below
         * by passing the given object to the heap's top iterator into which
         * the next value will be deserialized.
         */

        if (this.heap.size() > 0) {
            // get the smallest element
            final HeadStream<E> top = this.heap.peek();
            E result = top.getHead();

            // read an element
            if (!top.nextHead(reuse)) {
                this.heap.poll();
            } else {
                this.heap.adjustTop();
            }
            return result;
        } else {
            return null;
        }
    }
@Test
    void testInvalidMerge() throws Exception {
        // iterators
        List<MutableObjectIterator<Tuple2<Integer, String>>> iterators = new ArrayList<>();
        iterators.add(
                newIterator(new int[] {1, 2, 17, 23, 23}, new String[] {"A", "B", "C", "D", "E"}));
        iterators.add(
                newIterator(new int[] {2, 6, 7, 8, 9}, new String[] {"A", "B", "C", "D", "E"}));
        iterators.add(
                newIterator(new int[] {4, 10, 11, 11, 12}, new String[] {"A", "B", "C", "D", "E"}));
        iterators.add(
                newIterator(new int[] {3, 6, 10, 7, 12}, new String[] {"A", "B", "C", "D", "E"}));
        iterators.add(
                newIterator(new int[] {7, 10, 15, 19, 44}, new String[] {"A", "B", "C", "D", "E"}));
        iterators.add(
                newIterator(new int[] {6, 6, 11, 17, 18}, new String[] {"A", "B", "C", "D", "E"}));
        iterators.add(
                newIterator(new int[] {1, 2, 4, 5, 10}, new String[] {"A", "B", "C", "D", "E"}));
        iterators.add(
                newIterator(new int[] {5, 10, 19, 23, 29}, new String[] {"A", "B", "C", "D", "E"}));
        iterators.add(
                newIterator(new int[] {9, 9, 9, 9, 9}, new String[] {"A", "B", "C", "D", "E"}));
        iterators.add(
                newIterator(new int[] {8, 8, 14, 14, 15}, new String[] {"A", "B", "C", "D", "E"}));

        // comparator
        TypeComparator<Integer> comparator = new IntComparator(true);

        // merge iterator
        MutableObjectIterator<Tuple2<Integer, String>> iterator =
                new MergeIterator<>(iterators, this.comparator);

        boolean violationFound = false;

        // check expected order
        Tuple2<Integer, String> rec1 = new Tuple2<>();
        Tuple2<Integer, String> rec2 = new Tuple2<>();

        assertThat(rec1 = iterator.next(rec1)).isNotNull();
        while ((rec2 = iterator.next(rec2)) != null) {
            if (comparator.compare(rec1.f0, rec2.f0) > 0) {
                violationFound = true;
                break;
            }

            Tuple2<Integer, String> tmp = rec1;
            rec1 = rec2;
            rec2 = tmp;
        }

        assertThat(violationFound)
                .withFailMessage("Merge must have returned a wrong result")
                .isTrue();
    }