KAFKA-17645: KIP-1052: Enable warmup in producer performance test #19
Conversation
This commit should enable a test PR against the trunk
Now using an overallStats object with new constructor to contin the combined stats at end of test and a single printTotal method.
Refactor to remove declaration of overallStats and limit its scope to only the location where printTotal is called. Set Stats combine-constructor as package-private. Remove commented code.
- Fix callbacks after switch to steady state which were not being properly created. - Add constant for DEFAULT_REPORTING_INTERVAL_MS - Remove redundant producer.send calls - Add ternerary operator in state assignment - Add check for non-positive numRecords
- bytes are now properly accoutned for. - Refactor to use boolean isSteadyState over warmupRecords to track if a Stats object is of steady state
Stats objects now represent the whole test with 'stats' and the steady state portion of the test with 'steadyStateStats'. Remove merging of Stats latency arrays at end of test.
- Add boolean to stats objects to control if they should print or not - Add DEBUG print to show value of config.warmupRecords
- Use "long" for numRecords and warmupRecords instead of "Long" - Remove "* != null" for above - Remove unused "combiner" constructor
- Remove DEFAULT_REPORTING_INTERVAL_MS as a constructor parameter and move into Stats class; refactor Stats constructors to not use reportingInterval as a parameter - Remove DEBUG print statement
|
I was able to reproduce and fix the performance issue. I had failed to create the record callback for messages in steady state :( It should be fixed in the latest push, but I still think the logic in that conditional can be improved. re: checking warmup records before num records, I've added an exception for when numRecords <= 0. FYI, I will be out for the next week, but thanks so much for looking at this patch!! |
|
Thanks for the latest changes. Results look much better now: 2000000 records sent, 9998.350272 records/sec (0.00 MB/sec), 0.58 ms avg latency, 417.00 ms max latency, 417 ms 50th, 417 ms 95th, 417 ms 99th, 417 ms 99.9th.
1500000 steady state records sent, 9997.733847 records/sec (0.95 MB/sec), 0.52 ms avg latency, 9.00 ms max latency, 1 ms 50th, 1 ms 95th, 1 ms 99th, 2 ms 99.9th.There is still an issue with the overall throughput (0.00 MB/sec) though. I also see some checkstyle errors after rebasing.
No rush, take your time. |
|
Hi @matt-welch, there is one checkstyle error and you also probably need to rebase. |
|
@matt-welch sorry for the delay. Latest changes LGTM. Please, address the comments from @chia7712 and then I think we are good. Thanks. PS: Feel free to ping people for review when you do changes, as we don't get notifications for new commits. |
|
Hi @chia7712 @fvaleri . My latest commit refactors the use of a warmupStats object and a main stats object in favor of having one stats that covers the whole test and a second object, steadyStateStats, that will be used during steady state operation. |
|
@matt-welch thanks for your update. I will take a look later! |
|
Note Other AI code review bot(s) detectedCodeRabbit has detected other AI code review bot(s) in this pull request and will avoid duplicating their findings in the review comments. This may lead to a less comprehensive review. WalkthroughAdds a --warmup-records option to ProducerPerformance, implements steady-state metrics separate from warmup, propagates latencies to both main and steady-state Stats, updates reporting and validation, and adds unit tests for parsing, validation, and steady-state behavior. Changes
Sequence Diagram(s)sequenceDiagram
autonumber
participant User
participant ProducerPerformance
participant Stats_Main as Stats (main)
participant Stats_Steady as Stats (steady)
participant PerfCallback
User->>ProducerPerformance: start with --warmup-records=N
ProducerPerformance->>Stats_Main: initialize main Stats
loop send records
ProducerPerformance->>PerfCallback: send(record)
PerfCallback->>Stats_Main: record(latency)
alt warmup complete (after N)
ProducerPerformance->>Stats_Steady: initialize steadyStateStats
end
alt steadyStateActive
PerfCallback->>Stats_Steady: record(latency)
end
end
ProducerPerformance->>Stats_Main: printTotal()
alt steadyStateStats exists
ProducerPerformance->>Stats_Steady: printTotal() (annotated "steady state")
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes
Pre-merge checks and finishing touches❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✨ Finishing touches
🧪 Generate unit tests
Tip 👮 Agentic pre-merge checks are now available in preview!Pro plan users can now enable pre-merge checks in their settings to enforce checklists before merging PRs.
Please see the documentation for more information. Example: reviews:
pre_merge_checks:
custom_checks:
- name: "Undocumented Breaking Changes"
mode: "warning"
instructions: |
Pass/fail criteria: All breaking changes to public APIs, CLI flags, environment variables, configuration keys, database schemas, or HTTP/GraphQL endpoints must be documented in the "Breaking Change" section of the PR description and in CHANGELOG.md. Exclude purely internal or private changes (e.g., code not exported from package entry points or explicitly marked as internal).Please share your feedback with us on this Discord post. Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
@matt-welch sorry for late review. could you please sync the trunk? |
|
Refacto is reviewing this PR. Please wait for the review comments to be posted. |
|
/refacto-visz |
|
Refacto is reviewing this PR. Please wait for the review comments to be posted. |
Code Review: Producer Performance Warmup Implementation👍 Well Done
📌 Files Processed
📝 Additional Comments
|
![refacto-visz[bot]](https://avatars.githubusercontent.com/u/196302493?s=60&v=4){kind=small}
| if ( config.warmupRecords > 0 && i == config.warmupRecords ) { | ||
| steadyStateStats = new Stats(config.numRecords - config.warmupRecords, config.warmupRecords > 0); | ||
| stats.steadyStateActive = true; | ||
| } | ||
| cb = new PerfCallback(sendStartMs, payload.length, stats, steadyStateStats); |
There was a problem hiding this comment.
Potential NullPointerException Risk
The steadyStateStats variable is passed to PerfCallback constructor even when it's null (when warmupRecords is 0 or i hasn't reached warmupRecords yet). This will cause NullPointerException in PerfCallback.onCompletion() when it tries to access steadyStateStats methods.
if (config.warmupRecords > 0 && i == config.warmupRecords) {
steadyStateStats = new Stats(config.numRecords - config.warmupRecords, true);
stats.steadyStateActive = true;
}
Stats callbackSteadyStateStats = config.warmupRecords > 0 ? steadyStateStats : null;
cb = new PerfCallback(sendStartMs, payload.length, stats, callbackSteadyStateStats);
Commitable Suggestion
| if ( config.warmupRecords > 0 && i == config.warmupRecords ) { | |
| steadyStateStats = new Stats(config.numRecords - config.warmupRecords, config.warmupRecords > 0); | |
| stats.steadyStateActive = true; | |
| } | |
| cb = new PerfCallback(sendStartMs, payload.length, stats, steadyStateStats); | |
| if (config.warmupRecords > 0 && i == config.warmupRecords) { | |
| steadyStateStats = new Stats(config.numRecords - config.warmupRecords, true); | |
| stats.steadyStateActive = true; | |
| } | |
| Stats callbackSteadyStateStats = config.warmupRecords > 0 ? steadyStateStats : null; | |
| cb = new PerfCallback(sendStartMs, payload.length, stats, callbackSteadyStateStats); |
Standards
- ISO-IEC-25010-Reliability-Fault-Tolerance
- ISO-IEC-25010-Functional-Correctness-Appropriateness
- SRE-Error-Handling
| if (warmupRecords >= numRecords) { | ||
| throw new ArgumentParserException("The value for --warmup-records must be strictly fewer than the number of records in the test, --num-records.", parser); | ||
| } |
There was a problem hiding this comment.
Warmup Validation Flaw
The validation allows negative warmup records to pass through since they're less than numRecords. While line 543 sets a default of 0 and the code later uses Math.max(warmupRecords, 0), this validation should explicitly check for negative values to prevent confusing behavior.
if (warmupRecords < 0) {
throw new ArgumentParserException("--warmup-records should be greater than or equal to zero", parser);
}
if (warmupRecords >= numRecords) {
throw new ArgumentParserException("The value for --warmup-records must be strictly fewer than the number of records in the test, --num-records.", parser);
}
Commitable Suggestion
| if (warmupRecords >= numRecords) { | |
| throw new ArgumentParserException("The value for --warmup-records must be strictly fewer than the number of records in the test, --num-records.", parser); | |
| } | |
| if (warmupRecords < 0) { | |
| throw new ArgumentParserException("--warmup-records should be greater than or equal to zero", parser); | |
| } | |
| if (warmupRecords >= numRecords) { | |
| throw new ArgumentParserException("The value for --warmup-records must be strictly fewer than the number of records in the test, --num-records.", parser); | |
| } |
Standards
- Business-Rule-Input-Validation
- Logic-Verification-Boundary-Conditions
| if (steadyStateStats != null) { | ||
| this.steadyStateStats.record(latency, bytes, now); | ||
| this.steadyStateStats.iteration++; | ||
| } |
There was a problem hiding this comment.
Missing Initialization Check
The code checks if steadyStateStats is null before accessing it, but then uses this.steadyStateStats which might still be null if the field wasn't initialized. This inconsistent null check pattern could lead to NullPointerException.
if (steadyStateStats != null) {
steadyStateStats.record(latency, bytes, now);
steadyStateStats.iteration++;
}
Commitable Suggestion
| if (steadyStateStats != null) { | |
| this.steadyStateStats.record(latency, bytes, now); | |
| this.steadyStateStats.iteration++; | |
| } | |
| if (steadyStateStats != null) { | |
| steadyStateStats.record(latency, bytes, now); | |
| steadyStateStats.iteration++; | |
| } |
Standards
- Algorithm-Correctness-Null-Safety
- Logic-Verification-Defensive-Programming
|
|
||
| public Stats(long numRecords, int reportingInterval) { | ||
| public Stats(long numRecords, boolean isSteadyState) { |
There was a problem hiding this comment.
Parameter Name Change
The constructor parameter was changed from 'reportingInterval' to 'isSteadyState' with different types (int to boolean), but the parameter is still used as reportingInterval in the constructor body. This creates a significant maintainability issue as the parameter name no longer reflects its usage.
public Stats(long numRecords, boolean isSteadyState) {
this.start = System.currentTimeMillis();
this.windowStart = System.currentTimeMillis();
this.iteration = 0;
this.count = 0;
this.bytes = 0;
this.maxLatency = 0;
this.totalLatency = 0;
this.windowCount = 0;
this.windowMaxLatency = 0;
this.windowTotalLatency = 0;
this.windowBytes = 0;
this.totalLatency = 0;
this.reportingInterval = 5000;
this.isSteadyState = isSteadyState;
this.steadyStateActive = isSteadyState;
}
Commitable Suggestion
| public Stats(long numRecords, boolean isSteadyState) { | |
| public Stats(long numRecords, boolean isSteadyState) { | |
| this.start = System.currentTimeMillis(); | |
| this.windowStart = System.currentTimeMillis(); | |
| this.iteration = 0; | |
| this.count = 0; | |
| this.bytes = 0; | |
| this.maxLatency = 0; | |
| this.totalLatency = 0; | |
| this.windowCount = 0; | |
| this.windowMaxLatency = 0; | |
| this.windowTotalLatency = 0; | |
| this.windowBytes = 0; | |
| this.totalLatency = 0; | |
| this.reportingInterval = 5000; | |
| this.isSteadyState = isSteadyState; | |
| this.steadyStateActive = isSteadyState; | |
| } |
Standards
- Clean-Code-Meaningful-Names
- SOLID-SRP
- Maintainability-Quality-Interfaces
tools/src/main/java/org/apache/kafka/tools/ProducerPerformance.java
Outdated
Show resolved
Hide resolved
| /* print final results */ | ||
| stats.printTotal(); | ||
| /* print steady-state stats if relevant */ | ||
| if (steadyStateStats != null) { | ||
| steadyStateStats.printTotal(); | ||
| } | ||
| } else { | ||
| // Make sure all messages are sent before printing out the stats and the metrics | ||
| // We need to do this in a different branch for now since tests/kafkatest/sanity_checks/test_performance_services.py |
There was a problem hiding this comment.
Duplicate Code
Identical code blocks for printing stats appear in both branches of the conditional. This violates the DRY principle and creates maintenance burden when changes to the stats printing logic are needed, requiring updates in multiple places.
Standards
- Clean-Code-DRY
- Refactoring-Extract-Method
- Maintainability-Quality-Duplication
| @@ -360,7 +392,9 @@ public Stats(long numRecords, int reportingInterval) { | |||
| this.windowTotalLatency = 0; | |||
| this.windowBytes = 0; | |||
| this.totalLatency = 0; | |||
| this.reportingInterval = reportingInterval; | |||
| this.reportingInterval = 5000; | |||
There was a problem hiding this comment.
Hardcoded Value
The reporting interval is now hardcoded to 5000 instead of being passed as a parameter. This reduces flexibility and makes the code less configurable. The previous parameterized approach was more maintainable as it allowed for different reporting intervals without code changes.
Standards
- Clean-Code-Configuration
- Maintainability-Quality-Flexibility
| Stats stats; | ||
| Stats steadyStateStats; |
There was a problem hiding this comment.
Inconsistent Variable Declaration
The steadyStateStats variable is declared but not initialized, while stats is similarly declared. This inconsistency in variable initialization pattern could lead to confusion and potential null pointer exceptions if not properly managed throughout the code.
Standards
- ISO-IEC-25010-Reliability-Maturity
- ISO-IEC-25010-Functional-Correctness-Appropriateness
| public Stats(long numRecords) { | ||
| this(numRecords, false); | ||
| } |
There was a problem hiding this comment.
Redundant Parameter Check
The Stats constructor with one parameter calls the two-parameter constructor with false for isSteadyState, but later in the code, steadyStateActive is set based on config.warmupRecords > 0. This inconsistency in initialization could lead to incorrect behavior in steady state reporting.
Standards
- ISO-IEC-25010-Functional-Correctness-Appropriateness
- ISO-IEC-25010-Reliability-Maturity
| if (config.warmupRecords > 0) { | ||
| System.out.println("Warmup first " + config.warmupRecords + " records. Steady state results will print after the complete test summary."); | ||
| } | ||
| stats = new Stats(config.numRecords); |
There was a problem hiding this comment.
Stats Constructor Mismatch
The main Stats object is initialized with just numRecords, using the new constructor that defaults isSteadyState to false. However, the steadyStateActive field is later set to true on this object, creating an inconsistent state where the object is not initialized as steady state but is marked as active.
Standards
- Logic-Verification-State-Consistency
- Business-Rule-Object-Initialization
|
/refacto-visz |
|
PR already reviewed at the latest commit: 7def5f4. |
….java Co-authored-by: refacto-visz[bot] <230139763+refacto-visz[bot]@users.noreply.github.com>
|
/refacto-visz |
|
Refacto is reviewing this PR. Please wait for the review comments to be posted. |
Code Review: Warmup Records Implementation👍 Well Done
📌 Files Processed
📝 Additional Comments
|
| if (steadyStateStats != null) { | ||
| this.steadyStateStats.record(latency, bytes, now); | ||
| this.steadyStateStats.iteration++; | ||
| } |
There was a problem hiding this comment.
NullPointerException in PerfCallback.onCompletion
The code checks if steadyStateStats is null before accessing it, but then uses this.steadyStateStats which could still be null. This inconsistent null check pattern could lead to NullPointerException when steadyStateStats is null.
Standards
- ISO-IEC-25010-Reliability-Fault-Tolerance
- SRE-Error-Handling
- DbC-Resource-Mgmt
| if (steadyStateStats != null) { | ||
| this.steadyStateStats.record(latency, bytes, now); | ||
| this.steadyStateStats.iteration++; | ||
| } |
There was a problem hiding this comment.
Steady State Recording Logic
The code records all messages in steadyStateStats regardless of whether they're warmup or steady state records. This defeats the purpose of separating warmup from steady state measurements and will skew the steady state statistics.
Standards
- Algorithm-Correctness-Data-Collection
- Business-Rule-Measurement-Accuracy
| if ( config.warmupRecords > 0 && i == config.warmupRecords ) { | ||
| steadyStateStats = new Stats(config.numRecords - config.warmupRecords, config.warmupRecords > 0); | ||
| stats.steadyStateActive = true; | ||
| } | ||
| cb = new PerfCallback(sendStartMs, payload.length, stats, steadyStateStats); |
There was a problem hiding this comment.
Inconsistent null handling for steadyStateStats
The steadyStateStats variable is passed to PerfCallback constructor before it's initialized when warmupRecords is 0 or before reaching the warmup threshold. This creates a risk of NullPointerException when the callback tries to access steadyStateStats methods.
Standards
- Clean-Code-Defensive-Programming
- SOLID-SRP
- Refactoring-Null-Safety
| steadyStateStats = new Stats(config.numRecords - config.warmupRecords, config.warmupRecords > 0); | ||
| stats.steadyStateActive = true; | ||
| } | ||
| cb = new PerfCallback(sendStartMs, payload.length, stats, steadyStateStats); |
There was a problem hiding this comment.
Callback NPE Risk
The steadyStateStats variable is passed to PerfCallback constructor even when null (when warmupRecords is 0 or i hasn't reached warmupRecords). This creates a null pointer exception risk in PerfCallback.onCompletion() when accessing steadyStateStats methods.
Standards
- CWE-476
- OWASP-A06
| steadyStateStats = new Stats(config.numRecords - config.warmupRecords, config.warmupRecords > 0); | ||
| stats.steadyStateActive = true; |
There was a problem hiding this comment.
Redundant Stats Initialization
The code initializes steadyStateStats with a boolean parameter derived from config.warmupRecords > 0, which is always true at this point. This redundant check creates confusion about the initialization state and could lead to inconsistent behavior.
Standards
- Clean-Code-Logical-Clarity
- Algorithm-Correctness-Initialization
| /* print steady-state stats if relevant */ | ||
| /* print steady-state stats if warmup was configured */ | ||
| if (config.warmupRecords > 0 && steadyStateStats != null) { | ||
| steadyStateStats.printTotal(); | ||
| } |
There was a problem hiding this comment.
Redundant steadyStateStats null checks across multiple code paths
The code contains redundant null checks for steadyStateStats across different code paths with inconsistent conditions. This creates maintenance challenges as changes need to be synchronized across multiple locations and increases cognitive complexity.
Standards
- Clean-Code-DRY
- SOLID-SRP
- Refactoring-Extract-Method
|
|
||
| public Stats(long numRecords, int reportingInterval) { | ||
| public Stats(long numRecords, boolean isSteadyState) { |
There was a problem hiding this comment.
Inconsistent parameter naming in Stats constructor
The constructor parameter was changed from 'reportingInterval' to 'isSteadyState' with different types, but the parameter name no longer reflects its usage. This creates confusion as the parameter name suggests one purpose while the implementation uses it differently.
Standards
- Clean-Code-Meaningful-Names
- SOLID-SRP
- Maintainability-Quality-Interfaces
| if (warmupRecords >= numRecords) { | ||
| throw new ArgumentParserException("The value for --warmup-records must be strictly fewer than the number of records in the test, --num-records.", parser); | ||
| } |
There was a problem hiding this comment.
Negative Warmup Records Validation Missing
The validation allows negative warmup records to pass through since they're less than numRecords. While line 544 sets a default of 0 and Math.max is used later, explicit validation would prevent confusing behavior.
Standards
- ISO-IEC-25010-Reliability-Fault-Tolerance
- SRE-Input-Validation
- DbC-Precondition
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (3)
tools/src/main/java/org/apache/kafka/tools/ProducerPerformance.java (3)
101-106: Announce steady state at the cutover and isolate windowsPrint “Beginning steady state.” immediately at the warmup boundary and reset the main stats window to avoid a mixed warmup/steady-state window. This guarantees the marker appears even for short runs and avoids delayed logging tied to the 5s window.
Apply this diff at the cutover:
- if ( config.warmupRecords > 0 && i == config.warmupRecords ) { - steadyStateStats = new Stats(config.numRecords - config.warmupRecords, config.warmupRecords > 0); - stats.steadyStateActive = true; - } + if (config.warmupRecords > 0 && i == config.warmupRecords) { + // Announce the cutover and reset the main window to prevent mixed warmup/steady-state window output + System.out.println("Beginning steady state."); + stats.newWindow(); + steadyStateStats = new Stats(config.numRecords - config.warmupRecords, true); + stats.steadyStateActive = true; + }
416-421: Remove heuristic “Beginning steady state.” print from record()This message will be printed immediately at the cutover (see prior suggestion). Keeping it here can delay the message up to one reporting interval and risks duplication.
- if (this.isSteadyState && count == windowCount) { - System.out.println("Beginning steady state."); - } if (this.isSteadyState || !this.steadyStateActive) { printWindow(); }
346-355: Clarify help text about negative valuesRuntime coerces negative warmup values to zero via
Math.max(...). Consider making that explicit in the CLI help to avoid surprises.- .help("The number of records to treat as warmup; these initial records will not be included in steady-state statistics. " + - "An additional summary line will be printed describing the steady-state statistics. (default: 0)."); + .help("The number of records to treat as warmup; these initial records will not be included in steady-state statistics. " + + "Negative values are treated as 0. An additional summary line will be printed describing the steady-state statistics. (default: 0).");
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
tools/src/main/java/org/apache/kafka/tools/ProducerPerformance.java(16 hunks)
🔇 Additional comments (7)
tools/src/main/java/org/apache/kafka/tools/ProducerPerformance.java (7)
79-82: LGTM: clear warmup notice and overall Stats initializationThe user-facing message is helpful, and keeping
statsas the overall (warmup + steady state) aggregator aligns with the intended reporting semantics.
127-131: Simplify guard and reduce duplication when printing steady-state totals
config.warmupRecords > 0is redundant sincesteadyStateStatsis only non-null when warmup is configured. Also consider extracting a tiny helper to print totals to avoid duplicating this block in both branches.- /* print steady-state stats if warmup was configured */ - if (config.warmupRecords > 0 && steadyStateStats != null) { + /* print steady-state stats if available */ + if (steadyStateStats != null) { steadyStateStats.printTotal(); }
140-143: DRY: consider extracting a helper to print end-of-test summariesBoth branches print overall totals and optionally steady-state totals. Extracting a small method (e.g.,
printSummaries(stats, steadyStateStats)) will keep behavior consistent and easier to maintain.
383-399: Make reporting interval a named constantHardcoding
5000reduces clarity. Promote it to a constant for readability and discoverability.- this.reportingInterval = 5000; + this.reportingInterval = DEFAULT_REPORTING_INTERVAL_MS;Add this near the top of the Stats class:
private static final long DEFAULT_REPORTING_INTERVAL_MS = 5000L;
471-474: LGTM: steady-state label in total summaryClear labeling of steady-state totals improves readability of the final report.
500-506: LGTM: safe dual-recording into steadyStateStats after warmupNull-guarding
steadyStateStatsensures only steady-state records are captured; iteration increments remain consistent for both stats objects.Also applies to: 517-520
545-560: LGTM: validation and clamping for warmupRecords
numRecords <= 0check prevents degenerate runs.- Clamping
warmupRecordsto non-negative and enforcingwarmupRecords < numRecordsmaintains clear semantics.
3 participants
In order to better analyze steady-state performance of Kafka, this PR enables a warmup in the Producer Performance test. The warmup duration is specified as a number of records that are a subset of the total numRecords. If warmup records is greater than 0, the warmup is represented by a second Stats object which holds warmup results. Once warmup records have been exhausted, the test switches to using the existing Stats object. At end of test, if warmup was enabled, the summary of the whole test (warump + steady state) is printed followed by the summary of the steady-state portion of the test. If no warmup is used, summary prints don't change from existing behavior.
This contribution is an original work and is licensed to the Kafka project under the Apache license
Testing strategy comprises new Java unit tests added to ProducerPerformanceTests.java.
Committer Checklist (excluded from commit message)
Summary by CodeRabbit
New Features
Bug Fixes
Tests