kevin-wu24 commented on code in PR #22669: URL: https://github.com/apache/kafka/pull/22669#discussion_r3474818513
########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/LeaderBenchmark.java: ########## @@ -0,0 +1,67 @@ +/* Review Comment: We should rename this to `LeaderBenchmarks.java` since we will write multiple benchmarks. This applies to `ElectionBenchmark.java` too. ########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/LeaderBenchmark.java: ########## @@ -0,0 +1,67 @@ +/* + * 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.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.concurrent.TimeUnit; + +@State(Scope.Thread) +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +@Warmup(iterations = 5) +@Measurement(iterations = 10) +@Fork(3) +public class LeaderBenchmark { + private static final int LOCAL_ID = 0; + private static final int VOTER_COUNT = 3; + + private RaftClientBenchmarkContext context; + + @Setup(Level.Trial) + public void setup() throws Exception { + context = RaftClientBenchmarkContext.leader(LOCAL_ID, VOTER_COUNT); + context.advanceLeaderHwmToLogEnd(); + context.resetCounters(); + } + + /** + * Leader handles a valid FETCH from a fully caught-up follower (fetch offset == log end offset), + * which does not advance the high watermark — the steady-state heartbeat-style fetch. + */ + @Benchmark + public void leaderHandleFetch(ProtocolCounters counters) throws Exception { Review Comment: We may want to rename this to be more specific like `handleFetchFromCaughtUpFollower`. We don't need "leader" in the method name because this file is named `LeaderBenchmarks`. ########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/ElectionBenchmark.java: ########## @@ -0,0 +1,59 @@ +/* + * 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.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.concurrent.TimeUnit; + +@State(Scope.Thread) +@BenchmarkMode(Mode.SingleShotTime) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +@Warmup(iterations = 50) +@Measurement(iterations = 30) Review Comment: It would be nice to standardize these two numbers for the two benchmarking cases we have, `Mode.SingleShotTime`, and `Mode.AverageTime`. Something to keep in mind when you are writing future benchmarks. Perhaps these should be static constants in the `RaftClientBanchmarkContext`. ########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/ProtocolCounters.java: ########## @@ -0,0 +1,108 @@ +/* + * 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.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; + +/** + * Secondary, machine-independent work counters reported by the raft benchmarks alongside the timing + * score, as {@code benchmark:counter} rows. + * + * <p>Each benchmark calls {@link #drainFrom} every invocation to accumulate the work deltas drained + * from {@link RaftClientBenchmarkContext}. JMH reads the public numeric methods at the end of each + * measurement iteration. The public fields are raw totals intended for JSON post-processing; divide + * them by {@link #operations} to get the exact per-operation value. The public methods return + * per-operation values for each iteration, so the per-iteration console output shows stable values + * such as {@code logReadsPerOp = 1.0}. JMH aggregates {@code Type.EVENTS} secondary results with + * {@code SUM}, so the final summary row will add per-iteration method values together when more + * than one measurement iteration is used. + * + * <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 ProtocolCounters { Review Comment: Could we rename this class to something more descriptive? Perhaps `KRaftBenchmarkingCounters`? ########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/ProtocolCounters.java: ########## @@ -0,0 +1,108 @@ +/* + * 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.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; + +/** + * Secondary, machine-independent work counters reported by the raft benchmarks alongside the timing + * score, as {@code benchmark:counter} rows. + * + * <p>Each benchmark calls {@link #drainFrom} every invocation to accumulate the work deltas drained + * from {@link RaftClientBenchmarkContext}. JMH reads the public numeric methods at the end of each + * measurement iteration. The public fields are raw totals intended for JSON post-processing; divide + * them by {@link #operations} to get the exact per-operation value. The public methods return + * per-operation values for each iteration, so the per-iteration console output shows stable values + * such as {@code logReadsPerOp = 1.0}. JMH aggregates {@code Type.EVENTS} secondary results with + * {@code SUM}, so the final summary row will add per-iteration method values together when more + * than one measurement iteration is used. + * + * <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 ProtocolCounters { + public long logFlushesTotal; + public long logReadsTotal; + public long logTruncationsTotal; + public long rpcRequestsSentTotal; + public long rpcResponsesSentTotal; + public long quorumStateWritesTotal; Review Comment: Per the @AuxCounters javadoc: ``` "Only public fields and methods are considered as metrics... only numeric fields and numeric-returning methods are considered — all primitives and their boxed counterparts, except boolean/Boolean and char/Character." ``` We should make these fields `private`, since we don't really care about these numbers in the final benchmarking output. We care about the # of operations per invocation. ########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/ElectionBenchmark.java: ########## @@ -0,0 +1,59 @@ +/* + * 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.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.concurrent.TimeUnit; + +@State(Scope.Thread) +@BenchmarkMode(Mode.SingleShotTime) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +@Warmup(iterations = 50) +@Measurement(iterations = 30) +@Fork(5) +public class ElectionBenchmark { + private static final int LOCAL_ID = 0; + private static final int VOTER_COUNT = 3; Review Comment: For benchmarks which rely on the "number of voters" to do something, we may want to parametrize to run with multiple voter set sizes (e.g. 3, 5, etc.). We should expect that the benchmark time for these operations should scale linearly with the voter set size/2. We could add it here, and something to consider for future benchmarks regarding elections/advancing the HWM. ########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/ElectionBenchmark.java: ########## @@ -0,0 +1,59 @@ +/* + * 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.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.concurrent.TimeUnit; + +@State(Scope.Thread) Review Comment: I want to say we don't need the benchmark class itself to be a `@State` annotated object. I think it's this way because the `context` is an instance variable. ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientBenchmarkContext.java: ########## @@ -0,0 +1,182 @@ +/* + * 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.message.FetchRequestData; +import org.apache.kafka.common.protocol.ApiMessage; +import org.apache.kafka.server.common.KRaftVersion; + +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +public final class RaftClientBenchmarkContext { + private final RaftClientTestContext context; + private final MockLog log; + private final MockNetworkChannel channel; + private final ReplicaKey otherVoter; + + private int lastFlushCount; + private int lastReadCount; + private int lastTruncationCount; + private int lastRequestsSent; + private int lastQuorumWrites; + + private RaftClientBenchmarkContext(RaftClientTestContext context, ReplicaKey otherVoter) { + this.context = context; + this.log = context.log; + this.channel = context.channel; + this.otherVoter = otherVoter; + resetCounters(); Review Comment: We should not need to call `resetCounters()` in the constructor of this method. Everything would just be 0 at this point if we are calling the `RaftClientBenchmarkContext` constructor at the right time (i.e. the constructor should be invoked at the `@Setup` of a new `Iteration` for `AverageTime`/stead-state benchmarks, and at the `@Setup` of a new `Invocation` for `SingleShotTime`/state-changing benchmarks. ########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/ElectionBenchmark.java: ########## @@ -0,0 +1,59 @@ +/* + * 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.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.concurrent.TimeUnit; + +@State(Scope.Thread) +@BenchmarkMode(Mode.SingleShotTime) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +@Warmup(iterations = 50) +@Measurement(iterations = 30) +@Fork(5) +public class ElectionBenchmark { + private static final int LOCAL_ID = 0; + private static final int VOTER_COUNT = 3; + + private RaftClientBenchmarkContext context; + + @Setup(Level.Invocation) + public void setup() throws Exception { + context = RaftClientBenchmarkContext.unattached(LOCAL_ID, VOTER_COUNT); + } Review Comment: This is probably not the right way to set up benchmarks after we start writing more of them, since this means all `@Benchmark` methods in this class must use the same exact `setup()` method. That likely will not be true, especially for the starting states in the election part of the protocol. Did you consider using an inner `@State` class with its own `@Setup` method that you can pass as a parameter to the `@Benchmark` method? Similar to what you did with `ProtocolCounters`, except this class would hold the context. `@Benchmark` methods that need the same initial state could still reuse these inner classes, but now your overall class does not have only 1 setup method shared by all `@Benchmark` methods. For example: ``` @BenchmarkMode(Mode.Throughput) public class A { // plain class — NOT @State @State(Scope.Thread) public static class B { // public STATIC nested class Map<String, Object> map; @Setup(Level.Trial) public void setUp() { map = buildMapForBenchmark1(); } } @State(Scope.Thread) public static class C { // a different setup int[] data; @Setup(Level.Trial) public void setUp() { data = buildArrayForBenchmark2(); } } @Benchmark public Object benchmark1(B b) { // only B.setUp() runs for this method return b.map.get("k"); } @Benchmark public int benchmark2(C c) { // only C.setUp() runs for this method public void setUp() { data = buildArrayForBenchmark2(); } } @Benchmark public Object benchmark1(B b) { // only B.setUp() runs for this method return b.map.get("k"); } @Benchmark public int benchmark2(C c) { // only C.setUp() runs for this method return c.data[0]; } } ``` ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientBenchmarkContext.java: ########## @@ -0,0 +1,182 @@ +/* + * 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.message.FetchRequestData; +import org.apache.kafka.common.protocol.ApiMessage; +import org.apache.kafka.server.common.KRaftVersion; + +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +public final class RaftClientBenchmarkContext { + private final RaftClientTestContext context; + private final MockLog log; + private final MockNetworkChannel channel; + private final ReplicaKey otherVoter; + + private int lastFlushCount; + private int lastReadCount; + private int lastTruncationCount; + private int lastRequestsSent; + private int lastQuorumWrites; + + private RaftClientBenchmarkContext(RaftClientTestContext context, ReplicaKey otherVoter) { + this.context = context; + this.log = context.log; + this.channel = context.channel; + this.otherVoter = otherVoter; + resetCounters(); + } + + /** + * Builds an unattached node in a {@code voterCount}-node quorum (the local node is not yet the + * leader). Use {@link #unattachedToLeader()} as the measured operation to drive a full + * Unattached → Leader election. A single-voter quorum is rejected because such a node elects + * itself at initialization, before any measured poll. + */ + public static RaftClientBenchmarkContext unattached(int localId, int voterCount) throws Exception { + if (voterCount < 2) { + throw new IllegalArgumentException("voterCount must be at least 2; a single voter self-elects at init"); + } + return new RaftClientBenchmarkContext(buildContext(voterKeys(localId, voterCount)), null); + } + + public static RaftClientBenchmarkContext leader(int localId, int voterCount) throws Exception { + List<ReplicaKey> voterKeys = voterKeys(localId, voterCount); + RaftClientTestContext context = buildContext(voterKeys); + context.unattachedToLeader(); + + return new RaftClientBenchmarkContext(context, voterKeys.get(1)); + } + + /** {@code voterCount} voter keys, each with a random directory id, with {@code localId} first. */ + private static List<ReplicaKey> voterKeys(int localId, int voterCount) { + return IntStream.range(0, voterCount) + .mapToObj(i -> ReplicaKey.of(localId + i, Uuid.randomUuid())) + .collect(Collectors.toList()); + } + + /** + * Builds an unattached node in a KIP-1186 quorum of {@code voterKeys}, whose first entry is the + * local node. + */ + private static RaftClientTestContext buildContext(List<ReplicaKey> voterKeys) throws Exception { + ReplicaKey local = voterKeys.get(0); + VoterSet voters = VoterSetTestUtil.voterSet(voterKeys.stream()); + + return new RaftClientTestContext.Builder(local.id(), local.directoryId().get()) + .withStartingVoters(voters, KRaftVersion.KRAFT_VERSION_1) + .withRaftProtocol(RaftClientTestContext.RaftProtocol.KIP_1186_PROTOCOL) + .withPollIntervalMs(0) + .withUnknownLeader(0) + .build(); + } + + public void unattachedToLeader() throws Exception { + context.unattachedToLeader(); + } + + public void advanceLeaderHwmToLogEnd() throws InterruptedException { + context.advanceLocalLeaderHighWatermarkToLogEndOffset(); + } + + public void pollUntilResponse() throws InterruptedException { + context.pollUntilResponse(); + } + + public int currentEpoch() { + return context.currentEpoch(); + } + + public long logEndOffset() { + return log.endOffset().offset(); + } + + /** A voter other than the local node, e.g. to deliver a FETCH from on the leader. */ + public ReplicaKey otherVoter() { + if (otherVoter == null) { + throw new IllegalStateException("This context was not built with an other-voter key"); + } + return otherVoter; + } + + public void deliverRequest(ApiMessage request) { + context.deliverRequest(request); + } + + public FetchRequestData fetchRequest( + int epoch, + ReplicaKey replicaKey, + long fetchOffset, + int lastFetchedEpoch, + int maxWaitMs + ) { + return context.fetchRequest(epoch, replicaKey, fetchOffset, lastFetchedEpoch, maxWaitMs); + } + + /** Re-baselines all counters to the current totals. Call at the end of benchmark setup. */ + public void resetCounters() { + lastFlushCount = log.flushCount(); + lastReadCount = log.readCount(); + lastTruncationCount = log.truncationCount(); + lastRequestsSent = channel.requestsSent(); + lastQuorumWrites = context.quorumStateWriteCount(); + context.drainAllSentResponses(); Review Comment: If we are going to drain sent responses, we should not do it here. `ProtocolCounters#drainFrom` already does this. ########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/ProtocolCounters.java: ########## @@ -0,0 +1,108 @@ +/* + * 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.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; + +/** + * Secondary, machine-independent work counters reported by the raft benchmarks alongside the timing + * score, as {@code benchmark:counter} rows. + * + * <p>Each benchmark calls {@link #drainFrom} every invocation to accumulate the work deltas drained + * from {@link RaftClientBenchmarkContext}. JMH reads the public numeric methods at the end of each + * measurement iteration. The public fields are raw totals intended for JSON post-processing; divide + * them by {@link #operations} to get the exact per-operation value. The public methods return + * per-operation values for each iteration, so the per-iteration console output shows stable values + * such as {@code logReadsPerOp = 1.0}. JMH aggregates {@code Type.EVENTS} secondary results with + * {@code SUM}, so the final summary row will add per-iteration method values together when more + * than one measurement iteration is used. + * + * <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 ProtocolCounters { + public long logFlushesTotal; + public long logReadsTotal; + public long logTruncationsTotal; + public long rpcRequestsSentTotal; + public long rpcResponsesSentTotal; + public long quorumStateWritesTotal; + public long operations; + + @Setup(Level.Iteration) + public void reset() { + logFlushesTotal = 0; + logReadsTotal = 0; + logTruncationsTotal = 0; + rpcRequestsSentTotal = 0; + rpcResponsesSentTotal = 0; + quorumStateWritesTotal = 0; + operations = 0; + } Review Comment: FYI, since these fields are public right now, we don't need this method. Public numeric fields are zeroed on each iteration by jmh. However, in making the fields private, we need to keep this method. ``` Counters are per-iteration. JMH samples the values at the end of each measurement iteration and resets them between iterations, so each iteration's reported number reflects only that iteration — it doesn't accumulate across the whole run. (This is also why you don't reset them yourself.) ``` ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientBenchmarkContext.java: ########## @@ -0,0 +1,182 @@ +/* + * 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.message.FetchRequestData; +import org.apache.kafka.common.protocol.ApiMessage; +import org.apache.kafka.server.common.KRaftVersion; + +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +public final class RaftClientBenchmarkContext { + private final RaftClientTestContext context; + private final MockLog log; + private final MockNetworkChannel channel; + private final ReplicaKey otherVoter; + + private int lastFlushCount; + private int lastReadCount; + private int lastTruncationCount; + private int lastRequestsSent; + private int lastQuorumWrites; + + private RaftClientBenchmarkContext(RaftClientTestContext context, ReplicaKey otherVoter) { + this.context = context; + this.log = context.log; + this.channel = context.channel; + this.otherVoter = otherVoter; + resetCounters(); + } + + /** + * Builds an unattached node in a {@code voterCount}-node quorum (the local node is not yet the + * leader). Use {@link #unattachedToLeader()} as the measured operation to drive a full + * Unattached → Leader election. A single-voter quorum is rejected because such a node elects + * itself at initialization, before any measured poll. + */ + public static RaftClientBenchmarkContext unattached(int localId, int voterCount) throws Exception { + if (voterCount < 2) { + throw new IllegalArgumentException("voterCount must be at least 2; a single voter self-elects at init"); + } + return new RaftClientBenchmarkContext(buildContext(voterKeys(localId, voterCount)), null); + } + + public static RaftClientBenchmarkContext leader(int localId, int voterCount) throws Exception { + List<ReplicaKey> voterKeys = voterKeys(localId, voterCount); + RaftClientTestContext context = buildContext(voterKeys); + context.unattachedToLeader(); + + return new RaftClientBenchmarkContext(context, voterKeys.get(1)); + } + + /** {@code voterCount} voter keys, each with a random directory id, with {@code localId} first. */ + private static List<ReplicaKey> voterKeys(int localId, int voterCount) { + return IntStream.range(0, voterCount) + .mapToObj(i -> ReplicaKey.of(localId + i, Uuid.randomUuid())) + .collect(Collectors.toList()); + } + + /** + * Builds an unattached node in a KIP-1186 quorum of {@code voterKeys}, whose first entry is the + * local node. + */ + private static RaftClientTestContext buildContext(List<ReplicaKey> voterKeys) throws Exception { + ReplicaKey local = voterKeys.get(0); + VoterSet voters = VoterSetTestUtil.voterSet(voterKeys.stream()); + + return new RaftClientTestContext.Builder(local.id(), local.directoryId().get()) + .withStartingVoters(voters, KRaftVersion.KRAFT_VERSION_1) + .withRaftProtocol(RaftClientTestContext.RaftProtocol.KIP_1186_PROTOCOL) + .withPollIntervalMs(0) + .withUnknownLeader(0) + .build(); + } + + public void unattachedToLeader() throws Exception { + context.unattachedToLeader(); + } + + public void advanceLeaderHwmToLogEnd() throws InterruptedException { + context.advanceLocalLeaderHighWatermarkToLogEndOffset(); + } + + public void pollUntilResponse() throws InterruptedException { + context.pollUntilResponse(); + } + + public int currentEpoch() { + return context.currentEpoch(); + } + + public long logEndOffset() { + return log.endOffset().offset(); + } + + /** A voter other than the local node, e.g. to deliver a FETCH from on the leader. */ + public ReplicaKey otherVoter() { + if (otherVoter == null) { + throw new IllegalStateException("This context was not built with an other-voter key"); + } + return otherVoter; + } + + public void deliverRequest(ApiMessage request) { + context.deliverRequest(request); + } + + public FetchRequestData fetchRequest( + int epoch, + ReplicaKey replicaKey, + long fetchOffset, + int lastFetchedEpoch, + int maxWaitMs + ) { + return context.fetchRequest(epoch, replicaKey, fetchOffset, lastFetchedEpoch, maxWaitMs); + } + + /** Re-baselines all counters to the current totals. Call at the end of benchmark setup. */ + public void resetCounters() { Review Comment: I am still confused about what this method does. Each `drain...` method already sets its corresponding `last...` field so that the next `drain...` call correctly tracks the delta per invocation in a benchmark where the test context is constructed in a `@Setup(Level.Iteration)` method. ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientBenchmarkContext.java: ########## @@ -0,0 +1,182 @@ +/* + * 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.message.FetchRequestData; +import org.apache.kafka.common.protocol.ApiMessage; +import org.apache.kafka.server.common.KRaftVersion; + +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +public final class RaftClientBenchmarkContext { + private final RaftClientTestContext context; + private final MockLog log; + private final MockNetworkChannel channel; + private final ReplicaKey otherVoter; + + private int lastFlushCount; + private int lastReadCount; + private int lastTruncationCount; + private int lastRequestsSent; + private int lastQuorumWrites; + + private RaftClientBenchmarkContext(RaftClientTestContext context, ReplicaKey otherVoter) { + this.context = context; + this.log = context.log; + this.channel = context.channel; + this.otherVoter = otherVoter; + resetCounters(); + } + + /** + * Builds an unattached node in a {@code voterCount}-node quorum (the local node is not yet the + * leader). Use {@link #unattachedToLeader()} as the measured operation to drive a full + * Unattached → Leader election. A single-voter quorum is rejected because such a node elects + * itself at initialization, before any measured poll. + */ + public static RaftClientBenchmarkContext unattached(int localId, int voterCount) throws Exception { + if (voterCount < 2) { + throw new IllegalArgumentException("voterCount must be at least 2; a single voter self-elects at init"); + } + return new RaftClientBenchmarkContext(buildContext(voterKeys(localId, voterCount)), null); + } + + public static RaftClientBenchmarkContext leader(int localId, int voterCount) throws Exception { + List<ReplicaKey> voterKeys = voterKeys(localId, voterCount); + RaftClientTestContext context = buildContext(voterKeys); + context.unattachedToLeader(); + + return new RaftClientBenchmarkContext(context, voterKeys.get(1)); + } + + /** {@code voterCount} voter keys, each with a random directory id, with {@code localId} first. */ + private static List<ReplicaKey> voterKeys(int localId, int voterCount) { + return IntStream.range(0, voterCount) + .mapToObj(i -> ReplicaKey.of(localId + i, Uuid.randomUuid())) + .collect(Collectors.toList()); + } + + /** + * Builds an unattached node in a KIP-1186 quorum of {@code voterKeys}, whose first entry is the + * local node. + */ + private static RaftClientTestContext buildContext(List<ReplicaKey> voterKeys) throws Exception { + ReplicaKey local = voterKeys.get(0); + VoterSet voters = VoterSetTestUtil.voterSet(voterKeys.stream()); + + return new RaftClientTestContext.Builder(local.id(), local.directoryId().get()) + .withStartingVoters(voters, KRaftVersion.KRAFT_VERSION_1) + .withRaftProtocol(RaftClientTestContext.RaftProtocol.KIP_1186_PROTOCOL) + .withPollIntervalMs(0) + .withUnknownLeader(0) + .build(); + } + + public void unattachedToLeader() throws Exception { + context.unattachedToLeader(); + } + + public void advanceLeaderHwmToLogEnd() throws InterruptedException { + context.advanceLocalLeaderHighWatermarkToLogEndOffset(); + } + + public void pollUntilResponse() throws InterruptedException { + context.pollUntilResponse(); + } + + public int currentEpoch() { + return context.currentEpoch(); + } + + public long logEndOffset() { + return log.endOffset().offset(); + } + + /** A voter other than the local node, e.g. to deliver a FETCH from on the leader. */ + public ReplicaKey otherVoter() { + if (otherVoter == null) { + throw new IllegalStateException("This context was not built with an other-voter key"); + } + return otherVoter; + } + + public void deliverRequest(ApiMessage request) { + context.deliverRequest(request); + } + + public FetchRequestData fetchRequest( + int epoch, + ReplicaKey replicaKey, + long fetchOffset, + int lastFetchedEpoch, + int maxWaitMs + ) { + return context.fetchRequest(epoch, replicaKey, fetchOffset, lastFetchedEpoch, maxWaitMs); + } + + /** Re-baselines all counters to the current totals. Call at the end of benchmark setup. */ + public void resetCounters() { + lastFlushCount = log.flushCount(); + lastReadCount = log.readCount(); + lastTruncationCount = log.truncationCount(); + lastRequestsSent = channel.requestsSent(); + lastQuorumWrites = context.quorumStateWriteCount(); + context.drainAllSentResponses(); + } + + public int drainLogFlushes() { + int current = log.flushCount(); + int delta = current - lastFlushCount; + lastFlushCount = current; + return delta; + } + + public int drainLogReads() { + int current = log.readCount(); + int delta = current - lastReadCount; + lastReadCount = current; + return delta; + } + + public int drainLogTruncations() { + int current = log.truncationCount(); + int delta = current - lastTruncationCount; + lastTruncationCount = current; + return delta; + } + + public int drainRpcRequestsSent() { + int current = channel.requestsSent(); + int delta = current - lastRequestsSent; + lastRequestsSent = current; + return delta; + } + + public int drainQuorumStateWrites() { + int current = context.quorumStateWriteCount(); + int delta = current - lastQuorumWrites; + lastQuorumWrites = current; + return delta; + } + + public int drainRpcResponsesSent() { + return context.drainAllSentResponses(); + } Review Comment: We should have a `lastResponsesSent` and track that like we do with `lastRequestsSent`. ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientBenchmarkContext.java: ########## @@ -0,0 +1,182 @@ +/* + * 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.message.FetchRequestData; +import org.apache.kafka.common.protocol.ApiMessage; +import org.apache.kafka.server.common.KRaftVersion; + +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +public final class RaftClientBenchmarkContext { + private final RaftClientTestContext context; + private final MockLog log; + private final MockNetworkChannel channel; + private final ReplicaKey otherVoter; + + private int lastFlushCount; + private int lastReadCount; + private int lastTruncationCount; + private int lastRequestsSent; + private int lastQuorumWrites; + + private RaftClientBenchmarkContext(RaftClientTestContext context, ReplicaKey otherVoter) { + this.context = context; + this.log = context.log; + this.channel = context.channel; + this.otherVoter = otherVoter; + resetCounters(); + } + + /** + * Builds an unattached node in a {@code voterCount}-node quorum (the local node is not yet the + * leader). Use {@link #unattachedToLeader()} as the measured operation to drive a full + * Unattached → Leader election. A single-voter quorum is rejected because such a node elects + * itself at initialization, before any measured poll. + */ + public static RaftClientBenchmarkContext unattached(int localId, int voterCount) throws Exception { + if (voterCount < 2) { + throw new IllegalArgumentException("voterCount must be at least 2; a single voter self-elects at init"); + } + return new RaftClientBenchmarkContext(buildContext(voterKeys(localId, voterCount)), null); + } + + public static RaftClientBenchmarkContext leader(int localId, int voterCount) throws Exception { + List<ReplicaKey> voterKeys = voterKeys(localId, voterCount); + RaftClientTestContext context = buildContext(voterKeys); + context.unattachedToLeader(); + + return new RaftClientBenchmarkContext(context, voterKeys.get(1)); + } + + /** {@code voterCount} voter keys, each with a random directory id, with {@code localId} first. */ + private static List<ReplicaKey> voterKeys(int localId, int voterCount) { + return IntStream.range(0, voterCount) + .mapToObj(i -> ReplicaKey.of(localId + i, Uuid.randomUuid())) + .collect(Collectors.toList()); + } + + /** + * Builds an unattached node in a KIP-1186 quorum of {@code voterKeys}, whose first entry is the + * local node. + */ + private static RaftClientTestContext buildContext(List<ReplicaKey> voterKeys) throws Exception { + ReplicaKey local = voterKeys.get(0); + VoterSet voters = VoterSetTestUtil.voterSet(voterKeys.stream()); + + return new RaftClientTestContext.Builder(local.id(), local.directoryId().get()) + .withStartingVoters(voters, KRaftVersion.KRAFT_VERSION_1) + .withRaftProtocol(RaftClientTestContext.RaftProtocol.KIP_1186_PROTOCOL) + .withPollIntervalMs(0) + .withUnknownLeader(0) + .build(); + } + + public void unattachedToLeader() throws Exception { + context.unattachedToLeader(); + } + + public void advanceLeaderHwmToLogEnd() throws InterruptedException { + context.advanceLocalLeaderHighWatermarkToLogEndOffset(); + } + + public void pollUntilResponse() throws InterruptedException { + context.pollUntilResponse(); + } + + public int currentEpoch() { + return context.currentEpoch(); + } + + public long logEndOffset() { + return log.endOffset().offset(); + } + + /** A voter other than the local node, e.g. to deliver a FETCH from on the leader. */ + public ReplicaKey otherVoter() { + if (otherVoter == null) { + throw new IllegalStateException("This context was not built with an other-voter key"); + } + return otherVoter; + } + + public void deliverRequest(ApiMessage request) { + context.deliverRequest(request); + } + + public FetchRequestData fetchRequest( + int epoch, + ReplicaKey replicaKey, + long fetchOffset, + int lastFetchedEpoch, + int maxWaitMs + ) { + return context.fetchRequest(epoch, replicaKey, fetchOffset, lastFetchedEpoch, maxWaitMs); + } + + /** Re-baselines all counters to the current totals. Call at the end of benchmark setup. */ + public void resetCounters() { + lastFlushCount = log.flushCount(); + lastReadCount = log.readCount(); + lastTruncationCount = log.truncationCount(); + lastRequestsSent = channel.requestsSent(); + lastQuorumWrites = context.quorumStateWriteCount(); + context.drainAllSentResponses(); + } + + public int drainLogFlushes() { + int current = log.flushCount(); + int delta = current - lastFlushCount; + lastFlushCount = current; + return delta; + } + + public int drainLogReads() { + int current = log.readCount(); + int delta = current - lastReadCount; + lastReadCount = current; + return delta; + } + + public int drainLogTruncations() { + int current = log.truncationCount(); + int delta = current - lastTruncationCount; + lastTruncationCount = current; + return delta; + } + + public int drainRpcRequestsSent() { + int current = channel.requestsSent(); + int delta = current - lastRequestsSent; + lastRequestsSent = current; + return delta; + } + + public int drainQuorumStateWrites() { + int current = context.quorumStateWriteCount(); + int delta = current - lastQuorumWrites; + lastQuorumWrites = current; + return delta; + } + + public int drainRpcResponsesSent() { + return context.drainAllSentResponses(); Review Comment: In general, within a benchmark, we should know which API key for any request/response that is sent by the local node. This is because we as the test/benchmark write control the inputs into the state machine. We should use `RaftClientTestContext#drainSentResponses(ApiKeys apiKey)`, similar to the unit tests. ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientBenchmarkContext.java: ########## @@ -0,0 +1,182 @@ +/* + * 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.message.FetchRequestData; +import org.apache.kafka.common.protocol.ApiMessage; +import org.apache.kafka.server.common.KRaftVersion; + +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +public final class RaftClientBenchmarkContext { + private final RaftClientTestContext context; + private final MockLog log; + private final MockNetworkChannel channel; + private final ReplicaKey otherVoter; + + private int lastFlushCount; + private int lastReadCount; + private int lastTruncationCount; + private int lastRequestsSent; + private int lastQuorumWrites; + + private RaftClientBenchmarkContext(RaftClientTestContext context, ReplicaKey otherVoter) { + this.context = context; + this.log = context.log; + this.channel = context.channel; + this.otherVoter = otherVoter; + resetCounters(); + } + + /** + * Builds an unattached node in a {@code voterCount}-node quorum (the local node is not yet the + * leader). Use {@link #unattachedToLeader()} as the measured operation to drive a full + * Unattached → Leader election. A single-voter quorum is rejected because such a node elects + * itself at initialization, before any measured poll. + */ + public static RaftClientBenchmarkContext unattached(int localId, int voterCount) throws Exception { + if (voterCount < 2) { + throw new IllegalArgumentException("voterCount must be at least 2; a single voter self-elects at init"); + } + return new RaftClientBenchmarkContext(buildContext(voterKeys(localId, voterCount)), null); + } + + public static RaftClientBenchmarkContext leader(int localId, int voterCount) throws Exception { + List<ReplicaKey> voterKeys = voterKeys(localId, voterCount); + RaftClientTestContext context = buildContext(voterKeys); + context.unattachedToLeader(); + + return new RaftClientBenchmarkContext(context, voterKeys.get(1)); + } + + /** {@code voterCount} voter keys, each with a random directory id, with {@code localId} first. */ + private static List<ReplicaKey> voterKeys(int localId, int voterCount) { + return IntStream.range(0, voterCount) + .mapToObj(i -> ReplicaKey.of(localId + i, Uuid.randomUuid())) + .collect(Collectors.toList()); + } + + /** + * Builds an unattached node in a KIP-1186 quorum of {@code voterKeys}, whose first entry is the + * local node. + */ + private static RaftClientTestContext buildContext(List<ReplicaKey> voterKeys) throws Exception { + ReplicaKey local = voterKeys.get(0); + VoterSet voters = VoterSetTestUtil.voterSet(voterKeys.stream()); + + return new RaftClientTestContext.Builder(local.id(), local.directoryId().get()) + .withStartingVoters(voters, KRaftVersion.KRAFT_VERSION_1) + .withRaftProtocol(RaftClientTestContext.RaftProtocol.KIP_1186_PROTOCOL) Review Comment: Hmm, ideally, the benchmark writer should be able to configure the raft protocol and the kraft version. However, for now, we should just use the "latest version" for each via a method call, rather than using the literals directly. ########## raft/src/testFixtures/java/org/apache/kafka/raft/MockQuorumStateStore.java: ########## Review Comment: We also may want a counter for `QuorumState` reads. ########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/ProtocolCounters.java: ########## @@ -0,0 +1,108 @@ +/* + * 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.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; + +/** + * Secondary, machine-independent work counters reported by the raft benchmarks alongside the timing + * score, as {@code benchmark:counter} rows. + * + * <p>Each benchmark calls {@link #drainFrom} every invocation to accumulate the work deltas drained + * from {@link RaftClientBenchmarkContext}. JMH reads the public numeric methods at the end of each + * measurement iteration. The public fields are raw totals intended for JSON post-processing; divide + * them by {@link #operations} to get the exact per-operation value. The public methods return + * per-operation values for each iteration, so the per-iteration console output shows stable values + * such as {@code logReadsPerOp = 1.0}. JMH aggregates {@code Type.EVENTS} secondary results with + * {@code SUM}, so the final summary row will add per-iteration method values together when more + * than one measurement iteration is used. + * + * <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 ProtocolCounters { + public long logFlushesTotal; + public long logReadsTotal; + public long logTruncationsTotal; + public long rpcRequestsSentTotal; + public long rpcResponsesSentTotal; + public long quorumStateWritesTotal; + public long operations; + + @Setup(Level.Iteration) + public void reset() { + logFlushesTotal = 0; + logReadsTotal = 0; + logTruncationsTotal = 0; + rpcRequestsSentTotal = 0; + rpcResponsesSentTotal = 0; + quorumStateWritesTotal = 0; + operations = 0; + } + + /** Accumulates this invocation's work deltas drained from {@code context} into these counters. */ + public void drainFrom(RaftClientBenchmarkContext context) { Review Comment: See if we can annotate this with `@TearDown(Level.Invocation)`. Also, the method signature should be `drainFrom(RaftClientBenchmarkContext context, Optional<List<ApiKey>> requestsSent, Optional<List<ApiKey>> responsesSent`). That way we can we can filter for specific requests/responses we are expecting in a specific order. ########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/ProtocolCounters.java: ########## @@ -0,0 +1,108 @@ +/* + * 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.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; + +/** + * Secondary, machine-independent work counters reported by the raft benchmarks alongside the timing + * score, as {@code benchmark:counter} rows. + * + * <p>Each benchmark calls {@link #drainFrom} every invocation to accumulate the work deltas drained + * from {@link RaftClientBenchmarkContext}. JMH reads the public numeric methods at the end of each + * measurement iteration. The public fields are raw totals intended for JSON post-processing; divide + * them by {@link #operations} to get the exact per-operation value. The public methods return + * per-operation values for each iteration, so the per-iteration console output shows stable values + * such as {@code logReadsPerOp = 1.0}. JMH aggregates {@code Type.EVENTS} secondary results with + * {@code SUM}, so the final summary row will add per-iteration method values together when more + * than one measurement iteration is used. + * + * <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 ProtocolCounters { + public long logFlushesTotal; + public long logReadsTotal; + public long logTruncationsTotal; + public long rpcRequestsSentTotal; + public long rpcResponsesSentTotal; + public long quorumStateWritesTotal; + public long operations; Review Comment: Can we rename this field to `invocations`? ########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/ProtocolCounters.java: ########## @@ -0,0 +1,108 @@ +/* + * 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.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; + +/** + * Secondary, machine-independent work counters reported by the raft benchmarks alongside the timing + * score, as {@code benchmark:counter} rows. + * + * <p>Each benchmark calls {@link #drainFrom} every invocation to accumulate the work deltas drained + * from {@link RaftClientBenchmarkContext}. JMH reads the public numeric methods at the end of each + * measurement iteration. The public fields are raw totals intended for JSON post-processing; divide + * them by {@link #operations} to get the exact per-operation value. The public methods return + * per-operation values for each iteration, so the per-iteration console output shows stable values + * such as {@code logReadsPerOp = 1.0}. JMH aggregates {@code Type.EVENTS} secondary results with + * {@code SUM}, so the final summary row will add per-iteration method values together when more + * than one measurement iteration is used. + * + * <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 ProtocolCounters { + public long logFlushesTotal; + public long logReadsTotal; + public long logTruncationsTotal; + public long rpcRequestsSentTotal; + public long rpcResponsesSentTotal; + public long quorumStateWritesTotal; + public long operations; + + @Setup(Level.Iteration) + public void reset() { + logFlushesTotal = 0; + logReadsTotal = 0; + logTruncationsTotal = 0; + rpcRequestsSentTotal = 0; + rpcResponsesSentTotal = 0; + quorumStateWritesTotal = 0; + operations = 0; + } + + /** Accumulates this invocation's work deltas drained from {@code context} into these counters. */ + public void drainFrom(RaftClientBenchmarkContext context) { + logFlushesTotal += context.drainLogFlushes(); + logReadsTotal += context.drainLogReads(); + logTruncationsTotal += context.drainLogTruncations(); + rpcRequestsSentTotal += context.drainRpcRequestsSent(); + rpcResponsesSentTotal += context.drainRpcResponsesSent(); + quorumStateWritesTotal += context.drainQuorumStateWrites(); + operations += 1; + } + + public double logFlushesPerOp() { Review Comment: Can we rename these methods to `...PerInvocation`? That makes it more clear what they measure. It would also be nice to document why we are doing things this way in this class. ########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/LeaderBenchmark.java: ########## @@ -0,0 +1,67 @@ +/* + * 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.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.concurrent.TimeUnit; + +@State(Scope.Thread) +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +@Warmup(iterations = 5) +@Measurement(iterations = 10) +@Fork(3) +public class LeaderBenchmark { + private static final int LOCAL_ID = 0; Review Comment: We should use a random id. Take a look at how the unit tests are written. ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientTestContext.java: ########## @@ -1061,6 +1076,18 @@ List<RaftRequest.Outbound> assertSentBeginQuorumEpochRequest(int epoch, Set<Inte return requests; } + /** + * Removes and counts all responses the client has sent to inbound requests. Used by the JMH + * raft benchmarks to measure {@code rpcResponsesSent} per operation. Unlike the mock work + * counters, responses are collected here (not in a mock), so draining the collection both + * yields the per-operation delta and bounds its growth across a long benchmark iteration. + */ + int drainAllSentResponses() { + int count = sentResponses.size(); + sentResponses.clear(); + return count; + } + Review Comment: Please remove this method. ########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/LeaderBenchmark.java: ########## @@ -0,0 +1,67 @@ +/* + * 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.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.concurrent.TimeUnit; + +@State(Scope.Thread) +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +@Warmup(iterations = 5) +@Measurement(iterations = 10) +@Fork(3) +public class LeaderBenchmark { + private static final int LOCAL_ID = 0; + private static final int VOTER_COUNT = 3; + + private RaftClientBenchmarkContext context; + + @Setup(Level.Trial) + public void setup() throws Exception { + context = RaftClientBenchmarkContext.leader(LOCAL_ID, VOTER_COUNT); + context.advanceLeaderHwmToLogEnd(); + context.resetCounters(); Review Comment: If you are setting things up at `Level.Trial` you don't need need to call `context.resetCounters`. ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientBenchmarkContext.java: ########## @@ -0,0 +1,182 @@ +/* + * 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.message.FetchRequestData; +import org.apache.kafka.common.protocol.ApiMessage; +import org.apache.kafka.server.common.KRaftVersion; + +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +public final class RaftClientBenchmarkContext { + private final RaftClientTestContext context; + private final MockLog log; + private final MockNetworkChannel channel; + private final ReplicaKey otherVoter; Review Comment: Not every benchmarking method will have 1 remote node which is a voter. We may want to deliver requests from many nodes. We may not have another node at all in some scenarios. This is an assumption that has been hardcoded. If we want to store state about the local/remote replica keys it should be as a list. It looks like we need to for delivering requests like fetch. ########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/LeaderBenchmark.java: ########## @@ -0,0 +1,67 @@ +/* + * 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.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.concurrent.TimeUnit; + +@State(Scope.Thread) +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +@Warmup(iterations = 5) +@Measurement(iterations = 10) +@Fork(3) +public class LeaderBenchmark { + private static final int LOCAL_ID = 0; + private static final int VOTER_COUNT = 3; + + private RaftClientBenchmarkContext context; + + @Setup(Level.Trial) + public void setup() throws Exception { + context = RaftClientBenchmarkContext.leader(LOCAL_ID, VOTER_COUNT); + context.advanceLeaderHwmToLogEnd(); + context.resetCounters(); + } + + /** + * Leader handles a valid FETCH from a fully caught-up follower (fetch offset == log end offset), + * which does not advance the high watermark — the steady-state heartbeat-style fetch. + */ + @Benchmark + public void leaderHandleFetch(ProtocolCounters counters) throws Exception { + int epoch = context.currentEpoch(); + long endOffset = context.logEndOffset(); + context.deliverRequest(context.fetchRequest(epoch, context.otherVoter(), endOffset, epoch, 0)); + context.pollUntilResponse(); + + counters.drainFrom(context); Review Comment: It would be nice if we can drain these counters in a `@TearDown` method so they are not part of the benchmarked code execution. I think it is doable given some of the other changes I've proposed. ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientBenchmarkContext.java: ########## @@ -0,0 +1,182 @@ +/* + * 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.message.FetchRequestData; +import org.apache.kafka.common.protocol.ApiMessage; +import org.apache.kafka.server.common.KRaftVersion; + +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +public final class RaftClientBenchmarkContext { + private final RaftClientTestContext context; + private final MockLog log; + private final MockNetworkChannel channel; + private final ReplicaKey otherVoter; + + private int lastFlushCount; + private int lastReadCount; + private int lastTruncationCount; + private int lastRequestsSent; + private int lastQuorumWrites; + + private RaftClientBenchmarkContext(RaftClientTestContext context, ReplicaKey otherVoter) { + this.context = context; + this.log = context.log; + this.channel = context.channel; + this.otherVoter = otherVoter; + resetCounters(); + } + + /** + * Builds an unattached node in a {@code voterCount}-node quorum (the local node is not yet the + * leader). Use {@link #unattachedToLeader()} as the measured operation to drive a full + * Unattached → Leader election. A single-voter quorum is rejected because such a node elects + * itself at initialization, before any measured poll. + */ + public static RaftClientBenchmarkContext unattached(int localId, int voterCount) throws Exception { + if (voterCount < 2) { + throw new IllegalArgumentException("voterCount must be at least 2; a single voter self-elects at init"); + } + return new RaftClientBenchmarkContext(buildContext(voterKeys(localId, voterCount)), null); + } + + public static RaftClientBenchmarkContext leader(int localId, int voterCount) throws Exception { + List<ReplicaKey> voterKeys = voterKeys(localId, voterCount); + RaftClientTestContext context = buildContext(voterKeys); + context.unattachedToLeader(); + + return new RaftClientBenchmarkContext(context, voterKeys.get(1)); + } + + /** {@code voterCount} voter keys, each with a random directory id, with {@code localId} first. */ + private static List<ReplicaKey> voterKeys(int localId, int voterCount) { + return IntStream.range(0, voterCount) + .mapToObj(i -> ReplicaKey.of(localId + i, Uuid.randomUuid())) + .collect(Collectors.toList()); + } + + /** + * Builds an unattached node in a KIP-1186 quorum of {@code voterKeys}, whose first entry is the + * local node. + */ + private static RaftClientTestContext buildContext(List<ReplicaKey> voterKeys) throws Exception { + ReplicaKey local = voterKeys.get(0); + VoterSet voters = VoterSetTestUtil.voterSet(voterKeys.stream()); + + return new RaftClientTestContext.Builder(local.id(), local.directoryId().get()) + .withStartingVoters(voters, KRaftVersion.KRAFT_VERSION_1) + .withRaftProtocol(RaftClientTestContext.RaftProtocol.KIP_1186_PROTOCOL) + .withPollIntervalMs(0) + .withUnknownLeader(0) + .build(); + } + + public void unattachedToLeader() throws Exception { + context.unattachedToLeader(); + } + + public void advanceLeaderHwmToLogEnd() throws InterruptedException { + context.advanceLocalLeaderHighWatermarkToLogEndOffset(); + } + + public void pollUntilResponse() throws InterruptedException { + context.pollUntilResponse(); + } + + public int currentEpoch() { + return context.currentEpoch(); + } + + public long logEndOffset() { + return log.endOffset().offset(); + } + + /** A voter other than the local node, e.g. to deliver a FETCH from on the leader. */ + public ReplicaKey otherVoter() { + if (otherVoter == null) { + throw new IllegalStateException("This context was not built with an other-voter key"); + } + return otherVoter; + } + + public void deliverRequest(ApiMessage request) { + context.deliverRequest(request); + } + + public FetchRequestData fetchRequest( + int epoch, + ReplicaKey replicaKey, + long fetchOffset, + int lastFetchedEpoch, Review Comment: These are essentially code duplications because `RaftClientTestContext` is a final class which is a parameter in this class' constructor. Instead, we should add a public method to return the underlying `RaftClientTestContext`. Then we can call its methods directly. -- 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]
