From aef5a03fe4217232c4563d3c0e4bd1b4c7b5d59c Mon Sep 17 00:00:00 2001 From: Jameel Al-Aziz Date: Wed, 29 Jul 2026 23:09:47 -0700 Subject: [PATCH] fix: always panic in Context.Fail Context.Fail is documented to stop the callback immediately by panicking, but syncFailer skipped the panic when the partition's run context was already Done and returned normally instead. The run context of a partition processor descends from the sarama session context, so it is cancelled whenever a rebalance revokes the partition. A callback that is in flight at that moment, notices the cancellation and reports it through ctx.Fail therefore resumed execution and returned, letting processMessage reach msgContext.finish(nil). The error had been recorded in run's local rerr and never reached the cbContext, so tryCommit found no error and marked the offset of a message that was never processed successfully. Sarama flushes marked offsets while releasing the session, before rejoining the group, so the commit is accepted for the still valid generation and the next owner of the partition starts after the message, which is therefore lost. run does return an errProcessing and the processor shuts down, but only after the offset has been marked, so the shutdown does not prevent this. Panicking unconditionally restores the documented contract and skips the commit, because the panic keeps processMessage from calling finish. It is recovered by the existing handler in run, which records the error and performs the waitgroup accounting, so the processor still terminates with an errProcessing as before. This intentionally does not make the commit conditional on the run context being alive: a callback that completes successfully has already emitted its messages and written its table updates, so its offset must still be committed even if the partition was revoked in the meantime. --- partition_processor.go | 18 ++--- partition_processor_test.go | 138 ++++++++++++++++++++++++++++++++++++ 2 files changed, 144 insertions(+), 12 deletions(-) create mode 100644 partition_processor_test.go diff --git a/partition_processor.go b/partition_processor.go index 6a4a53b3..d6fc82bd 100644 --- a/partition_processor.go +++ b/partition_processor.go @@ -343,19 +343,13 @@ func (pp *PartitionProcessor) run(ctx context.Context) (rerr error) { var ( // syncFailer is called synchronously from the callback within *this* - // goroutine + // goroutine. + // It must panic unconditionally, as documented by Context.Fail: the panic is + // what stops the callback and prevents processMessage from reaching + // msgContext.finish, which would commit the offset of a message that was + // never processed successfully. The panic is recovered below, which also + // takes care of the waitgroup accounting. syncFailer = func(err error) { - // only fail processor if context not already Done - select { - case <-ctx.Done(): - mutexErr.Lock() - rerr = multierror.Append(rerr, - newErrProcessing(pp.partition, fmt.Errorf("synchronous error in callback: %w", err)), - ) - mutexErr.Unlock() - return - default: - } panic(err) } diff --git a/partition_processor_test.go b/partition_processor_test.go new file mode 100644 index 00000000..3ca12bd3 --- /dev/null +++ b/partition_processor_test.go @@ -0,0 +1,138 @@ +package goka + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/lovoo/goka/codec" + "github.com/stretchr/testify/require" +) + +// newTestPartitionProcessor creates a minimal stateless partition processor that +// records the offsets it commits instead of marking them in a consumer group +// session. +func newTestPartitionProcessor(t *testing.T, cb ProcessCallback) (*PartitionProcessor, func() []int64) { + t.Helper() + + opts := new(poptions) + opts.log = defaultLogger + opts.contextWrapper = func(ctx Context) Context { return ctx } + opts.partitionChannelSize = defaultPartitionChannelSize + + var ( + mCommitted sync.Mutex + committed []int64 + ) + commit := func(msg *message, meta string) { + mCommitted.Lock() + defer mCommitted.Unlock() + committed = append(committed, msg.offset) + } + + ctrl, bm := createMockBuilder(t) + t.Cleanup(ctrl.Finish) + + pp := newPartitionProcessor(0, + DefineGroup("test-group", Input("input", new(codec.Int64), cb)), + commit, defaultLogger, opts, runModeActive, + nil, NewMockAutoConsumer(t, nil), bm.producer, bm.tmgr, + NewSimpleBackoff(defaultBackoffStep, defaultBackoffMax), time.Minute) + + return pp, func() []int64 { + mCommitted.Lock() + defer mCommitted.Unlock() + return append([]int64(nil), committed...) + } +} + +// runPartitionProcessor starts pp.run and returns a function that waits for it +// to terminate, returning its error. +func runPartitionProcessor(t *testing.T, pp *PartitionProcessor, ctx context.Context) func() error { + t.Helper() + + runDone := make(chan error, 1) + go func() { runDone <- pp.run(ctx) }() + + return func() error { + t.Helper() + select { + case err := <-runDone: + return err + case <-time.After(30 * time.Second): + t.Fatal("partition processor did not shut down") + return nil + } + } +} + +// TestPartitionProcessor_FailWithCancelledContext verifies that Context.Fail +// aborts the callback and skips the commit even when the partition context has +// already been cancelled, e.g. because a rebalance revoked the partition while +// the callback was in flight. +// +// Fail used to return normally in that situation, which let processMessage reach +// msgContext.finish and commit the offset of a message that was never processed +// successfully. The message was then never redelivered to the new owner of the +// partition. +func TestPartitionProcessor_FailWithCancelledContext(t *testing.T) { + var ( + callbackEntered = make(chan struct{}) + resumedAfterFail bool + ) + + cb := func(ctx Context, msg interface{}) { + close(callbackEntered) + // emulate a blocking request that is aborted when the partition context + // is cancelled + <-ctx.Context().Done() + ctx.Fail(fmt.Errorf("request aborted: %w", ctx.Context().Err())) + resumedAfterFail = true + } + + pp, committed := newTestPartitionProcessor(t, cb) + + ctx, cancel := context.WithCancel(context.Background()) + waitRun := runPartitionProcessor(t, pp, ctx) + + pp.input <- &message{topic: "input", partition: 0, offset: 42, key: "key", value: []byte("1")} + + <-callbackEntered + cancel() + + require.Error(t, waitRun(), "run must report the processing error") + require.False(t, resumedAfterFail, "Fail must stop the callback by panicking") + require.Empty(t, committed(), "offset of a failed message must not be committed") +} + +// TestPartitionProcessor_SuccessWithCancelledContext verifies the counterpart of +// TestPartitionProcessor_FailWithCancelledContext: a callback that completes +// successfully has already emitted its messages and written its table updates, +// so its offset must be committed even if the partition context was cancelled +// meanwhile. Withholding the commit would turn completed work into a guaranteed +// reprocessing by the next owner of the partition. +func TestPartitionProcessor_SuccessWithCancelledContext(t *testing.T) { + callbackEntered := make(chan struct{}) + + cb := func(ctx Context, msg interface{}) { + close(callbackEntered) + // the partition is revoked while the callback runs, but the callback does + // not depend on the context and completes successfully + <-ctx.Context().Done() + } + + pp, committed := newTestPartitionProcessor(t, cb) + + ctx, cancel := context.WithCancel(context.Background()) + waitRun := runPartitionProcessor(t, pp, ctx) + + pp.input <- &message{topic: "input", partition: 0, offset: 7, key: "key", value: []byte("1")} + + <-callbackEntered + cancel() + waitRun() + + require.Equal(t, []int64{7}, committed(), "successfully processed message must be committed") +}