Repository: apache/kafka
File: storage/src/test/java/org/apache/kafka/storage/internals/log/LogConcurrencyTest.java (Java)

Code before:
                    }
                }
            })
        );
    }

    /**
     * Simple consumption task which reads the log in ascending order and collects
     * consumed batches for validation
     */
    private static class ConsumerTask implements Callable<Void> {
        private final UnifiedLog log;
        private final int lastOffset;
        private final List<FetchedBatch> consumedBatches = new ArrayList<>();

        public ConsumerTask(UnifiedLog log, int lastOffset) {
            this.log = log;
            this.lastOffset = lastOffset;
        }

        public List<FetchedBatch> getConsumedBatches() {
            return consumedBatches;
        }

        @Override
        public Void call() throws Exception {
            long fetchOffset = 0L;
            while (log.highWatermark() < lastOffset) {
                final FetchDataInfo readInfo = log.read(fetchOffset, 1, FetchIsolation.HIGH_WATERMARK, true);
                for (RecordBatch batch : readInfo.records.batches()) {
                    consumedBatches.add(new FetchedBatch(batch.baseOffset(), batch.partitionLeaderEpoch()));
                    fetchOffset = batch.lastOffset() + 1;
                }
            }
            return null;
        }
    }

    /**
     * This class simulates basic leader/follower behavior.
     */
    private record LogAppendTask(UnifiedLog log, long lastOffset) implements Callable<Void> {

        @Override
        public Void call() throws Exception {
            int leaderEpoch = 1;
            boolean isLeader = true;
            while (log.highWatermark() < lastOffset) {
                switch (TestUtils.RANDOM.nextInt(2)) {
                    case 0 -> {
                        final LogOffsetMetadata logEndOffsetMetadata = log.logEndOffsetMetadata();

Review comment (performance):
the method `getConsumedBatches` could be removed if `call` returns `List<FetchedBatch>`

Code after:
                                    ),
                                    Integer.MAX_VALUE
                            );
                            log.updateHighWatermark(logEndOffset);
                        }
                    }
                    case 1 -> {
                        isLeader = !isLeader;
                        leaderEpoch += 1;

                        if (!isLeader) {
                            log.truncateTo(log.highWatermark());
                        }
                    }
                }
            }
        });
    }

    private UnifiedLog createLog() throws IOException {
        return createLog(new LogConfig(Map.of()));
    }

    private UnifiedLog createLog(LogConfig config) throws IOException {
        log = UnifiedLog.create(
            logDir,
            config,
            0L,
            0L,
            scheduler,
            brokerTopicStats,
            Time.SYSTEM,
            5 * 60 * 1000,
            new ProducerStateManagerConfig(TransactionLogConfig.PRODUCER_ID_EXPIRATION_MS_DEFAULT, false),
            TransactionLogConfig.PRODUCER_ID_EXPIRATION_CHECK_INTERVAL_MS_DEFAULT,
            new LogDirFailureChannel(10),
            true,
            Optional.empty()
        );

        return log;
    }

    private void validateConsumedData(UnifiedLog log, Map<Long, Integer> consumedBatches) {
        Iterator<Map.Entry<Long, Integer>> iter = consumedBatches.entrySet().iterator();
        log.logSegments().forEach(segment ->
                segment.log().batches().forEach(batch -> {
                    if (iter.hasNext()) {
                        final Map.Entry<Long, Integer> consumedBatch = iter.next();
                        final long consumedBatchBaseOffset = consumedBatch.getKey();
                        final int consumedBatchEpoch = consumedBatch.getValue();