Repository: apache/kafka
File: jmh-benchmarks/src/main/java/org/apache/kafka/jmh/log/TestLinearWriteSpeed.java (Java)

Code before:
            this.messages = messages;
            Utils.delete(dir);
            this.compression = compression;
            this.records = recordsList.toArray(new SimpleRecord[0]);
            this.log = UnifiedLog.create(
                dir,
                config,
                0L,
                0L,
                scheduler,
                new 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(),
                new CopyOnWriteMap<>(),
                false,
                LogOffsetsListener.NO_OP_OFFSETS_LISTENER
            );
        }

        public int write() {
            this.messages = MemoryRecords.withRecords(compression, records);
            log.appendAsLeader(
                messages,
                0,
                AppendOrigin.CLIENT,
                RequestLocal.noCaching(),
                VerificationGuard.SENTINEL
            );
            return messages.sizeInBytes();
        }

        public void close() throws IOException {
            log.close();
            Utils.delete(log.dir());
        }
    }

    private record Options(OptionSpec<String> dirOpt, OptionSpec<Long> bytesOpt, OptionSpec<Integer> sizeOpt,
                           OptionSpec<Integer> messageSizeOpt, OptionSpec<Integer> filesOpt,
                           OptionSpec<Long> reportingIntervalOpt, OptionSpec<Integer> maxThroughputOpt,
                           OptionSpec<Long> flushIntervalOpt, OptionSpec<String> compressionCodecOpt,
                           OptionSpec<Integer> compressionLevelOpt, OptionSpec<Void> channelOpt,
                           OptionSpec<Void> logOpt, OptionSpec<Void> mmapOpt) {
    }

    private static Options createOptions(OptionParser parser) {

Review comment (performance):
Instead of creating new `messages`, it might be better to reset the base offset. For example:
```java
messages.batches().forEach(b -> b.setLastOffset(0));
```

Code after:
            this.log = UnifiedLog.create(
                dir,
                config,
                0L,
                0L,
                scheduler,
                new 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(),
                new CopyOnWriteMap<>(),
                false,
                LogOffsetsListener.NO_OP_OFFSETS_LISTENER
            );
        }

        public int write() {
            // reset the last offset for each batch to avoid failure caused by offset check
            messages.batches().forEach(b -> b.setLastOffset(b.lastOffset() - b.baseOffset()));
            log.appendAsLeader(
                messages,
                0,
                AppendOrigin.CLIENT,
                RequestLocal.noCaching(),
                VerificationGuard.SENTINEL
            );
            return messages.sizeInBytes();
        }

        public void close() throws IOException {
            log.close();
            Utils.delete(log.dir());
        }
    }

    private record Options(OptionSpec<String> dirOpt, OptionSpec<Long> bytesOpt, OptionSpec<Integer> sizeOpt,
                           OptionSpec<Integer> messageSizeOpt, OptionSpec<Integer> filesOpt,
                           OptionSpec<Long> reportingIntervalOpt, OptionSpec<Integer> maxThroughputOpt,
                           OptionSpec<Long> flushIntervalOpt, OptionSpec<String> compressionCodecOpt,
                           OptionSpec<Integer> compressionLevelOpt, OptionSpec<Void> channelOpt,
                           OptionSpec<Void> logOpt, OptionSpec<Void> mmapOpt) {
    }

    private static Options createOptions(OptionParser parser) {
        OptionSpec<String> dirOpt = parser.accepts("dir", "The directory to write to.")
            .withRequiredArg()
            .describedAs("path")