public int boundedControlledPoll(
        final ControlledFragmentHandler handler, final long limitPosition, final int fragmentLimit)
    {
        if (isClosed)
        {
            return 0;
        }

        long initialPosition = subscriberPosition.get();
        if (initialPosition >= limitPosition)
        {
            return 0;
        }

        int fragmentsRead = 0;
        int initialOffset = (int)initialPosition & termLengthMask;
        int offset = initialOffset;
        final UnsafeBuffer termBuffer = activeTermBuffer(initialPosition);
        final int limitOffset = (int)Math.min(termBuffer.capacity(), (limitPosition - initialPosition) + offset);
        final Header header = this.header;
        header.buffer(termBuffer);

        try
        {
            while (fragmentsRead < fragmentLimit && offset < limitOffset)
            {
                final int length = frameLengthVolatile(termBuffer, offset);
                if (length <= 0)
                {
                    break;
                }

                final int frameOffset = offset;
                final int alignedLength = BitUtil.align(length, FRAME_ALIGNMENT);
                offset += alignedLength;

                if (isPaddingFrame(termBuffer, frameOffset))
                {
                    continue;
                }

                ++fragmentsRead;
                header.offset(frameOffset);

                final Action action = handler.onFragment(
                    termBuffer, frameOffset + HEADER_LENGTH, length - HEADER_LENGTH, header);

                if (ABORT == action)
                {
                    --fragmentsRead;
                    offset -= alignedLength;
                    break;
                }

                if (BREAK == action)
                {
                    break;
                }

                if (COMMIT == action)
                {
                    initialPosition += (offset - initialOffset);
                    initialOffset = offset;
                    subscriberPosition.setOrdered(initialPosition);
                }
            }
        }
        catch (final Exception ex)
        {
            errorHandler.onError(ex);
        }
        finally
        {
            final long resultingPosition = initialPosition + (offset - initialOffset);
            if (resultingPosition > initialPosition)
            {
                subscriberPosition.setOrdered(resultingPosition);
            }
        }

        return fragmentsRead;
    }
@Test
    void shouldPollFragmentsToBoundedControlledFragmentHandlerWithInitialOffsetNotZero()
    {
        final long initialPosition = computePosition(
            INITIAL_TERM_ID, offsetForFrame(1), POSITION_BITS_TO_SHIFT, INITIAL_TERM_ID);
        final long maxPosition = initialPosition + ALIGNED_FRAME_LENGTH;
        position.setOrdered(initialPosition);
        final Image image = createImage();

        insertDataFrame(INITIAL_TERM_ID, offsetForFrame(1));
        insertDataFrame(INITIAL_TERM_ID, offsetForFrame(2));

        when(mockControlledFragmentHandler.onFragment(any(DirectBuffer.class), anyInt(), anyInt(), any(Header.class)))
            .thenReturn(Action.CONTINUE);

        final int fragmentsRead = image.boundedControlledPoll(
            mockControlledFragmentHandler, maxPosition, Integer.MAX_VALUE);

        assertThat(fragmentsRead, is(1));

        assertThat(position.get(), is(maxPosition));
        verify(mockControlledFragmentHandler).onFragment(
            any(UnsafeBuffer.class), anyInt(), anyInt(), any(Header.class));
    }