From 62d13e41e6883696e528aa90a84cc51401976709 Mon Sep 17 00:00:00 2001 From: mariano Date: Mon, 22 Jun 2026 16:47:05 -0500 Subject: [PATCH] chore(consumer): reflow stranded comment lines in CrashRestartReprocessingIntegrationTest MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three '//' comment blocks had mid-sentence single words stranded on their own lines (spotless wraps at ~100 chars and pushed the overflow word down). Re-wrapped each block so every line stays under the wrap width with natural multi-word continuations. Cosmetic only — no code or assertion changes. (Sentence-final single-word lines are left as-is.) --- ...ashRestartReprocessingIntegrationTest.java | 26 ++++++++----------- 1 file changed, 11 insertions(+), 15 deletions(-) diff --git a/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/CrashRestartReprocessingIntegrationTest.java b/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/CrashRestartReprocessingIntegrationTest.java index 08eeffca..b661464e 100644 --- a/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/CrashRestartReprocessingIntegrationTest.java +++ b/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/CrashRestartReprocessingIntegrationTest.java @@ -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(); @@ -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; } @@ -312,13 +311,10 @@ private KPipeConsumer 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 -> {