bench: restore JsonPipelineBenchmark + AvroPipelineBenchmark (closes #151)#154
Merged
Conversation
…151) Both benchmarks were broken on main since the architecture-deepening pass removed the byte-level MessagePipeline.apply(byte[]) shim — they still declared the field as Function<byte[], byte[]> and called .apply(jsonBytes) on what was now a MessagePipeline<Map<String,Object>> (or <GenericRecord>). The classes didn't compile, which broke compileJmhJava across the whole :benchmarks module. Rewrite the kpipe-side benchmark methods against the current API: final var value = pipeline.deserializeOrFail(bytes); final var result = pipeline.process(value); if (!(result instanceof Result.Passed<T> passed)) { throw new AssertionError("benchmark input should always pass: " + result); } bh.consume(pipeline.serialize(passed.value())); Same per-record work as before (deserialize → operators → serialize), just unrolled to match the new return shape. The pattern-match-or-fail shape keeps the bench loud if the input ever falls into Filtered or Failed — silent assumption breakage would falsify the numbers. Smoke run (1 iter × 1 fork × 1 warmup) numbers, sanity-only: - JSON kpipe pipeline: 663k ops/s - JSON manual SerDe chained: 259k ops/s (the SerDe tax this bench exists to show) - JSON manual single SerDe: 836k ops/s - Avro kpipe pipeline: 867k ops/s - Avro kpipe magic (zero-copy): 658k ops/s - Avro manual magic (copyRange): 1.04M ops/s The single-SerDe-tax story is intact: kpipe ~2.6× faster than the chained manual path at the JSON cell, which is what these benches were designed to surface. Manual side untouched — those paths never used .apply(byte[]).
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #154 +/- ##
============================================
+ Coverage 76.48% 76.52% +0.03%
- Complexity 760 761 +1
============================================
Files 75 75
Lines 2871 2871
Branches 357 357
============================================
+ Hits 2196 2197 +1
Misses 503 503
+ Partials 172 171 -1 ☔ View full report in Codecov by Harness. |
eschizoid
added a commit
that referenced
this pull request
Jun 8, 2026
Two follow-ups from the IDE-inspection sweep, this time covering benchmarks/ and examples/ (the first pass only hit lib/). **Benchmark teardown dedup.** Two invocation-context inner classes in ParallelProcessingBenchmarkInfrastructure (Raw + Share) had the same 17-line shutdown block: lower the `running` flag, join the poll loop, shutdown the executor, await termination (return ignored). IntelliJ flagged it as duplicated-code; the ignored-return was a separate finding sitting in the same blocks. Extract to `stopPollLoopAndExecutor(AtomicBoolean, Thread, ExecutorService)` — single source of truth, and the `awaitTermination` return is now logged at WARNING (don't silently swallow) so a slow executor doesn't disappear from the report. **Plain-text URL in Javadoc.** Two example files had bare `http://...` URLs inside triple-slash markdown doc comments, which IntelliJ flags as plain-text-url-in-javadoc. Wrap in angle brackets so javadoc renders them as autolinks (Markdown URL convention). **Drop now-redundant Json/Avro exclusion.** PR #153 added a benchmarks/build.gradle.kts source-set exclusion to unblock JMH compilation when those benchmarks were broken. PR #154 restored the benchmarks but #153 merged after, leaving the exclusion in place against now-compilable files. Remove the dead exclusion block.
eschizoid
added a commit
that referenced
this pull request
Jun 8, 2026
* chore: address IntelliJ inspection warnings (safe sweep) Mechanical clean-up driven by a multi-agent IDE-inspection survey (workflow over 12 lib modules, 375 findings total). Restricts itself to the categories that are safe to fix without judgment. Three categories addressed: 1. unused-pattern-variable (15 sites): rename `__` to `_` in sealed-Result pattern matches. Java 22+ unnamed pattern; mechanical with no behavior change. 2. redundant-cast (13 → 12 sites): remove casts the compiler can infer from context. The (long) cast on DispatcherIntegrationTest.java:105 stays — it's the boxing path from int to Map<TopicPartition, Long>, where IntelliJ's "redundant" verdict was actually wrong (compile-fail without it). 3. redundant-suppression (14 sites): delete stale @SuppressWarnings annotations that no longer match any real warning. Plus two unresolved-javadoc-symbol fixes (ERROR severity): - KPipeDlqTest: qualify [Builder#withDeadLetterQueue(...)] → [KPipeConsumer.Builder#withDeadLetterQueue(...)] - ParallelDispatcherTest: qualify [#close()] → [ParallelDispatcher#close()] What's NOT in this PR (deferred to per-site review): - unused-parameter (51): almost all are deliberate-ignore in test lambdas; needs `_` rename per-site, not a blanket fix. - try-with-resources (39): rewriting integration tests would change Handle / KPipeConsumer shutdown semantics. - busy-wait (17): intentional poll-until-condition loops in shutdown drain and KEY_ORDERED retry paths. - nullable-handling (16): per-site judgment, especially AvroFormat registry-mode hot path. - unused-method (11): public fluent-facade methods kept as escape hatches; documented in docs/escape-hatches.md. - unchecked-assignment (9): Mockito raw-type captures and Kafka ConsumerRecord/ProducerRecord generics need per-site decisions. Test plan: - ./gradlew test (green locally) - ./gradlew spotlessCheck (green) - ./gradlew compileJava compileTestJava (green) * refactor: dedup common consumer-config block in DefaultSink + DefaultBatchSink Both sinks ended their start() with the same 6-line "apply optional consumer config to the builder" block (retry, backpressure, metrics, errorHandler, deadLetterTopic, pollTimeout). IntelliJ flagged it as a 9-line code-duplication. Extract to a package-private DefaultStream.applyCommonConsumerConfig helper. DefaultSink calls it and then adds its tracer + circuit-breaker (those two are intentionally not on DefaultBatchSink today — adding them would be a behavior change, deferred). * refactor: dedup JMH teardown boilerplate + fix javadoc URL lints Two follow-ups from the IDE-inspection sweep, this time covering benchmarks/ and examples/ (the first pass only hit lib/). **Benchmark teardown dedup.** Two invocation-context inner classes in ParallelProcessingBenchmarkInfrastructure (Raw + Share) had the same 17-line shutdown block: lower the `running` flag, join the poll loop, shutdown the executor, await termination (return ignored). IntelliJ flagged it as duplicated-code; the ignored-return was a separate finding sitting in the same blocks. Extract to `stopPollLoopAndExecutor(AtomicBoolean, Thread, ExecutorService)` — single source of truth, and the `awaitTermination` return is now logged at WARNING (don't silently swallow) so a slow executor doesn't disappear from the report. **Plain-text URL in Javadoc.** Two example files had bare `http://...` URLs inside triple-slash markdown doc comments, which IntelliJ flags as plain-text-url-in-javadoc. Wrap in angle brackets so javadoc renders them as autolinks (Markdown URL convention). **Drop now-redundant Json/Avro exclusion.** PR #153 added a benchmarks/build.gradle.kts source-set exclusion to unblock JMH compilation when those benchmarks were broken. PR #154 restored the benchmarks but #153 merged after, leaving the exclusion in place against now-compilable files. Remove the dead exclusion block. * test: pin Handle defaults + DefaultHandle.startAndWrap exception path Targets the two real coverage gaps in lib/kpipe-api/src/main that codecov flagged: 1. HandleDefaultsTest — Handle.java was at 0% / 4 tracked lines. The defaults aren't decoration: topKeyQueueDepths(n) has a real guard (n <= 0 → IllegalArgumentException) and a real return shape (immutable empty list), and close() commits to a 5-second shutdownGracefully timeout that every try-with-resources caller relies on. The test class uses a minimal Handle subclass that overrides only the abstract methods so the defaults actually run, then pins each: bad n, valid n, default close timeout. 2. DefaultHandleStartAndWrapTest — DefaultHandle.startAndWrap is the only non-delegator in the class (everything else just forwards to the consumer, which is not worth a mock-against-mock test). The factory has a non-trivial error path: if consumer.start() throws, it must close the consumer (so AutoCloseable holds when the caller never gets a Handle back), and if close() also throws, the secondary exception must be addSuppressed onto the original. The test pins happy path, start-failure, start+close-failure (suppressed exception attached), and null-consumer rejection. Skipped on purpose: - DefaultHandle delegators (isHealthy / metrics / awaitShutdown / shutdownGracefully) — testing them with mocks verifies the mock, not behavior. Coverage % would rise, real safety wouldn't.
4 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Resolves #151. Both pipeline benchmarks have been compile-broken on `main` since the architecture-deepening pass removed the byte-level `MessagePipeline.apply(byte[])` shim — they still declared the field as `Function<byte[], byte[]>` and called `.apply(jsonBytes)` on what was now a `MessagePipeline<Map<String, Object>>` / `MessagePipeline`. `compileJmhJava` was failing across the whole `:benchmarks` module on `main`, which is why PR #153 had to exclude these two files to run any benchmark at all.
Changes
Rewrite the kpipe-side benchmark methods against the current pipeline API:
```java
final var value = pipeline.deserializeOrFail(bytes);
final var result = pipeline.process(value);
if (!(result instanceof Result.Passed passed)) {
throw new AssertionError("benchmark input should always pass: " + result);
}
bh.consume(pipeline.serialize(passed.value()));
```
Same per-record work as the old `pipeline.apply(bytes)` (deserialize → operators → serialize), just unrolled to match the new sealed-`Result` return shape. The pattern-match-or-fail keeps the bench loud if input ever falls into `Filtered` or `Failed` — silent assumption breakage would falsify the numbers.
Manual SerDe paths are untouched; they never used `.apply(byte[])`.
Smoke numbers (1 iter × 1 fork × 1 warmup, sanity-only)
The single-SerDe-tax story these benches exist to surface is intact (kpipe `663k` vs manual chained `259k` on JSON, ~2.6×).
Coordination with PR #153
PR #153 added a temporary exclusion of both files in `benchmarks/build.gradle.kts` to unblock the JMH source set. If this PR merges after #153, the exclusion lines should also be removed (they'll point at compilable files at that point). If this PR merges before #153, the exclusion still works but is no longer necessary — #153's rebase can drop it.
Test plan