feat: kpipe-test module — TestStream<T> for Docker-free pipeline testing#231
Open
eschizoid wants to merge 2 commits into
Open
feat: kpipe-test module — TestStream<T> for Docker-free pipeline testing#231eschizoid wants to merge 2 commits into
eschizoid wants to merge 2 commits into
Conversation
… Docker-free pipeline tests New published module lib/kpipe-test (io.github.eschizoid.kpipe.test, JPMS module of the same name) implementing the PLAN P2 surface: - TestStream<T>: MockConsumer-backed driver around a real KPipeConsumer (PLAN drive-model option b), builder(format) → pipe/filter/peek/toCustom/toBatch → build() → send/sendRaw → flush() → close(). SEQUENTIAL by default so capture order matches send order; withProcessingMode overrides for PARALLEL / KEY_ORDERED. - Deterministic flush(): waits until every sent record is polled, terminal (processed / filtered / failed) or batch-buffered, and the dispatcher in-flight count has drained (waitForInFlightDrain + metrics accounting), so assertions never race the consumer thread. - Batch v1 semantics per PLAN: size-triggered flushes observable after flush(), trailing partial batch drains on close() (shutdown-flush); no age-based flush (policy age cap set beyond test scale). - CapturingSink<T>: thread-safe capture with immutable captured() snapshots; no assertion DSL by design. - errors() records every ProcessingError (operator throw, malformed payload, null value) ahead of any user error handler; metrics() exposes the consumer metric snapshot. Wired into settings.gradle.kts and the BOM; build.gradle.kts and module-info.java modeled on the sibling format modules, publishing via the existing :lib subprojects machinery. 29 new Docker-free tests, including the PLAN target-ergonomics sketch verbatim, filter and error paths, batch size/shutdown flush, and non-sequential modes.
Owner
Author
|
@copilot please review |
Copilot stopped work on behalf of
eschizoid due to an error
July 10, 2026 23:18
- README: kpipe-test row in the module catalog, and a "Testing your pipeline" subsection under Testing showing the TestStream + CapturingSink flow (deterministic flush, filter/error visibility); the existing docker-compose walkthrough moves under an "Integration testing against a real broker" heading.
❌ 1 Tests Failed:
View the top 1 failed test(s) by shortest run time
To view more test analytics, go to the Test Analytics Dashboard |
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.
What
The top adoption-friction item from the roadmap (P2 #1): testing a KPipe pipeline no longer requires Testcontainers/Docker or a hand-rolled MockConsumer harness.
Drive model (per the roadmap's recorded decision): a real
KPipeConsumerover a seededMockConsumer— tests exercise the production dispatcher/offset/sink paths, not a simulation. Default SEQUENTIAL, PARALLEL/KEY_ORDERED overridable.flush(): polled → accounted (processed+errors+inFlight) → drain, with post-drain re-read;AssertionErrorwith metrics diagnostic on timeout — no sleeps, no flakes.CapturingSink<T>— thread-safe, snapshotcaptured(), no assertion DSL (per the roadmap decision: users bring AssertJ/JUnit).lib/kpipe-test, JPMS moduleio.github.eschizoid.kpipe.test, wired into settings/BOM/publishing like siblings.Tests
29 new, all Docker-free: the PLAN ergonomics sketch verbatim, ordering, operator chaining, filter semantics, operator-throw/malformed/null error paths, throwing error handler, PARALLEL & KEY_ORDERED delivery, lifecycle/builder validation, batch flush semantics, concurrent captures. Full
:lib:kpipe-consumer:testsuite green on the branch (392/392, exercises #229's PendingOffsetSet).Notes
peek,withProcessingMode,errors(), etc.) — needed by the mandated error-path tests.kpipe-tracing-otel/kpipe-schema-registry-confluentmissing from the BOM.