public AbortedTxnWithPosition next() {
                if (!queue.isEmpty()) {
                    AbortedTxnWithPosition item = queue.poll();
                    if (item.txn.version() > AbortedTxn.CURRENT_VERSION)
                        throw new KafkaException("Unexpected aborted transaction version " + item.txn.version()
                            + " in transaction index " + path + ", current version is "
                            + AbortedTxn.CURRENT_VERSION);
                    return item;
                }
                int endOffset = Math.min(position.value + AbortedTxn.TOTAL_SIZE * 128, endPosition);
                Optional<ByteBuf> cacheDataOpt = cache.get(cacheId, position.value, endOffset - position.value);
                ByteBuf buf;
                if (cacheDataOpt.isPresent()) {
                    buf = cacheDataOpt.get();
                } else {
                    FetchResult records = fetchStream(position.value, endOffset, endOffset - position.value);
                    ByteBuf txnListBuf = Unpooled.buffer(records.recordBatchList().size() * AbortedTxn.TOTAL_SIZE);
                    records.recordBatchList().forEach(r -> txnListBuf.writeBytes(r.rawPayload()));
                    cache.put(cacheId, position.value, txnListBuf);
                    records.free();
                    buf = txnListBuf;
                }
                while (buf.readableBytes() > 0) {
                    AbortedTxn abortedTxn = new AbortedTxn(buf.slice(buf.readerIndex(), AbortedTxn.TOTAL_SIZE).nioBuffer());
                    queue.add(new AbortedTxnWithPosition(abortedTxn, position.value));
                    position.value += AbortedTxn.TOTAL_SIZE;
                    buf.skipBytes(AbortedTxn.TOTAL_SIZE);
                }
                return queue.poll();
            }
Note that nextOffset in stream here actually represents the physical size (or position).
core/src/main/scala/kafka/log/streamaspect/ElasticTransactionIndex.java