KAFKA-19672: Fix aggregate statistics parsing in producer performance…

[Alwaysgaurav1]

Committer Checklist (fill in after opening the PR)

• Added Jira: https://issues.apache.org/jira/browse/KAFKA-19672
• Ran relevant unit/system tests

Description

In Apache Kafka system test benchmarks (kafkatest/benchmarks/core/benchmark_test.py), ProducerPerformanceService runs org.apache.kafka.tools.ProducerPerformance and parses output from stdout. When tests completed or encountered early termination, they could fail with IndexError: list index out of range and Unable to parse aggregate performance statistics on node %d.

This PR fixes multiple related issues identified in KAFKA-19672:

1. parse_stats(line) handling of window lines:

   • ProducerPerformance.java emits periodic window metrics (printWindow()) with 4 comma-separated values, whereas the aggregate summary (printTotal()) emits 8 values including percentiles (50th, 95th, 99th, 99.9th).
   • parse_stats(line) accessed parts[4] through parts[7] unconditionally, throwing IndexError when encountering window lines. This broke self.intermediate_stats accumulation and failed whenever a window line was evaluated as the final stat.
   • Updated parse_stats to parse 4-element window stat lines gracefully and only include percentile fields when present (len(parts) >= 8).

2. Accurate total statistics tracking:

   • Instead of blindly assigning last = line for every line in STDOUT_CAPTURE, the parsing loop now tracks lines containing latency_50th_ms, ensuring extraneous lines (such as --print-metrics tables, log statements, or trailing empty lines) do not overwrite the aggregate results.
   • When intermediate_stats=True, window stats are properly collected into self.stats[idx-1].

3. Stderr diagnostics on failure:

   • If no total statistics line was emitted (e.g., the process crashed or failed early), STDERR_CAPTURE is read and attached to the raised Exception to surface the root cause immediately rather than obscuring it with a parse failure.

4. Suppress late window reporting in ProducerPerformance.java:

   • Added this.suppressPrint = true; in Stats.printTotal(). This ensures that callbacks or shutdown flushes occurring after the total report has been printed do not emit trailing window reports.
   • Added unit test testSuppressPrintAfterPrintTotal in ProducerPerformanceTest.java.

5. Hardened ConsumerPerformanceService & ShareConsumerPerformanceService:

   • Applied similar robust parsing to skip trailing logs/blank lines and capture stderr on failure.

Testing

• Ran ./gradlew :tools:checkstyleMain :tools:checkstyleTest :tools:spotbugsMain :tools:test --tests "org.apache.kafka.tools.ProducerPerformanceTest" (all 43 tests passed cleanly).
• Verified window and total line parsing, steady-state parsing, and stderr reporting across ducktape services.