josefk31 commented on code in PR #22669: URL: https://github.com/apache/kafka/pull/22669#discussion_r3561437547
########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/ElectionBenchmarks.java: ########## @@ -0,0 +1,78 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.jmh.raft; + +import org.apache.kafka.raft.RaftClientBenchmarkContext; +import org.apache.kafka.raft.RaftClientTestContext; + +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Warmup; + +import java.util.Optional; +import java.util.concurrent.TimeUnit; + +/** + * Benchmarks for the leader-election path. The outer class is intentionally not a JMH {@code @State}: + * each benchmark declares the starting state it needs as a nested {@code @State} parameter, so + * different election scenarios (e.g. a future Prospective or Candidate start) can have their own + * setup without forcing a single shared {@code @Setup} on the whole class. + */ +@BenchmarkMode(Mode.SingleShotTime) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +@Warmup(iterations = RaftClientBenchmarkContext.SINGLE_SHOT_WARMUP_ITERATIONS) +@Measurement(iterations = RaftClientBenchmarkContext.SINGLE_SHOT_MEASUREMENT_ITERATIONS) +@Fork(RaftClientBenchmarkContext.SINGLE_SHOT_FORKS) +public class ElectionBenchmarks { + + /** + * Starting state: the local node is Unattached in a {@code voterCount}-node cluster. A fresh + * context is built per invocation because driving the election to completion consumes it. + */ + @State(Scope.Thread) Review Comment: Are we committing any concurrency sins? ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientBenchmarkContext.java: ########## @@ -0,0 +1,277 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.raft; + +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.raft.RaftClientTestContext.RaftProtocol; +import org.apache.kafka.server.common.KRaftVersion; + +import java.util.Arrays; +import java.util.Comparator; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.ThreadLocalRandom; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +public final class RaftClientBenchmarkContext { + // Standardized JMH iteration counts, shared by all raft benchmarks so every benchmark of a given + // mode is configured identically. + + // SingleShotTime measures a single operation per iteration, so it needs many iterations to build + // a stable distribution. + public static final int SINGLE_SHOT_WARMUP_ITERATIONS = 50; + public static final int SINGLE_SHOT_MEASUREMENT_ITERATIONS = 30; + public static final int SINGLE_SHOT_FORKS = 5; + + // AverageTime averages many operations within each timed iteration, so it needs fewer. + public static final int AVERAGE_TIME_WARMUP_ITERATIONS = 5; + public static final int AVERAGE_TIME_MEASUREMENT_ITERATIONS = 10; + public static final int AVERAGE_TIME_FORKS = 3; + + // Default to the newest version of each (the highest-ordinal enum constant). Enum natural order + // is ordinal order, so this picks the last-declared constant; relies on the constants being + // declared oldest-to-newest, which avoids updating these when a new version is added. + public static final KRaftVersion DEFAULT_KRAFT_VERSION = + Arrays.stream(KRaftVersion.values()).max(Comparator.naturalOrder()).orElseThrow(); + public static final RaftProtocol DEFAULT_RAFT_PROTOCOL = + Arrays.stream(RaftProtocol.values()).max(Comparator.naturalOrder()).orElseThrow(); + + private final RaftClientTestContext context; + private final MockLog log; + private final MockNetworkChannel channel; + + private final ReplicaKey localKey; + private final List<ReplicaKey> startingVoters; + private final List<ReplicaKey> startingObservers; + + // Each tracks one cumulative mock counter as a drainable delta against a baseline. The baseline is + // snapshotted at construction and re-baselined by zeroCountersOnSetup() at the end of benchmark setup. + private final DrainableCounter logFlushes; + private final DrainableCounter logReads; + private final DrainableCounter logTruncations; + private final DrainableCounter rpcRequestsSent; + private final DrainableCounter quorumStateWrites; + private final DrainableCounter quorumStateReads; + + private RaftClientBenchmarkContext( + RaftClientTestContext context, + ReplicaKey localKey, + List<ReplicaKey> startingVoters, + List<ReplicaKey> startingObservers + ) { + this.context = context; + this.log = context.log; + this.channel = context.channel; + this.localKey = localKey; + this.startingVoters = List.copyOf(startingVoters); + this.startingObservers = List.copyOf(startingObservers); + this.logFlushes = new DrainableCounter(log::flushCount); + this.logReads = new DrainableCounter(log::readCount); + this.logTruncations = new DrainableCounter(log::truncationCount); + this.rpcRequestsSent = new DrainableCounter(channel::requestsSent); + this.quorumStateWrites = new DrainableCounter(context::quorumStateWriteCount); + this.quorumStateReads = new DrainableCounter(context::quorumStateReadCount); + } + + /** + * Builds the local node as a voter in the Unattached state (a voter with no leader yet) in a + * {@code voterCount}-node cluster. The local node is a voter because only a voter can drive the + * measured operation {@link RaftClientTestContext#unattachedToLeader()} (on {@link #testContext()}) + * — an Unattached → Leader election, a path observers cannot take. A single-voter cluster is + * rejected because such a node elects itself at initialization, before any measured poll. + */ + public static RaftClientBenchmarkContext unattachedVoter(int voterCount) throws Exception { + return unattachedVoter(voterCount, DEFAULT_KRAFT_VERSION, DEFAULT_RAFT_PROTOCOL); + } + + public static RaftClientBenchmarkContext unattachedVoter( + int voterCount, + KRaftVersion kraftVersion, + RaftProtocol raftProtocol + ) throws Exception { + if (voterCount < 2) { + throw new IllegalArgumentException("voterCount must be at least 2; a single voter self-elects at init"); + } + List<ReplicaKey> voterKeys = replicaKeys(randomReplicaId(), voterCount); + ReplicaKey local = voterKeys.get(0); + return new RaftClientBenchmarkContext( + buildContext(local, voterKeys, kraftVersion, raftProtocol), local, voterKeys, List.of()); + } + + public static RaftClientBenchmarkContext leader(int voterCount) throws Exception { + return leader(voterCount, 0, DEFAULT_KRAFT_VERSION, DEFAULT_RAFT_PROTOCOL); + } + + public static RaftClientBenchmarkContext leader(int voterCount, int observerCount) throws Exception { + return leader(voterCount, observerCount, DEFAULT_KRAFT_VERSION, DEFAULT_RAFT_PROTOCOL); + } + + /** + * Builds a leader in a cluster of {@code voterCount} voters and {@code observerCount} observers. + */ + public static RaftClientBenchmarkContext leader( + int voterCount, + int observerCount, + KRaftVersion kraftVersion, + RaftProtocol raftProtocol + ) throws Exception { Review Comment: Can we be more specific what exception this throws? IIUC it's the `IOException` from `RaftClientBenchmarkContext` ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientBenchmarkContext.java: ########## @@ -0,0 +1,277 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.raft; + +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.raft.RaftClientTestContext.RaftProtocol; +import org.apache.kafka.server.common.KRaftVersion; + +import java.util.Arrays; +import java.util.Comparator; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.ThreadLocalRandom; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +public final class RaftClientBenchmarkContext { + // Standardized JMH iteration counts, shared by all raft benchmarks so every benchmark of a given + // mode is configured identically. + + // SingleShotTime measures a single operation per iteration, so it needs many iterations to build + // a stable distribution. + public static final int SINGLE_SHOT_WARMUP_ITERATIONS = 50; + public static final int SINGLE_SHOT_MEASUREMENT_ITERATIONS = 30; + public static final int SINGLE_SHOT_FORKS = 5; + + // AverageTime averages many operations within each timed iteration, so it needs fewer. + public static final int AVERAGE_TIME_WARMUP_ITERATIONS = 5; + public static final int AVERAGE_TIME_MEASUREMENT_ITERATIONS = 10; + public static final int AVERAGE_TIME_FORKS = 3; + + // Default to the newest version of each (the highest-ordinal enum constant). Enum natural order + // is ordinal order, so this picks the last-declared constant; relies on the constants being + // declared oldest-to-newest, which avoids updating these when a new version is added. + public static final KRaftVersion DEFAULT_KRAFT_VERSION = + Arrays.stream(KRaftVersion.values()).max(Comparator.naturalOrder()).orElseThrow(); Review Comment: Is there a reason to compute this constant? What about not having it and just using `KRaftVersion#LATEST_PRODUCTION`? I think it's safe to assume that value will always be kept up to date. ########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/LeaderBenchmarks.java: ########## @@ -0,0 +1,96 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.jmh.raft; + +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.raft.RaftClientBenchmarkContext; +import org.apache.kafka.raft.RaftClientTestContext; + +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Warmup; + +import java.util.Optional; +import java.util.concurrent.TimeUnit; + +/** + * Benchmarks for the leader request-handling path. The outer class is intentionally not a JMH + * {@code @State}: each benchmark declares the starting state it needs as a nested {@code @State} + * parameter, so future leader scenarios (e.g. a lagging-follower fetch or a commit) can have their own + * setup without forcing a single shared {@code @Setup} on the whole class. + */ +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +@Warmup(iterations = RaftClientBenchmarkContext.AVERAGE_TIME_WARMUP_ITERATIONS) +@Measurement(iterations = RaftClientBenchmarkContext.AVERAGE_TIME_MEASUREMENT_ITERATIONS) +@Fork(RaftClientBenchmarkContext.AVERAGE_TIME_FORKS) +public class LeaderBenchmarks { + + /** + * Starting state: the local node is leader with the high watermark at the log end. + */ + @State(Scope.Thread) + public static class LeaderWithHwmAtLogEnd { + static final int VOTER_COUNT = 3; + + RaftClientBenchmarkContext benchmark; + RaftClientTestContext context; + + int epoch; + long endOffset; + + @Setup(Level.Trial) + public void setup() throws Exception { + benchmark = RaftClientBenchmarkContext.leader(VOTER_COUNT); + context = benchmark.testContext(); + context.advanceLocalLeaderHighWatermarkToLogEndOffset(); + epoch = context.currentEpoch(); + endOffset = benchmark.logEndOffset(); + benchmark.zeroCountersOnSetup(); + } + } + + /** + * Leader handles a FETCH from a fully caught-up follower (fetch offset == log end offset) that asks + * not to wait ({@code maxWaitMs = 0}), so the leader replies immediately rather than deferring. + * + * <p>Note: a real caught-up follower long-polls with {@code maxWaitMs > 0}, and such a fetch is + * <em>deferred</em> (held until new data arrives or the wait times out) — it would not produce an + * immediate response. This benchmark deliberately uses {@code maxWaitMs = 0} to measure the + * immediate-reply path. + */ + @Benchmark + public void handleNoWaitFetchFromCaughtUpFollower( + LeaderWithHwmAtLogEnd state, + KRaftBenchmarkingCounters counters + ) throws Exception { Review Comment: Is there a reason we cannot use the more specific `throws InteruptException`? ########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/ElectionBenchmarks.java: ########## @@ -0,0 +1,78 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.jmh.raft; + +import org.apache.kafka.raft.RaftClientBenchmarkContext; +import org.apache.kafka.raft.RaftClientTestContext; + +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Warmup; + +import java.util.Optional; +import java.util.concurrent.TimeUnit; + +/** + * Benchmarks for the leader-election path. The outer class is intentionally not a JMH {@code @State}: + * each benchmark declares the starting state it needs as a nested {@code @State} parameter, so + * different election scenarios (e.g. a future Prospective or Candidate start) can have their own + * setup without forcing a single shared {@code @Setup} on the whole class. + */ +@BenchmarkMode(Mode.SingleShotTime) Review Comment: I'm not super familiar with JMH benchmarks so this is both a question for my learning as well as a comment. What is the reasoning behind this being a `Mode.SingleShot`? I think we might also care about the # of elections we can perform over a given period of time - in other words throughput. Should we also record `Mode.Throughput`? ########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/KRaftBenchmarkingCounters.java: ########## @@ -0,0 +1,157 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.jmh.raft; + +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.raft.RaftClientBenchmarkContext; + +import org.openjdk.jmh.annotations.AuxCounters; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.infra.BenchmarkParams; + +import java.util.Optional; + +/** + * Secondary, machine-independent work counters reported by the raft benchmarks alongside the timing + * score, as {@code benchmark:counter} rows. + * + * <p>Throughout this class, an <em>operation</em> is JMH's unit of work: a single invocation of a + * {@code @Benchmark}-annotated method. (One operation equals one invocation here because we don't use + * {@code @OperationsPerInvocation}.) JMH reports the timing score in {@code ns/op}, and these work + * counters are reported {@code PerOp} to match. + * + * <p>The per-operation values are integer-exact and should be stable across a correct refactor of + * {@code KafkaRaftClient}: a flush count moving from 1 to 2 per operation is a behavioral diff, not + * measurement noise. The counters that are zero on a path (e.g. log flushes on a caught-up fetch) + * are the most useful tripwires, since zero is speed-independent. + */ +@State(Scope.Thread) +@AuxCounters(AuxCounters.Type.EVENTS) +public class KRaftBenchmarkingCounters { + // Private accumulators: not reported directly (we report the per-op values below). Being private, + // JMH does not touch them between iterations, so reset() must zero them. + private long logFlushesTotal; + private long logReadsTotal; + private long logTruncationsTotal; + private long rpcRequestsSentTotal; + private long rpcResponsesSentTotal; + private long quorumStateWritesTotal; + private long quorumStateReadsTotal; + + // Reported: the number of operations (i.e. @Benchmark method invocations) measured in the + // iteration, and the divisor for the per-operation values below. Being a public @AuxCounters + // field, JMH zeroes it automatically at the start of every iteration (which is why, unlike the + // private totals above, it is not reset in reset()). + public long operations; + + // The divisor for the per-op methods below: (forks x measurement iterations). Set once per fork + // by captureRunShape(). + private double measurementDataPoints = 1.0; + + /** + * Captures the number of measurement data points — {@code forks x measurement iterations} — that + * JMH will SUM the {@code *PerOp()} methods over ({@code Type.EVENTS} secondary results are + * SUM-aggregated across iterations and forks). Each per-op method pre-divides by this count so that + * the SUM reports the exact per-operation value (e.g. {@code logReadsPerOp = 1.0}) in the summary + * row. Reading it from {@link BenchmarkParams} tracks the actual run shape (including + * {@code -f}/{@code -i} overrides) rather than hardcoding the annotation values. + */ + @Setup(Level.Trial) + public void captureRunShape(BenchmarkParams params) { + if (params.getThreads() != 1) { + throw new IllegalStateException( + "raft benchmarks are single-threaded (one client over shared mocks); got " + + params.getThreads() + " threads"); + } + // forks() is 0 when forking is disabled (in-process), which is still one set of iterations. + int forks = Math.max(1, params.getForks()); + measurementDataPoints = (double) forks * params.getMeasurement().getCount(); + } + + @Setup(Level.Iteration) + public void reset() { + logFlushesTotal = 0; + logReadsTotal = 0; Review Comment: If we're counting reads should we consider also counting appends as well? I suspect that for some scenarios writes may play an important part. Everything which is written is also fetched twice assuming we have a 3-node quorum. So roughly speaking we'd expect simple appends to disk to be an important part of disk IO. ########## raft/src/test/java/org/apache/kafka/raft/DrainableCounterTest.java: ########## @@ -0,0 +1,98 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.raft; + +import org.junit.jupiter.api.Test; + +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +public final class DrainableCounterTest { + + @Test + public void testDrainDeltaReturnsIncreaseSinceLastDrain() { + AtomicInteger source = new AtomicInteger(); + DrainableCounter counter = new DrainableCounter(source::get); + + source.addAndGet(3); + assertEquals(3, counter.drainDelta()); + + source.addAndGet(2); + assertEquals(2, counter.drainDelta()); + } + + @Test + public void testDrainDeltaIsZeroWhenSourceUnchanged() { + AtomicInteger source = new AtomicInteger(7); + DrainableCounter counter = new DrainableCounter(source::get); + + assertEquals(0, counter.drainDelta()); + assertEquals(0, counter.drainDelta()); + } + + @Test + public void testDrainDeltaAdvancesTheBaseline() { + AtomicInteger source = new AtomicInteger(); + DrainableCounter counter = new DrainableCounter(source::get); + + source.addAndGet(6); + assertEquals(6, counter.drainDelta()); + // The previous drainDelta() advanced the baseline, so the same increase is not counted twice. + assertEquals(0, counter.drainDelta()); + } + + @Test + public void testBaselineIsSnapshottedAtConstruction() { + AtomicInteger source = new AtomicInteger(9); + DrainableCounter counter = new DrainableCounter(source::get); + + // The constructor snapshots the current value, so the first drain only reflects increases + // since construction, not the source's prior history. + assertEquals(0, counter.drainDelta()); + + source.addAndGet(3); + assertEquals(3, counter.drainDelta()); + } + + @Test + public void testIgnoredDrainDeltaExcludesPriorIncreasesFromTheNextDrain() { + AtomicInteger source = new AtomicInteger(); + DrainableCounter counter = new DrainableCounter(source::get); + + source.addAndGet(3); + assertEquals(3, counter.drainDelta()); + + // Draining with the result ignored re-baselines the counter, e.g. to exclude work done while + // setting up a benchmark from the measured region. + source.addAndGet(5); + counter.drainDelta(); + assertEquals(0, counter.drainDelta()); + + source.addAndGet(4); + assertEquals(4, counter.drainDelta()); + } + + @Test + public void testDrainDeltaIsCorrectWhenSourceOverflows() { Review Comment: I don't think the overflow case is correct. ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientBenchmarkContext.java: ########## @@ -0,0 +1,277 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.raft; + +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.raft.RaftClientTestContext.RaftProtocol; +import org.apache.kafka.server.common.KRaftVersion; + +import java.util.Arrays; +import java.util.Comparator; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.ThreadLocalRandom; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +public final class RaftClientBenchmarkContext { + // Standardized JMH iteration counts, shared by all raft benchmarks so every benchmark of a given + // mode is configured identically. + + // SingleShotTime measures a single operation per iteration, so it needs many iterations to build + // a stable distribution. + public static final int SINGLE_SHOT_WARMUP_ITERATIONS = 50; + public static final int SINGLE_SHOT_MEASUREMENT_ITERATIONS = 30; + public static final int SINGLE_SHOT_FORKS = 5; + + // AverageTime averages many operations within each timed iteration, so it needs fewer. + public static final int AVERAGE_TIME_WARMUP_ITERATIONS = 5; + public static final int AVERAGE_TIME_MEASUREMENT_ITERATIONS = 10; + public static final int AVERAGE_TIME_FORKS = 3; + + // Default to the newest version of each (the highest-ordinal enum constant). Enum natural order + // is ordinal order, so this picks the last-declared constant; relies on the constants being + // declared oldest-to-newest, which avoids updating these when a new version is added. + public static final KRaftVersion DEFAULT_KRAFT_VERSION = + Arrays.stream(KRaftVersion.values()).max(Comparator.naturalOrder()).orElseThrow(); + public static final RaftProtocol DEFAULT_RAFT_PROTOCOL = + Arrays.stream(RaftProtocol.values()).max(Comparator.naturalOrder()).orElseThrow(); + + private final RaftClientTestContext context; + private final MockLog log; + private final MockNetworkChannel channel; + + private final ReplicaKey localKey; + private final List<ReplicaKey> startingVoters; + private final List<ReplicaKey> startingObservers; + + // Each tracks one cumulative mock counter as a drainable delta against a baseline. The baseline is + // snapshotted at construction and re-baselined by zeroCountersOnSetup() at the end of benchmark setup. + private final DrainableCounter logFlushes; + private final DrainableCounter logReads; + private final DrainableCounter logTruncations; + private final DrainableCounter rpcRequestsSent; + private final DrainableCounter quorumStateWrites; + private final DrainableCounter quorumStateReads; + + private RaftClientBenchmarkContext( + RaftClientTestContext context, + ReplicaKey localKey, + List<ReplicaKey> startingVoters, + List<ReplicaKey> startingObservers + ) { + this.context = context; + this.log = context.log; + this.channel = context.channel; + this.localKey = localKey; + this.startingVoters = List.copyOf(startingVoters); + this.startingObservers = List.copyOf(startingObservers); + this.logFlushes = new DrainableCounter(log::flushCount); + this.logReads = new DrainableCounter(log::readCount); + this.logTruncations = new DrainableCounter(log::truncationCount); + this.rpcRequestsSent = new DrainableCounter(channel::requestsSent); + this.quorumStateWrites = new DrainableCounter(context::quorumStateWriteCount); + this.quorumStateReads = new DrainableCounter(context::quorumStateReadCount); + } + + /** + * Builds the local node as a voter in the Unattached state (a voter with no leader yet) in a + * {@code voterCount}-node cluster. The local node is a voter because only a voter can drive the + * measured operation {@link RaftClientTestContext#unattachedToLeader()} (on {@link #testContext()}) + * — an Unattached → Leader election, a path observers cannot take. A single-voter cluster is + * rejected because such a node elects itself at initialization, before any measured poll. + */ + public static RaftClientBenchmarkContext unattachedVoter(int voterCount) throws Exception { + return unattachedVoter(voterCount, DEFAULT_KRAFT_VERSION, DEFAULT_RAFT_PROTOCOL); + } + + public static RaftClientBenchmarkContext unattachedVoter( + int voterCount, + KRaftVersion kraftVersion, + RaftProtocol raftProtocol + ) throws Exception { + if (voterCount < 2) { + throw new IllegalArgumentException("voterCount must be at least 2; a single voter self-elects at init"); + } + List<ReplicaKey> voterKeys = replicaKeys(randomReplicaId(), voterCount); + ReplicaKey local = voterKeys.get(0); + return new RaftClientBenchmarkContext( + buildContext(local, voterKeys, kraftVersion, raftProtocol), local, voterKeys, List.of()); + } + + public static RaftClientBenchmarkContext leader(int voterCount) throws Exception { Review Comment: I might be wrong but it seems that we only use `leader(voterCount)`. Ideally we should remove code which is not meaningfully used. This reduces the lines of code and thus makes things easier to read. Id suggest either inlining `leader(int, int, KRaftVersion, RaftProtocol)` or exposing just `leader(int, int, KRaftVersion, RaftProtocol)` itself to callers. IE delete `leader(int)` and `leader(int, int)`. I think this comment also applies to `unattachedVoter`. I think applying these two changes would result in -100 lines of code. What do you think? ########## raft/src/test/java/org/apache/kafka/raft/DrainableCounterTest.java: ########## @@ -0,0 +1,98 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.raft; + +import org.junit.jupiter.api.Test; + +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +public final class DrainableCounterTest { + + @Test + public void testDrainDeltaReturnsIncreaseSinceLastDrain() { + AtomicInteger source = new AtomicInteger(); + DrainableCounter counter = new DrainableCounter(source::get); + + source.addAndGet(3); + assertEquals(3, counter.drainDelta()); + + source.addAndGet(2); + assertEquals(2, counter.drainDelta()); + } + + @Test + public void testDrainDeltaIsZeroWhenSourceUnchanged() { + AtomicInteger source = new AtomicInteger(7); + DrainableCounter counter = new DrainableCounter(source::get); + + assertEquals(0, counter.drainDelta()); + assertEquals(0, counter.drainDelta()); + } + + @Test + public void testDrainDeltaAdvancesTheBaseline() { + AtomicInteger source = new AtomicInteger(); + DrainableCounter counter = new DrainableCounter(source::get); + + source.addAndGet(6); + assertEquals(6, counter.drainDelta()); + // The previous drainDelta() advanced the baseline, so the same increase is not counted twice. + assertEquals(0, counter.drainDelta()); + } + + @Test + public void testBaselineIsSnapshottedAtConstruction() { + AtomicInteger source = new AtomicInteger(9); + DrainableCounter counter = new DrainableCounter(source::get); + + // The constructor snapshots the current value, so the first drain only reflects increases + // since construction, not the source's prior history. + assertEquals(0, counter.drainDelta()); + + source.addAndGet(3); + assertEquals(3, counter.drainDelta()); + } + + @Test + public void testIgnoredDrainDeltaExcludesPriorIncreasesFromTheNextDrain() { + AtomicInteger source = new AtomicInteger(); + DrainableCounter counter = new DrainableCounter(source::get); + + source.addAndGet(3); + assertEquals(3, counter.drainDelta()); + + // Draining with the result ignored re-baselines the counter, e.g. to exclude work done while + // setting up a benchmark from the measured region. + source.addAndGet(5); + counter.drainDelta(); + assertEquals(0, counter.drainDelta()); + + source.addAndGet(4); + assertEquals(4, counter.drainDelta()); + } + + @Test + public void testDrainDeltaIsCorrectWhenSourceOverflows() { Review Comment: I don't think the overflow case is correct. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
