@Test
    public void testRecoverTransactionIndex() throws IOException {
        ElasticLogSegment segment = createOrLoadSegment(100);
        short producerEpoch = (short) 0;
        int partitionLeaderEpoch = 15;
        int sequence = 100;

        long pid1 = 5L;
        long pid2 = 10L;

        // append transactional records from pid1
        segment.append(101L, RecordBatch.NO_TIMESTAMP,
            100L, MemoryRecords.withTransactionalRecords(100L, Compression.NONE,
                pid1, producerEpoch, sequence, partitionLeaderEpoch, new SimpleRecord("a".getBytes()), new SimpleRecord("b".getBytes())));

        // append transactional records from pid2
        segment.append(103L, RecordBatch.NO_TIMESTAMP, 102L, MemoryRecords.withTransactionalRecords(102L, Compression.NONE,
            pid2, producerEpoch, sequence, partitionLeaderEpoch, new SimpleRecord("a".getBytes()), new SimpleRecord("b".getBytes())));

        // append non-transactional records
        segment.append(105L, RecordBatch.NO_TIMESTAMP, 104L, MemoryRecords.withRecords(104L, Compression.NONE,
            partitionLeaderEpoch, new SimpleRecord("a".getBytes()), new SimpleRecord("b".getBytes())));

        // abort the transaction from pid2 (note LSO should be 100L since the txn from pid1 has not completed)
        segment.append(106L, RecordBatch.NO_TIMESTAMP,
            106L, endTxnRecords(ControlRecordType.ABORT, pid2, producerEpoch, 106L));

        // commit the transaction from pid1
        segment.append(107L, RecordBatch.NO_TIMESTAMP,
            107L, endTxnRecords(ControlRecordType.COMMIT, pid1, producerEpoch, 107L));

        ProducerStateManager stateManager = newProducerStateManager();
        segment.recover(stateManager, Optional.empty());
        assertEquals(108L, stateManager.mapEndOffset());

        var abortedTxns = segment.txnIndex().allAbortedTxns();
        assertEquals(1, abortedTxns.size());
        var abortedTxn = abortedTxns.get(0);
        assertEquals(pid2, abortedTxn.producerId());
        assertEquals(102L, abortedTxn.firstOffset());
        assertEquals(106L, abortedTxn.lastOffset());
        assertEquals(100L, abortedTxn.lastStableOffset());

        // recover again, but this time assuming the transaction from pid2 began on a previous segment
        // TODO: simulate this case
        // the elastic log segment instance cannot be repeated recover, so we need to reset the index to support it.
//        segment.timeIndex.reset()
//        segment.txnIndex.reset()
//        stateManager = newProducerStateManager()
//        stateManager.loadProducerEntry(new ProducerStateEntry(pid2,
//            mutable.Queue[BatchMetadata](BatchMetadata(10, 10L, 5, RecordBatch.NO_TIMESTAMP)), producerEpoch,
//            0, RecordBatch.NO_TIMESTAMP, Some(75L)))
//        segment.recover(stateManager)
//        assertEquals(108L, stateManager.mapEndOffset)
//
//        abortedTxns = segment.txnIndex.allAbortedTxns
//        assertEquals(1, abortedTxns.size)
//        abortedTxn = abortedTxns.head
//        assertEquals(pid2, abortedTxn.producerId)
//        assertEquals(75L, abortedTxn.firstOffset)
//        assertEquals(106L, abortedTxn.lastOffset)
//        assertEquals(100L, abortedTxn.lastStableOffset)
    }
This test was intended to test truncation and rolling in LogSegment. However, truncation is not supported in
ElasticLogSegment. Therefore, we cut this test down to just verify the rolling and leave the test name as is.
core/src/test/java/kafka/log/streamaspect/ElasticLogSegmentTest.java