public OptionalLong loadLastOffset() {
        if (stream.nextOffset() == 0) {
            return OptionalLong.empty();
        } else {
            long nextOffset = stream.nextOffset();
            FetchResult rst = fetchStream(nextOffset - 1, nextOffset, Integer.MAX_VALUE);
            try {
                RecordBatchWithContext record = rst.recordBatchList().get(0);
                ByteBuffer readBuf = record.rawPayload();
                AbortedTxn abortedTxn = new AbortedTxn(readBuf);
                return OptionalLong.of(abortedTxn.lastOffset());
            } finally {
                rst.free();
            }
        }
    }
In the case of unclean shutdown, the last entry needs to be recovered from the txn index.
core/src/main/scala/kafka/log/streamaspect/ElasticTransactionIndex.java