public Optional<LogOffsetMetadata> firstUnstableOffset() {
        Optional<LogOffsetMetadata> unreplicatedFirstOffset = Optional.ofNullable(unreplicatedTxns.firstEntry()).map(e -> e.getValue().firstOffset);
        Optional<LogOffsetMetadata> undecidedFirstOffset = Optional.ofNullable(ongoingTxns.firstEntry()).map(e -> e.getValue().firstOffset);

        if (!unreplicatedFirstOffset.isPresent())
            return undecidedFirstOffset;
        else if (!undecidedFirstOffset.isPresent())
            return unreplicatedFirstOffset;
        else if (undecidedFirstOffset.get().messageOffset < unreplicatedFirstOffset.get().messageOffset)
            return undecidedFirstOffset;
        else
            return unreplicatedFirstOffset;
    }
@Test
    public void testFirstUnstableOffset() {
        short epoch = 5;

        assertEquals(OptionalLong.empty(), stateManager.firstUndecidedOffset());

        appendClientEntry(stateManager, producerId, epoch, defaultSequence, 99, true);
        assertEquals(OptionalLong.of(99L), stateManager.firstUndecidedOffset());
        assertEquals(Optional.of(99L), stateManager.firstUnstableOffset().map(x -> x.messageOffset));

        long anotherPid = 2L;
        appendClientEntry(stateManager, anotherPid, epoch, defaultSequence, 105, true);
        assertEquals(OptionalLong.of(99L), stateManager.firstUndecidedOffset());
        assertEquals(Optional.of(99L), stateManager.firstUnstableOffset().map(x -> x.messageOffset));

        appendEndTxnMarker(stateManager, producerId, epoch, ControlRecordType.COMMIT, 109);
        assertEquals(OptionalLong.of(105L), stateManager.firstUndecidedOffset());
        assertEquals(Optional.of(99L), stateManager.firstUnstableOffset().map(x -> x.messageOffset));

        stateManager.onHighWatermarkUpdated(100L);
        assertEquals(Optional.of(99L), stateManager.firstUnstableOffset().map(x -> x.messageOffset));

        stateManager.onHighWatermarkUpdated(110L);
        assertEquals(Optional.of(105L), stateManager.firstUnstableOffset().map(x -> x.messageOffset));

        appendEndTxnMarker(stateManager, anotherPid, epoch, ControlRecordType.ABORT, 112);
        assertFalse(stateManager.firstUndecidedOffset().isPresent());
        assertEquals(Optional.of(105L), stateManager.firstUnstableOffset().map(x -> x.messageOffset));

        stateManager.onHighWatermarkUpdated(113L);
        assertFalse(stateManager.firstUnstableOffset().map(x -> x.messageOffset).isPresent());
    }