@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 testMergeOfTwoStreams() throws Exception {
        // iterators
        List<MutableObjectIterator<Tuple2<Integer, String>>> iterators = new ArrayList<>();
        iterators.add(
                newIterator(new int[] {1, 2, 4, 5, 10}, new String[] {"1", "2", "4", "5", "10"}));
        iterators.add(
                newIterator(new int[] {3, 6, 7, 10, 12}, new String[] {"3", "6", "7", "10", "12"}));

        final int[] expected = new int[] {1, 2, 3, 4, 5, 6, 7, 10, 10, 12};

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

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

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

        int pos = 1;

        assertThat(rec1 = iterator.next(rec1)).isNotNull();
        assertThat(rec1.f0).isEqualTo(expected[0]);

        while ((rec2 = iterator.next(rec2)) != null) {
            k1 = rec1.f0;
            k2 = rec2.f0;

            assertThat(comparator.compare(k1, k2)).isLessThanOrEqualTo(0);
            assertThat(k2).isEqualTo(expected[pos++]);

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