Skip to content

chore(consumer): reflow stranded comment lines in CrashRestartReprocessingIntegrationTest #222

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
eschizoid merged 1 commit into from
Jun 22, 2026
Merged

chore(consumer): reflow stranded comment lines in CrashRestartReprocessingIntegrationTest #222

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 11 additions & 15 deletions ...test/java/io/github/eschizoid/kpipe/consumer/CrashRestartReprocessingIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -151,11 +151,10 @@ void atLeastOnceSurvivesHardCrashRestart() throws Exception {
);
final var observedByABeforeCrash = Set.copyOf(observedA);

// CRASH: stop the manager (no further commit; markOffsetProcessed becomes a no-op) and abandon
// A's consumer thread without a graceful drain — no shutdownGracefully(). On interrupt A's loop
// closes its Kafka consumer in its finally, which leaves the group promptly; B then inherits
// the
// partition and resumes from the manually-committed prefix.
// CRASH: stop the manager (no further commit; markOffsetProcessed becomes a no-op) and
// abandon A's consumer thread without a graceful drain (no shutdownGracefully()). On
// interrupt, A's loop closes its Kafka consumer in the finally, which leaves the group
// promptly; B then inherits the partition and resumes from the manually-committed prefix.
offsetManagerA.get().stop();
threadA.interrupt();

Expand Down Expand Up @@ -230,9 +229,9 @@ void atLeastOnceSurvivesHardCrashRestart() throws Exception {
final var tp = entry.getKey();
final var logEnd = entry.getValue();
final var committedMeta = committedFinal.get(tp);
// A missing commit is an acceptable uncommitted tail (reprocessed on a later restart);
// no-loss
// above already covers it. Only bound-check partitions that actually committed.
// A missing commit is an acceptable uncommitted tail (reprocessed on a later
// restart); no-loss above already covers it. Only bound-check partitions that
// actually committed.
if (committedMeta == null) {
continue;
}
Expand Down Expand Up @@ -312,13 +311,10 @@ private KPipeConsumer<byte[]> buildConsumerWithManagedOffsets(
.withProperties(consumerProperties(groupId))
.withTopic(topic)
// SEQUENTIAL on purpose: in-order processing means the committed prefix is a clean
// contiguous
// range, and combined with the small max.poll.records (see consumerProperties) the
// consumer
// drains the command queue between small batches so the manual commit lands as a
// bounded
// prefix. At-least-once is mode-independent; the PARALLEL path is covered by the offset
// property/stress/jcstress suites.
// contiguous range, and combined with the small max.poll.records (see
// consumerProperties) the consumer drains the command queue between small batches so
// the manual commit lands as a bounded prefix. At-least-once is mode-independent; the
// PARALLEL path is covered by the offset property/stress/jcstress suites.
.withProcessingMode(ProcessingMode.SEQUENTIAL)
.withPipeline(
TestPipelines.sideEffect(value -> {
Expand Down
Loading