protected Iterable<AbortedTxnWithPosition> iterable(Supplier<ByteBuffer> allocate) {
        FileChannel channel = channelOrNull();
        if (channel == null)
            return Collections.emptyList();

        PrimitiveRef.IntRef position = PrimitiveRef.ofInt(0);

        return () -> new Iterator<AbortedTxnWithPosition>() {

            @Override
            public boolean hasNext() {
                try {
                    return channel.position() - position.value >= AbortedTxn.TOTAL_SIZE;
                } catch (IOException e) {
                    throw new KafkaException("Failed read position from the transaction index " + file.getAbsolutePath(), e);
                }
            }

            @Override
            public AbortedTxnWithPosition next() {
                try {
                    ByteBuffer buffer = allocate.get();
                    Utils.readFully(channel, buffer, position.value);
                    buffer.flip();

                    AbortedTxn abortedTxn = new AbortedTxn(buffer);
                    if (abortedTxn.version() > AbortedTxn.CURRENT_VERSION)
                        throw new KafkaException("Unexpected aborted transaction version " + abortedTxn.version()
                            + " in transaction index " + file.getAbsolutePath() + ", current version is "
                            + AbortedTxn.CURRENT_VERSION);
                    AbortedTxnWithPosition nextEntry = new AbortedTxnWithPosition(abortedTxn, position.value);
                    position.value += AbortedTxn.TOTAL_SIZE;
                    return nextEntry;
                } catch (IOException e) {
                    // We received an unexpected error reading from the index file. We propagate this as an
                    // UNKNOWN error to the consumer, which will cause it to retry the fetch.
                    throw new KafkaException("Failed to read from the transaction index " + file.getAbsolutePath(), e);
                }
            }

        };
    }
Do a basic sanity check on this index to detect obvious problems.

@throws CorruptIndexException if any problems are found.
storage/src/main/java/org/apache/kafka/storage/internals/log/TransactionIndex.java