From 3c271f0770b53f3d086b2a73299a7aa46e84f1d2 Mon Sep 17 00:00:00 2001 From: Joshua Temple Date: Sat, 22 Aug 2026 22:28:10 -0400 Subject: [PATCH 1/2] =?UTF-8?q?feat(source):=20v1=20hardening=20=E2=80=94?= =?UTF-8?q?=20tx-Term=20contract,=20dependency=20isolation,=20test-gap=20c?= =?UTF-8?q?losure?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P2-9 (decided + implemented + documented): settling Term directly on a transactional subscription is rejected with the new exported ErrTermInsideTransaction — a DLQ produce outside the open EOS session would break atomicity with the consumed offset. The supported poison path is inside Transactional.Begin: produce to the dead-letter topic via the handed Tx and return success, committing DLQ record and offset atomically. Documented in kafka.go, README divergence section, Begin doc; pinned by unit tests for both the rejection and the Begin DLQ pattern. P2-7: Hopper.exitErr drops its ignored ctx parameter. P2-8: the RedPanda integration leg moves to its own test-only nested module source/kafka/integration with its own go.mod, so testcontainers-go, redpanda, and kadm no longer appear in source/kafka's require block and downstream module graphs stop inheriting them. Workflow matrix and mage Integration target follow the new path; the -tags integration convention and Docker-skip guard are kept. P2-10/P2-11 comments landed with checkpoint 1 (requeue cost model; poll-error record discard). Test gaps closed: drain-on-cancel with mid-flight backend settle (Close+cancel race: exactly one settle, zero post-Close yields, no stranded goroutines); duplicate-settle idempotence on both adapters (memsource ledger + kafka marks); rebalance revoke-with-in-flight end-to-end on RedPanda (two members join/leave, 40 records delivered exactly once across the group); tx-Term pair above. Also fixes vet shadow warnings in checkpoint-1's drain tests. --- .github/workflows/source.yml | 14 +- magefiles/magefile.go | 12 +- source/hopper.go | 16 +- source/hopper_test.go | 124 +++++ source/kafka/README.md | 11 +- source/kafka/adapter_internal_test.go | 58 ++ source/kafka/capability.go | 6 + source/kafka/go.mod | 55 +- source/kafka/go.sum | 139 ----- source/kafka/integration/integration_test.go | 535 +++++++++++++++++++ source/kafka/kafka.go | 16 +- source/kafka/subscription.go | 9 + source/memsource/memsource_test.go | 42 ++ 13 files changed, 826 insertions(+), 211 deletions(-) create mode 100644 source/kafka/integration/integration_test.go diff --git a/.github/workflows/source.yml b/.github/workflows/source.yml index faf8d98..02adbcb 100644 --- a/.github/workflows/source.yml +++ b/.github/workflows/source.yml @@ -103,11 +103,13 @@ jobs: awk -v p="$pct" -v t="$THRESHOLD" 'BEGIN { exit (p+0 < t+0) ? 1 : 0 }' \ || { echo "::error::coverage ${pct}% is below the ${THRESHOLD}% threshold"; exit 1; } - # Integration / E2E leg, behind the `integration` build tag, run with - # GOWORK=off. The in-workspace source modules run end-to-end; the SDK-backed - # source modules drive a real broker via testcontainers (Docker is available on - # GitHub-hosted ubuntu runners) and skip cleanly if a daemon is ever absent. - # Kept off the default test matrix so the hermetic checks stay fast. + # Integration / E2E leg, run with GOWORK=off. The in-workspace source modules + # run end-to-end; the SDK-backed source modules drive a real broker via + # testcontainers (Docker is available on GitHub-hosted ubuntu runners) and skip + # cleanly if a daemon is ever absent. source/kafka's RedPanda leg lives in the + # test-only nested module source/kafka/integration so its testcontainers, + # redpanda, and kadm dependencies never enter source/kafka's go.mod. Kept off + # the default test matrix so the hermetic checks stay fast. integration: strategy: fail-fast: false @@ -115,7 +117,7 @@ jobs: module: - source - source/statemachine - - source/kafka + - source/kafka/integration - source/jetstream - source/redis - source/cloudevents diff --git a/magefiles/magefile.go b/magefiles/magefile.go index 8b0e6c9..bf4c272 100644 --- a/magefiles/magefile.go +++ b/magefiles/magefile.go @@ -445,8 +445,16 @@ var integrationModules = func() []string { mods := []string{"sink/file", "sink/http", "sink/prometheus", "sink/slog"} mods = append(mods, sinkDestinations...) mods = append(mods, "source", "source/statemachine") - mods = append(mods, sourceDestinations...) - return mods + for _, mod := range sourceDestinations { + if mod == "source/kafka" { + // The RedPanda leg lives in its own test-only module so its + // testcontainers/redpanda/kadm deps stay out of source/kafka's + // go.mod; appended below under its nested path. + continue + } + mods = append(mods, mod) + } + return append(mods, "source/kafka/integration") }() // Integration runs the //go:build integration leg for every sink and source diff --git a/source/hopper.go b/source/hopper.go index 2df4eb0..cf9ef1c 100644 --- a/source/hopper.go +++ b/source/hopper.go @@ -209,7 +209,7 @@ func (hp *Hopper) run(ctx context.Context, sub Subscription, h Handler) error { select { case inFlight <- struct{}{}: case <-runCtx.Done(): - return hp.exitErr(ctx, runErr) + return hp.exitErr(runErr) } } @@ -219,10 +219,10 @@ func (hp *Hopper) run(ctx context.Context, sub Subscription, h Handler) error { <-inFlight // release the slot we reserved but did not use } if errors.Is(err, ErrDrained) || errors.Is(err, context.Canceled) || runCtx.Err() != nil { - return hp.exitErr(ctx, runErr) + return hp.exitErr(runErr) } errOnce.Do(func() { runErr = err }) - return hp.exitErr(ctx, runErr) + return hp.exitErr(runErr) } hp.received.Add(runCtx, 1) @@ -236,7 +236,7 @@ func (hp *Hopper) run(ctx context.Context, sub Subscription, h Handler) error { if inFlight != nil { <-inFlight } - return hp.exitErr(ctx, runErr) + return hp.exitErr(runErr) } workWG.Add(1) @@ -250,17 +250,15 @@ func (hp *Hopper) run(ctx context.Context, sub Subscription, h Handler) error { if inFlight != nil { <-inFlight } - return hp.exitErr(ctx, runErr) + return hp.exitErr(runErr) } } } // exitErr maps a run's terminal error onto the public contract: a clean drain // ([ErrDrained]), a context cancellation, and a Close are all graceful and -// return nil; only a genuine fetch error propagates. The parent context is -// accepted to document that its cancellation is the expected, non-error -// shutdown path. -func (hp *Hopper) exitErr(_ context.Context, runErr error) error { +// return nil; only a genuine fetch error propagates. +func (hp *Hopper) exitErr(runErr error) error { return runErr } diff --git a/source/hopper_test.go b/source/hopper_test.go index fcee6fe..62c9097 100644 --- a/source/hopper_test.go +++ b/source/hopper_test.go @@ -591,3 +591,127 @@ func TestHopper_RedeliveryStormBoundedResources(t *testing.T) { time.Sleep(10 * time.Millisecond) } } + +// settleBlockingSub mimics a backend whose Settle honors ctx and can be slow — +// a Kafka commit or JetStream ack mid-round-trip when shutdown lands. Next +// yields the queued messages, then blocks on ctx; after Close it never yields +// again. Settle blocks until the context is canceled, then completes exactly +// once, so the test can prove the engine neither strands nor double-settles +// in-flight work across a Close+cancel race. +type settleBlockingSub struct { + mu sync.Mutex + msgs []source.Message + closed bool + nextCalls int + yieldsAfterClose int + + settleStarted chan struct{} + startOnce sync.Once + settleCalls atomic.Int32 +} + +func newSettleBlockingSub(msgs ...source.Message) *settleBlockingSub { + return &settleBlockingSub{msgs: msgs, settleStarted: make(chan struct{})} +} + +func (s *settleBlockingSub) Next(ctx context.Context) (source.Message, error) { + s.mu.Lock() + s.nextCalls++ + closed := s.closed + if len(s.msgs) > 0 { + m := s.msgs[0] + s.msgs = s.msgs[1:] + if closed { + s.yieldsAfterClose++ + } + s.mu.Unlock() + return m, nil + } + s.mu.Unlock() + if closed { + return nil, source.ErrDrained + } + <-ctx.Done() + return nil, ctx.Err() +} + +func (s *settleBlockingSub) Settle(ctx context.Context, _ source.Message, _ source.Result) error { + s.settleCalls.Add(1) + s.startOnce.Do(func() { close(s.settleStarted) }) + <-ctx.Done() // the backend settle only completes when its context ends + return nil +} + +func (s *settleBlockingSub) Close() error { + s.mu.Lock() + s.closed = true + s.mu.Unlock() + return nil +} + +func (s *settleBlockingSub) yieldsAfterCloseCount() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.yieldsAfterClose +} + +// TestHopper_DrainOnCancelWithMidFlightSettle pins the shutdown contract when a +// backend settle is mid-flight: a Close and a context cancel racing must leave +// exactly one settle (no double-settle), yield nothing new after Close, strand +// no goroutines, and let Run return. +func TestHopper_DrainOnCancelWithMidFlightSettle(t *testing.T) { + t.Parallel() + + sub := newSettleBlockingSub(testMsg{key: []byte("k"), value: []byte("v")}) + hp := source.New(source.WithConcurrency(1)) + t.Cleanup(func() { _ = hp.Close() }) + + baseline := runtime.NumGoroutine() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { + done <- hp.Run(ctx, sub, func(context.Context, source.Message) source.Result { + return source.Ack() + }) + }() + + select { + case <-sub.settleStarted: + case <-time.After(3 * time.Second): + t.Fatal("message never reached its (blocking) settle") + } + + // Race graceful Close against cancellation, like an operator shutdown and + // a deadline expiring at once. + var wg sync.WaitGroup + wg.Add(2) + go func() { defer wg.Done(); _ = sub.Close() }() + go func() { defer wg.Done(); cancel() }() + wg.Wait() + + select { + case err := <-done: + if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, source.ErrDrained) { + t.Fatalf("Run = %v, want nil, context.Canceled, or ErrDrained", err) + } + case <-time.After(3 * time.Second): + t.Fatal("Run did not return after the Close+cancel race") + } + + if got := sub.settleCalls.Load(); got != 1 { + t.Fatalf("settle calls = %d, want exactly 1 (no double-settle)", got) + } + if got := sub.yieldsAfterCloseCount(); got != 0 { + t.Fatalf("Next yielded %d messages after Close, want 0", got) + } + deadline := time.Now().Add(2 * time.Second) + for runtime.NumGoroutine() > baseline+2 { + if time.Now().After(deadline) { + t.Fatalf("goroutines after run = %d, want within +2 of baseline %d (stranded in-flight)", + runtime.NumGoroutine(), baseline) + } + time.Sleep(10 * time.Millisecond) + } +} diff --git a/source/kafka/README.md b/source/kafka/README.md index 2346959..ce5d3f5 100644 --- a/source/kafka/README.md +++ b/source/kafka/README.md @@ -35,7 +35,7 @@ and the marked offsets are committed on graceful drain and on rebalance | `Ack` | mark the record for commit (commit-after-process) | | `Nak` | never mark; pause, re-seek to the record's offset, resume — the record is fetched again in this session | | `NakAfter(d)` | same as `Nak`, waiting out `d` between pause and re-seek (best-effort) | -| `Term` | produce the record to the dead-letter topic, then mark | +| `Term` | produce the record to the dead-letter topic, then mark (transactional subscriptions: rejected with `ErrTermInsideTransaction`; route poison through `Begin` instead) | | `InProgress` | no-op (Kafka has no per-message ack deadline) | | `Manual` | no-op (the handler settled via `Message.As` + the client) | @@ -55,6 +55,15 @@ concurrently committed higher offset can advance past a nacked record, so cross-restart redelivery is **best-effort**, not guaranteed. This is the one divergence from JetStream's native nak semantics. +### Divergence: Term inside a transaction + +On a subscription built with `WithTransactional`, settling `Term` directly is +rejected with `ErrTermInsideTransaction`: a DLQ produce outside the open EOS +session would not be atomic with the consumed offset. Inside +`Transactional.Begin`, produce the poison record to the dead-letter topic via +the handed `source.Tx` and return nil — the DLQ record and the input offset +then commit as one atomic unit. + ## Capabilities The subscription satisfies these optional `source` capability interfaces, diff --git a/source/kafka/adapter_internal_test.go b/source/kafka/adapter_internal_test.go index cff0227..2eaa4bd 100644 --- a/source/kafka/adapter_internal_test.go +++ b/source/kafka/adapter_internal_test.go @@ -747,3 +747,61 @@ func TestSubscribeBuildsClientAndAs(t *testing.T) { type stringCursor string func (c stringCursor) String() string { return string(c) } + +// TestSettleTermOnTransactionalSubscriptionRejected pins the P2-9 decision: on +// a transactional subscription a direct Term would produce the DLQ record +// outside the open EOS session, breaking atomicity with the offset mark, so it +// is rejected with ErrTermInsideTransaction and has no side effects. +func TestSettleTermOnTransactionalSubscriptionRejected(t *testing.T) { + t.Parallel() + + ft := &fakeTransactor{committed: true} + sub, fp := newTxSub(ft) + m := newMessage(rec("orders", 0, 9, "A-1", "poison")) + + err := sub.Settle(context.Background(), m, source.Term(errors.New("poison"))) + if !errors.Is(err, ErrTermInsideTransaction) { + t.Fatalf("Settle(Term) on tx subscription = %v, want ErrTermInsideTransaction", err) + } + if got := fp.markedCount(); got != 0 { + t.Errorf("marked = %d, want 0 (rejected settle must not advance offsets)", got) + } + if got := fp.producedCount(); got != 0 { + t.Errorf("produced = %d, want 0 (no DLQ write outside the transaction)", got) + } + if len(ft.calls) != 0 { + t.Errorf("transact calls = %v, want none", ft.calls) + } +} + +// TestBeginDeadLettersThroughTransactionOnPoison pins the supported EOS DLQ +// pattern for poison messages: inside Begin's fn, produce the rejected record +// to the dead-letter topic via the handed Tx and return nil, so the DLQ +// record and the consumed offset commit as one atomic unit. +func TestBeginDeadLettersThroughTransactionOnPoison(t *testing.T) { + t.Parallel() + + ft := &fakeTransactor{committed: true} + sub, fp := newTxSub(ft) + m := newMessage(rec("orders", 0, 11, "A-1", "poison")) + + err := sub.Begin(context.Background(), m, func(ctx context.Context, tx source.Tx) error { + return tx.Produce(ctx, source.ProducedRecord{ + Topic: "orders.DLQ", + Key: []byte("A-1"), + Value: []byte("poison"), + }) + }) + if err != nil { + t.Fatalf("Begin() error = %v, want nil", err) + } + if len(ft.produced) != 1 || len(ft.produced[0]) != 1 || ft.produced[0][0].Topic != "orders.DLQ" { + t.Fatalf("produced = %#v, want one record on orders.DLQ", ft.produced) + } + if got := fp.markedCount(); got != 1 { + t.Errorf("marked = %d, want 1 (consumed offset committed atomically with the DLQ record)", got) + } + if got, want := ft.calls, []string{"begin", "produce", "end-commit"}; !equalStrings(got, want) { + t.Errorf("call order = %v, want %v", got, want) + } +} diff --git a/source/kafka/capability.go b/source/kafka/capability.go index 374a344..19ad33a 100644 --- a/source/kafka/capability.go +++ b/source/kafka/capability.go @@ -432,6 +432,12 @@ func toRecordHeaders(hs source.Headers) []kgo.RecordHeader { // without a transaction. A rebalance during the transaction fences the producer, // so End reports the transaction did not commit and the work is retried after // reassignment. +// +// Poison handling on a transactional subscription flows through Begin too: +// settling [source.Term] directly is rejected with [ErrTermInsideTransaction]. +// Inside fn, produce the rejected record to the dead-letter topic via the +// handed [source.Tx] and return nil, so the DLQ write and the consumed offset +// commit atomically. func (s *subscription) Begin(ctx context.Context, m source.Message, fn func(ctx context.Context, tx source.Tx) error) error { if s.transactSess == nil { return fmt.Errorf("source/kafka: transactional: %w", errNotTransactional) diff --git a/source/kafka/go.mod b/source/kafka/go.mod index 143b35b..222186b 100644 --- a/source/kafka/go.mod +++ b/source/kafka/go.mod @@ -1,71 +1,20 @@ module github.com/stablekernel/crucible/source/kafka go 1.25.11 + toolchain go1.26.4 + require github.com/stablekernel/crucible/source v0.0.0 require ( - github.com/testcontainers/testcontainers-go v0.42.0 - github.com/testcontainers/testcontainers-go/modules/redpanda v0.42.0 github.com/twmb/franz-go v1.21.2 - github.com/twmb/franz-go/pkg/kadm v1.11.0 github.com/twmb/franz-go/pkg/kmsg v1.13.1 ) require ( - dario.cat/mergo v1.0.2 // indirect - github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c // indirect - github.com/Microsoft/go-winio v0.6.2 // indirect - github.com/cenkalti/backoff/v4 v4.3.0 // indirect - github.com/cespare/xxhash/v2 v2.3.0 // indirect - github.com/containerd/errdefs v1.0.0 // indirect - github.com/containerd/errdefs/pkg v0.3.0 // indirect - github.com/containerd/log v0.1.0 // indirect - github.com/containerd/platforms v0.2.1 // indirect - github.com/cpuguy83/dockercfg v0.3.2 // indirect - github.com/davecgh/go-spew v1.1.1 // indirect - github.com/distribution/reference v0.6.0 // indirect - github.com/docker/go-connections v0.6.0 // indirect - github.com/docker/go-units v0.5.0 // indirect - github.com/ebitengine/purego v0.10.0 // indirect - github.com/felixge/httpsnoop v1.0.4 // indirect - github.com/go-logr/logr v1.4.3 // indirect - github.com/go-logr/stdr v1.2.2 // indirect - github.com/go-ole/go-ole v1.2.6 // indirect - github.com/google/uuid v1.6.0 // indirect github.com/klauspost/compress v1.18.6 // indirect - github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect - github.com/magiconair/properties v1.8.10 // indirect - github.com/moby/docker-image-spec v1.3.1 // indirect - github.com/moby/go-archive v0.2.0 // indirect - github.com/moby/moby/api v1.54.1 // indirect - github.com/moby/moby/client v0.4.0 // indirect - github.com/moby/patternmatcher v0.6.1 // indirect - github.com/moby/sys/sequential v0.6.0 // indirect - github.com/moby/sys/user v0.4.0 // indirect - github.com/moby/sys/userns v0.1.0 // indirect - github.com/moby/term v0.5.2 // indirect - github.com/opencontainers/go-digest v1.0.0 // indirect - github.com/opencontainers/image-spec v1.1.1 // indirect github.com/pierrec/lz4/v4 v4.1.26 // indirect - github.com/pmezard/go-difflib v1.0.0 // indirect - github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect - github.com/shirou/gopsutil/v4 v4.26.3 // indirect - github.com/sirupsen/logrus v1.9.4 // indirect github.com/stablekernel/crucible/telemetry v0.0.0 // indirect - github.com/stretchr/testify v1.11.1 // indirect - github.com/tklauser/go-sysconf v0.3.16 // indirect - github.com/tklauser/numcpus v0.11.0 // indirect - github.com/yusufpapurcu/wmi v1.2.4 // indirect - go.opentelemetry.io/auto/sdk v1.2.1 // indirect - go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 // indirect - go.opentelemetry.io/otel v1.41.0 // indirect - go.opentelemetry.io/otel/metric v1.41.0 // indirect - go.opentelemetry.io/otel/trace v1.41.0 // indirect - golang.org/x/crypto v0.51.0 // indirect - golang.org/x/mod v0.33.0 // indirect - golang.org/x/sys v0.44.0 // indirect - gopkg.in/yaml.v3 v3.0.1 // indirect ) replace github.com/stablekernel/crucible/source => ../ diff --git a/source/kafka/go.sum b/source/kafka/go.sum index c32579a..b5d6d34 100644 --- a/source/kafka/go.sum +++ b/source/kafka/go.sum @@ -1,149 +1,10 @@ -dario.cat/mergo v1.0.2 h1:85+piFYR1tMbRrLcDwR18y4UKJ3aH1Tbzi24VRW1TK8= -dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA= -github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6 h1:He8afgbRMd7mFxO99hRNu+6tazq8nFF9lIwo9JFroBk= -github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8= -github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c h1:udKWzYgxTojEKWjV8V+WSxDXJ4NFATAsZjh8iIbsQIg= -github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= -github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= -github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= -github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= -github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= -github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= -github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= -github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG8PI= -github.com/containerd/errdefs v1.0.0/go.mod h1:+YBYIdtsnF4Iw6nWZhJcqGSg/dwvV7tyJ/kCkyJ2k+M= -github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151Xdx3ZPPE= -github.com/containerd/errdefs/pkg v0.3.0/go.mod h1:NJw6s9HwNuRhnjJhM7pylWwMyAkmCQvQ4GpJHEqRLVk= -github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I= -github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3EhrzVo= -github.com/containerd/platforms v0.2.1 h1:zvwtM3rz2YHPQsF2CHYM8+KtB5dvhISiXh5ZpSBQv6A= -github.com/containerd/platforms v0.2.1/go.mod h1:XHCb+2/hzowdiut9rkudds9bE5yJ7npe7dG/wG+uFPw= -github.com/cpuguy83/dockercfg v0.3.2 h1:DlJTyZGBDlXqUZ2Dk2Q3xHs/FtnooJJVaad2S9GKorA= -github.com/cpuguy83/dockercfg v0.3.2/go.mod h1:sugsbF4//dDlL/i+S+rtpIWp+5h0BHJHfjj5/jFyUJc= -github.com/creack/pty v1.1.24 h1:bJrF4RRfyJnbTJqzRLHzcGaZK1NeM5kTC9jGgovnR1s= -github.com/creack/pty v1.1.24/go.mod h1:08sCNb52WyoAwi2QDyzUCTgcvVFhUzewun7wtTfvcwE= -github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= -github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk= -github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E= -github.com/docker/go-connections v0.6.0 h1:LlMG9azAe1TqfR7sO+NJttz1gy6KO7VJBh+pMmjSD94= -github.com/docker/go-connections v0.6.0/go.mod h1:AahvXYshr6JgfUJGdDCs2b5EZG/vmaMAntpSFH5BFKE= -github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4= -github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= -github.com/ebitengine/purego v0.10.0 h1:QIw4xfpWT6GWTzaW5XEKy3HXoqrJGx1ijYHzTF0/ISU= -github.com/ebitengine/purego v0.10.0/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ= -github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= -github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= -github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= -github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= -github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= -github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= -github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= -github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY= -github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0= -github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= -github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= -github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= -github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= -github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao= github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= -github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= -github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= -github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= -github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= -github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ81pIr0yLvtUWk2if982qA3F3QD6H4= -github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I= -github.com/magiconair/properties v1.8.10 h1:s31yESBquKXCV9a/ScB3ESkOjUYYv+X0rg8SYxI99mE= -github.com/magiconair/properties v1.8.10/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0= -github.com/mdelapenya/tlscert v0.2.0 h1:7H81W6Z/4weDvZBNOfQte5GpIMo0lGYEeWbkGp5LJHI= -github.com/mdelapenya/tlscert v0.2.0/go.mod h1:O4njj3ELLnJjGdkN7M/vIVCpZ+Cf0L6muqOG4tLSl8o= -github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= -github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo= -github.com/moby/go-archive v0.2.0 h1:zg5QDUM2mi0JIM9fdQZWC7U8+2ZfixfTYoHL7rWUcP8= -github.com/moby/go-archive v0.2.0/go.mod h1:mNeivT14o8xU+5q1YnNrkQVpK+dnNe/K6fHqnTg4qPU= -github.com/moby/moby/api v1.54.1 h1:TqVzuJkOLsgLDDwNLmYqACUuTehOHRGKiPhvH8V3Nn4= -github.com/moby/moby/api v1.54.1/go.mod h1:+RQ6wluLwtYaTd1WnPLykIDPekkuyD/ROWQClE83pzs= -github.com/moby/moby/client v0.4.0 h1:S+2XegzHQrrvTCvF6s5HFzcrywWQmuVnhOXe2kiWjIw= -github.com/moby/moby/client v0.4.0/go.mod h1:QWPbvWchQbxBNdaLSpoKpCdf5E+WxFAgNHogCWDoa7g= -github.com/moby/patternmatcher v0.6.1 h1:qlhtafmr6kgMIJjKJMDmMWq7WLkKIo23hsrpR3x084U= -github.com/moby/patternmatcher v0.6.1/go.mod h1:hDPoyOpDY7OrrMDLaYoY3hf52gNCR/YOUYxkhApJIxc= -github.com/moby/sys/sequential v0.6.0 h1:qrx7XFUd/5DxtqcoH1h438hF5TmOvzC/lspjy7zgvCU= -github.com/moby/sys/sequential v0.6.0/go.mod h1:uyv8EUTrca5PnDsdMGXhZe6CCe8U/UiTWd+lL+7b/Ko= -github.com/moby/sys/user v0.4.0 h1:jhcMKit7SA80hivmFJcbB1vqmw//wU61Zdui2eQXuMs= -github.com/moby/sys/user v0.4.0/go.mod h1:bG+tYYYJgaMtRKgEmuueC0hJEAZWwtIbZTB+85uoHjs= -github.com/moby/sys/userns v0.1.0 h1:tVLXkFOxVu9A64/yh59slHVv9ahO9UIev4JZusOLG/g= -github.com/moby/sys/userns v0.1.0/go.mod h1:IHUYgu/kao6N8YZlp9Cf444ySSvCmDlmzUcYfDHOl28= -github.com/moby/term v0.5.2 h1:6qk3FJAFDs6i/q3W/pQ97SX192qKfZgGjCQqfCJkgzQ= -github.com/moby/term v0.5.2/go.mod h1:d3djjFCrjnB+fl8NJux+EJzu0msscUP+f8it8hPkFLc= -github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= -github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= -github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040= -github.com/opencontainers/image-spec v1.1.1/go.mod h1:qpqAh3Dmcf36wStyyWU+kCeDgrGnAve2nCC8+7h8Q0M= github.com/pierrec/lz4/v4 v4.1.26 h1:GrpZw1gZttORinvzBdXPUXATeqlJjqUG/D87TKMnhjY= github.com/pierrec/lz4/v4 v4.1.26/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= -github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= -github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 h1:o4JXh1EVt9k/+g42oCprj/FisM4qX9L3sZB3upGN2ZU= -github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= -github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= -github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= -github.com/shirou/gopsutil/v4 v4.26.3 h1:2ESdQt90yU3oXF/CdOlRCJxrP+Am1aBYubTMTfxJ1qc= -github.com/shirou/gopsutil/v4 v4.26.3/go.mod h1:LZ6ewCSkBqUpvSOf+LsTGnRinC6iaNUNMGBtDkJBaLQ= -github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w= -github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g= -github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4= -github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0= -github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= -github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= -github.com/testcontainers/testcontainers-go v0.42.0 h1:He3IhTzTZOygSXLJPMX7n44XtK+qhjat1nI9cneBbUY= -github.com/testcontainers/testcontainers-go v0.42.0/go.mod h1:vZjdY1YmUA1qEForxOIOazfsrdyORJAbhi0bp8plN30= -github.com/testcontainers/testcontainers-go/modules/redpanda v0.42.0 h1:mcPLvf3rzvvwG46i4jcyOk7/KcapE/IKnBgwo27k68M= -github.com/testcontainers/testcontainers-go/modules/redpanda v0.42.0/go.mod h1:Yq0WrUIIsMkJoZ2DadCJ1Zq4RFhjPPpSNYHpjRrGfDU= -github.com/tklauser/go-sysconf v0.3.16 h1:frioLaCQSsF5Cy1jgRBrzr6t502KIIwQ0MArYICU0nA= -github.com/tklauser/go-sysconf v0.3.16/go.mod h1:/qNL9xxDhc7tx3HSRsLWNnuzbVfh3e7gh/BmM179nYI= -github.com/tklauser/numcpus v0.11.0 h1:nSTwhKH5e1dMNsCdVBukSZrURJRoHbSEQjdEbY+9RXw= -github.com/tklauser/numcpus v0.11.0/go.mod h1:z+LwcLq54uWZTX0u/bGobaV34u6V7KNlTZejzM6/3MQ= github.com/twmb/franz-go v1.21.2 h1:WrvV/spF48JzcRylqDQy02Vm6V6W4lhtD9Y4BOYNMu4= github.com/twmb/franz-go v1.21.2/go.mod h1:rfoMTnVk7107fhTGxfEKIHP/e7tPe6oyij/ywzO0czk= -github.com/twmb/franz-go/pkg/kadm v1.11.0 h1:FfeWJ0qadntFpAcQt8JzNXW4dijjytZNLrzJuzzzuxA= -github.com/twmb/franz-go/pkg/kadm v1.11.0/go.mod h1:qrhkdH+SWS3ivmbqOgHbpgVHamhaKcjH0UM+uOp0M1A= github.com/twmb/franz-go/pkg/kmsg v1.13.1 h1:fG5kItwysTk5UXqVwb64EpQEy3TydF3vYYK21nUQ+bI= github.com/twmb/franz-go/pkg/kmsg v1.13.1/go.mod h1:+DPt4NC8RmI6hqb8G09+3giKObE6uD2Eya6CfqBpeJY= -github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0= -github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= -go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= -go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= -go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 h1:sbiXRNDSWJOTobXh5HyQKjq6wUC5tNybqjIqDpAY4CU= -go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0/go.mod h1:69uWxva0WgAA/4bu2Yy70SLDBwZXuQ6PbBpbsa5iZrQ= -go.opentelemetry.io/otel v1.41.0 h1:YlEwVsGAlCvczDILpUXpIpPSL/VPugt7zHThEMLce1c= -go.opentelemetry.io/otel v1.41.0/go.mod h1:Yt4UwgEKeT05QbLwbyHXEwhnjxNO6D8L5PQP51/46dE= -go.opentelemetry.io/otel/metric v1.41.0 h1:rFnDcs4gRzBcsO9tS8LCpgR0dxg4aaxWlJxCno7JlTQ= -go.opentelemetry.io/otel/metric v1.41.0/go.mod h1:xPvCwd9pU0VN8tPZYzDZV/BMj9CM9vs00GuBjeKhJps= -go.opentelemetry.io/otel/sdk v1.35.0 h1:iPctf8iprVySXSKJffSS79eOjl9pvxV9ZqOWT0QejKY= -go.opentelemetry.io/otel/sdk v1.35.0/go.mod h1:+ga1bZliga3DxJ3CQGg3updiaAJoNECOgJREo9KHGQg= -go.opentelemetry.io/otel/sdk/metric v1.35.0 h1:1RriWBmCKgkeHEhM7a2uMjMUfP7MsOF5JpUCaEqEI9o= -go.opentelemetry.io/otel/sdk/metric v1.35.0/go.mod h1:is6XYCUMpcKi+ZsOvfluY5YstFnhW0BidkR+gL+qN+w= -go.opentelemetry.io/otel/trace v1.41.0 h1:Vbk2co6bhj8L59ZJ6/xFTskY+tGAbOnCtQGVVa9TIN0= -go.opentelemetry.io/otel/trace v1.41.0/go.mod h1:U1NU4ULCoxeDKc09yCWdWe+3QoyweJcISEVa1RBzOis= golang.org/x/crypto v0.51.0 h1:IBPXwPfKxY7cWQZ38ZCIRPI50YLeevDLlLnyC5wRGTI= golang.org/x/crypto v0.51.0/go.mod h1:8AdwkbraGNABw2kOX6YFPs3WM22XqI4EXEd8g+x7Oc8= -golang.org/x/mod v0.33.0 h1:tHFzIWbBifEmbwtGz65eaWyGiGZatSrT9prnU8DbVL8= -golang.org/x/mod v0.33.0/go.mod h1:swjeQEj+6r7fODbD2cqrnje9PnziFuw4bmLbBZFrQ5w= -golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.44.0 h1:ildZl3J4uzeKP07r2F++Op7E9B29JRUy+a27EibtBTQ= -golang.org/x/sys v0.44.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= -golang.org/x/term v0.43.0 h1:S4RLU2sB31O/NCl+zFN9Aru9A/Cq2aqKpTZJ6B+DwT4= -golang.org/x/term v0.43.0/go.mod h1:lrhlHNdQJHO+1qVYiHfFKVuVioJIheAc3fBSMFYEIsk= -golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= -gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= -gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= -gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= -gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= -gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= -gotest.tools/v3 v3.5.2 h1:7koQfIKdy+I8UTetycgUqXWSDwpgv193Ka+qRsmBY8Q= -gotest.tools/v3 v3.5.2/go.mod h1:LtdLGcnqToBH83WByAAi/wiwSFCArdFIUV/xxN4pcjA= -pgregory.net/rapid v1.2.0 h1:keKAYRcjm+e1F0oAuU5F5+YPAWcyxNNRK2wud503Gnk= -pgregory.net/rapid v1.2.0/go.mod h1:PY5XlDGj0+V1FCq0o192FdRhpKHGTRIWBgqjDBTrq04= diff --git a/source/kafka/integration/integration_test.go b/source/kafka/integration/integration_test.go new file mode 100644 index 0000000..d5b9c9b --- /dev/null +++ b/source/kafka/integration/integration_test.go @@ -0,0 +1,535 @@ +// SPDX-License-Identifier: Apache-2.0 + +//go:build integration + +package integration + +import ( + "context" + "errors" + "fmt" + "sync" + "testing" + "time" + + "github.com/testcontainers/testcontainers-go" + "github.com/testcontainers/testcontainers-go/modules/redpanda" + "github.com/twmb/franz-go/pkg/kadm" + "github.com/twmb/franz-go/pkg/kerr" + "github.com/twmb/franz-go/pkg/kgo" + + "github.com/stablekernel/crucible/source" + kafkasource "github.com/stablekernel/crucible/source/kafka" +) + +const redpandaImage = "docker.redpanda.com/redpandadata/redpanda:v23.3.3" + +// TestIntegrationConsumeAckTermRoundTrip starts a real RedPanda broker, produces +// records to a topic, consumes them through the Inlet, settles one Ack and one +// Term, and proves the committed offset advanced and the termed record landed on +// the dead-letter topic. It skips cleanly when Docker is unreachable. +func TestIntegrationConsumeAckTermRoundTrip(t *testing.T) { + skipWithoutDocker(t) + + ctx := context.Background() + container, err := redpanda.Run(ctx, redpandaImage) + if err != nil { + t.Skipf("redpanda.Run unavailable (image pull or startup failed); skipping: %v", err) + } + t.Cleanup(func() { _ = testcontainers.TerminateContainer(container) }) + + broker, err := container.KafkaSeedBroker(ctx) + if err != nil { + t.Fatalf("KafkaSeedBroker() error = %v", err) + } + + const ( + topic = "orders" + dlq = "orders.DLQ" + group = "orders-consumer" + ) + + // Create the source and dead-letter topics up front so neither the producer + // below nor the Inlet's internal DLQ producer races topic auto-creation. The + // Inlet's DLQ client does not enable AllowAutoTopicCreation, so the DLQ topic + // must exist before the first Term settles. + createTopics(ctx, t, broker, topic, dlq) + + // Produce two records with a separate client. + prod, err := kgo.NewClient( + kgo.SeedBrokers(broker), + kgo.AllowAutoTopicCreation(), + ) + if err != nil { + t.Fatalf("producer client error = %v", err) + } + t.Cleanup(prod.Close) + + produce(ctx, t, prod, topic, "A-1", "good") + produce(ctx, t, prod, topic, "A-2", "poison") + + // Consume through the Inlet. + inlet, err := kafkasource.New( + kafkasource.WithSeedBrokers(broker), + kafkasource.WithClientID("it-consumer"), + kafkasource.WithDLQTopic(dlq), + kafkasource.WithClientOptions(kgo.ConsumeResetOffset(kgo.NewOffset().AtStart())), + ) + if err != nil { + t.Fatalf("New() error = %v", err) + } + t.Cleanup(func() { _ = inlet.Close() }) + + sub, err := inlet.Subscribe(ctx, source.SubscribeConfig{Topics: []string{topic}, Group: group}) + if err != nil { + t.Fatalf("Subscribe() error = %v", err) + } + + // Pull both records and settle: A-1 ack, A-2 term (dead-letter). + pollCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + + seen := map[string]bool{} + for len(seen) < 2 { + m, nerr := sub.Next(pollCtx) + if nerr != nil { + t.Fatalf("Next() error = %v (saw %v)", nerr, seen) + } + key := string(m.Key()) + seen[key] = true + switch key { + case "A-1": + if serr := sub.Settle(pollCtx, m, source.Ack()); serr != nil { + t.Fatalf("Settle(ack) error = %v", serr) + } + case "A-2": + if serr := sub.Settle(pollCtx, m, source.Term(errors.New("poison payload"))); serr != nil { + t.Fatalf("Settle(term) error = %v", serr) + } + default: + t.Fatalf("unexpected key %q", key) + } + } + + // Close commits marked offsets. + if cerr := sub.Close(); cerr != nil { + t.Fatalf("Close() error = %v", cerr) + } + + // Prove the termed record landed on the dead-letter topic. + dlqClient, err := kgo.NewClient( + kgo.SeedBrokers(broker), + kgo.ConsumeTopics(dlq), + kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()), + ) + if err != nil { + t.Fatalf("dlq client error = %v", err) + } + t.Cleanup(dlqClient.Close) + + dlqCtx, dlqCancel := context.WithTimeout(ctx, 30*time.Second) + defer dlqCancel() + fetches := dlqClient.PollFetches(dlqCtx) + if errs := fetches.Errors(); len(errs) > 0 { + t.Fatalf("dlq PollFetches errors = %v", errs) + } + recs := fetches.Records() + if len(recs) != 1 || string(recs[0].Key) != "A-2" { + t.Fatalf("dlq records = %#v, want one A-2", recs) + } + if !hasHeader(recs[0].Headers, "crucible-source-topic", "orders") { + t.Errorf("dlq record missing crucible-source-topic=orders header: %+v", recs[0].Headers) + } + if !hasHeader(recs[0].Headers, "crucible-class", "poison") { + t.Errorf("dlq record missing crucible-class=poison header: %+v", recs[0].Headers) + } +} + +// TestIntegrationTransactionalEOSRoundTrip starts a real RedPanda broker and +// proves exactly-once consume-process-produce: it produces an input record, +// consumes it through a transactional Inlet, and runs two transactions for it. +// The first transaction produces an output record but is aborted (its work +// function fails), so neither the output record nor the input offset is +// committed; a read-committed reader sees no output, and the input is +// redelivered. The second transaction produces the same output and commits, so +// exactly one output record exists and the offset advances. The aborted attempt +// leaves no duplicate. +func TestIntegrationTransactionalEOSRoundTrip(t *testing.T) { + skipWithoutDocker(t) + + ctx := context.Background() + container, err := redpanda.Run(ctx, redpandaImage) + if err != nil { + t.Skipf("redpanda.Run unavailable (image pull or startup failed); skipping: %v", err) + } + t.Cleanup(func() { _ = testcontainers.TerminateContainer(container) }) + + broker, err := container.KafkaSeedBroker(ctx) + if err != nil { + t.Fatalf("KafkaSeedBroker() error = %v", err) + } + + const ( + inTopic = "eos.in" + outTopic = "eos.out" + group = "eos-consumer" + txID = "eos-it-v1" + ) + createTopics(ctx, t, broker, inTopic, outTopic) + + prod, err := kgo.NewClient(kgo.SeedBrokers(broker)) + if err != nil { + t.Fatalf("producer client error = %v", err) + } + t.Cleanup(prod.Close) + produce(ctx, t, prod, inTopic, "K-1", "input") + + inlet, err := kafkasource.New( + kafkasource.WithSeedBrokers(broker), + kafkasource.WithClientID("eos-consumer"), + kafkasource.WithTransactional(txID), + kafkasource.WithClientOptions(kgo.ConsumeResetOffset(kgo.NewOffset().AtStart())), + ) + if err != nil { + t.Fatalf("New() error = %v", err) + } + t.Cleanup(func() { _ = inlet.Close() }) + + sub, err := inlet.Subscribe(ctx, source.SubscribeConfig{Topics: []string{inTopic}, Group: group}) + if err != nil { + t.Fatalf("Subscribe() error = %v", err) + } + tx, ok := sub.(source.Transactional) + if !ok { + t.Fatal("subscription does not satisfy source.Transactional with WithTransactional") + } + + pollCtx, cancel := context.WithTimeout(ctx, 60*time.Second) + defer cancel() + + // First receive: open a transaction that produces the output, then fail the + // work function so the transaction aborts. Nothing should be committed. + m1, err := sub.Next(pollCtx) + if err != nil { + t.Fatalf("Next() #1 error = %v", err) + } + if string(m1.Key()) != "K-1" { + t.Fatalf("first record key = %q, want K-1", m1.Key()) + } + abortErr := errors.New("deliberate abort") + beginErr := tx.Begin(pollCtx, m1, func(c context.Context, txn source.Tx) error { + if perr := txn.Produce(c, source.ProducedRecord{Topic: outTopic, Key: []byte("K-1"), Value: []byte("output")}); perr != nil { + return perr + } + return abortErr // abort: discard the produced record, do not commit the offset + }) + if !errors.Is(beginErr, abortErr) { + t.Fatalf("Begin() #1 = %v, want the abort error", beginErr) + } + + // The input must be redelivered because its offset was not committed. Re-seek + // the subscription to start so the redelivery is deterministic even though the + // aborted transaction left the consumer position past the record. + if sk, ok := sub.(source.Seekable); ok { + if serr := sk.SeekToStart(pollCtx); serr != nil { + t.Fatalf("SeekToStart() error = %v", serr) + } + } + + m2, err := sub.Next(pollCtx) + if err != nil { + t.Fatalf("Next() #2 (redelivery) error = %v", err) + } + if string(m2.Key()) != "K-1" { + t.Fatalf("redelivered key = %q, want K-1 (aborted offset not committed)", m2.Key()) + } + + // Second transaction: produce the same output and commit. This is the only + // transaction that lands an output record and advances the offset. + commitErr := tx.Begin(pollCtx, m2, func(c context.Context, txn source.Tx) error { + return txn.Produce(c, source.ProducedRecord{Topic: outTopic, Key: []byte("K-1"), Value: []byte("output")}) + }) + if commitErr != nil { + t.Fatalf("Begin() #2 (commit) error = %v", commitErr) + } + + if cerr := sub.Close(); cerr != nil { + t.Fatalf("Close() error = %v", cerr) + } + + // A read-committed reader must see exactly one output record: the aborted + // transaction's produce is invisible, so there is no duplicate. + outClient, err := kgo.NewClient( + kgo.SeedBrokers(broker), + kgo.ConsumeTopics(outTopic), + kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()), + kgo.FetchIsolationLevel(kgo.ReadCommitted()), + ) + if err != nil { + t.Fatalf("out client error = %v", err) + } + t.Cleanup(outClient.Close) + + var got []*kgo.Record + readCtx, readCancel := context.WithTimeout(ctx, 20*time.Second) + defer readCancel() + for len(got) < 1 { + f := outClient.PollFetches(readCtx) + if errs := f.Errors(); len(errs) > 0 { + for _, fe := range errs { + if fe.Err != nil && !errors.Is(fe.Err, context.DeadlineExceeded) { + t.Fatalf("out PollFetches error = %v", fe.Err) + } + } + break + } + got = append(got, f.Records()...) + } + if len(got) != 1 { + t.Fatalf("committed output records = %d, want exactly 1 (no duplicate across the aborted transaction)", len(got)) + } + if string(got[0].Key) != "K-1" || string(got[0].Value) != "output" { + t.Errorf("output record = %s/%s, want K-1/output", got[0].Key, got[0].Value) + } +} + +// createTopics creates the given topics (one partition, replication factor one) +// against the broker and waits for the admin call to succeed, so produces never +// race auto-creation. An already-exists result is treated as success. +func createTopics(ctx context.Context, t *testing.T, broker string, topics ...string) { + t.Helper() + admClient, err := kgo.NewClient(kgo.SeedBrokers(broker)) + if err != nil { + t.Fatalf("admin client error = %v", err) + } + defer admClient.Close() + + adm := kadm.NewClient(admClient) + createCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + + resp, err := adm.CreateTopics(createCtx, 1, 1, nil, topics...) + if err != nil { + t.Fatalf("CreateTopics(%v) error = %v", topics, err) + } + for _, ct := range resp { + if ct.Err != nil && !errors.Is(ct.Err, kerr.TopicAlreadyExists) { + t.Fatalf("CreateTopics(%q) error = %v", ct.Topic, ct.Err) + } + } +} + +func produce(ctx context.Context, t *testing.T, c *kgo.Client, topic, key, value string) { + t.Helper() + r := &kgo.Record{Topic: topic, Key: []byte(key), Value: []byte(value)} + if err := c.ProduceSync(ctx, r).FirstErr(); err != nil { + t.Fatalf("produce %s error = %v", key, err) + } +} + +func hasHeader(hs []kgo.RecordHeader, key, value string) bool { + for _, h := range hs { + if h.Key == key && string(h.Value) == value { + return true + } + } + return false +} + +func skipWithoutDocker(t *testing.T) { + t.Helper() + provider, err := testcontainers.NewDockerProvider() + if err != nil { + t.Skipf("docker unavailable: %v", err) + } + defer func() { _ = provider.Close() }() + if err := provider.Health(context.Background()); err != nil { + t.Skipf("docker unavailable: %v", err) + } +} + +// deliveryLog records every (topic, partition, offset) a group member +// delivered, so the rebalance test can assert the union across members covers +// every produced record exactly once. +type deliveryLog struct { + mu sync.Mutex + counts map[string]int +} + +func newDeliveryLog() *deliveryLog { return &deliveryLog{counts: map[string]int{}} } + +func (d *deliveryLog) record(cursor string) { + d.mu.Lock() + defer d.mu.Unlock() + d.counts[cursor]++ +} + +func (d *deliveryLog) snapshot() map[string]int { + d.mu.Lock() + defer d.mu.Unlock() + out := make(map[string]int, len(d.counts)) + for k, v := range d.counts { + out[k] = v + } + return out +} + +// TestIntegrationRebalanceRevokeNoDuplicates drives two members of one consumer +// group through a membership change: B joins mid-stream, then leaves gracefully, +// forcing two rebalances while A keeps consuming. The adapter's revoke hook +// commits marked offsets before releasing partitions, so every record must be +// delivered exactly once across the whole group despite the churn. +func TestIntegrationRebalanceRevokeNoDuplicates(t *testing.T) { + skipWithoutDocker(t) + + ctx := context.Background() + container, err := redpanda.Run(ctx, redpandaImage) + if err != nil { + t.Skipf("redpanda.Run unavailable (image pull or startup failed); skipping: %v", err) + } + t.Cleanup(func() { _ = testcontainers.TerminateContainer(container) }) + + broker, err := container.KafkaSeedBroker(ctx) + if err != nil { + t.Fatalf("KafkaSeedBroker() error = %v", err) + } + + const ( + topic = "rebalance" + group = "rebalance-it" + records = 40 + warmup = 8 // records A consumes before B joins + runFor = 8 * time.Second + tailFor = 3 * time.Second + pollStep = 500 * time.Millisecond + ) + + // Two partitions so the group actually splits work between members. + admClient, err := kgo.NewClient(kgo.SeedBrokers(broker), kgo.AllowAutoTopicCreation()) + if err != nil { + t.Fatalf("admin client error = %v", err) + } + adm := kadm.NewClient(admClient) + if _, err = adm.CreateTopics(ctx, 2, 1, nil, topic); err != nil { + admClient.Close() + t.Fatalf("CreateTopics(%s) error = %v", topic, err) + } + admClient.Close() + + prod, err := kgo.NewClient(kgo.SeedBrokers(broker), kgo.AllowAutoTopicCreation()) + if err != nil { + t.Fatalf("producer client error = %v", err) + } + t.Cleanup(prod.Close) + for i := range records { + produce(ctx, t, prod, topic, fmt.Sprintf("k%02d", i), fmt.Sprintf("v%02d", i)) + } + + log := newDeliveryLog() + + // drive consumes from sub until stop closes, acking and logging every + // delivery keyed by its cursor ("topic/partition@offset"). Transient + // rebalance-window errors are retried, not fatal. + drive := func(sub source.Subscription, stop <-chan struct{}, wg *sync.WaitGroup) { + defer wg.Done() + for { + select { + case <-stop: + return + default: + } + pollCtx, cancel := context.WithTimeout(context.Background(), pollStep) + m, err := sub.Next(pollCtx) + cancel() + if err != nil { + continue // quiet broker, rebalance in progress, or step deadline + } + settleCtx, scancel := context.WithTimeout(context.Background(), 5*time.Second) + err = sub.Settle(settleCtx, m, source.Ack()) + scancel() + if err == nil { + log.record(m.Cursor().String()) + } + } + } + _ = drive + + subA, inletA := mustSubscribe(t, broker, group, topic) + defer func() { _ = inletA.Close() }() + + stopA := make(chan struct{}) + var wgA sync.WaitGroup + wgA.Add(1) + go drive(subA, stopA, &wgA) + + // Warm up: let A deliver at least warmup records so its marks are pending + // when the first rebalance hits. + deadline := time.Now().Add(20 * time.Second) + for len(log.snapshot()) < warmup && time.Now().Before(deadline) { + time.Sleep(50 * time.Millisecond) + } + if len(log.snapshot()) < warmup { + t.Fatalf("consumer A delivered %d records in 20s, want >= %d", len(log.snapshot()), warmup) + } + + // B joins: forces a rebalance that moves partitions while A has settled- + // but-maybe-uncommitted work, exercising the revoke-commit path. + subB, inletB := mustSubscribe(t, broker, group, topic) + stopB := make(chan struct{}) + var wgB sync.WaitGroup + wgB.Add(1) + go drive(subB, stopB, &wgB) + + time.Sleep(runFor) + close(stopB) + wgB.Wait() + // Graceful leave: Close commits B's marks before releasing its partitions. + if err := subB.Close(); err != nil { + t.Fatalf("member B Close() error = %v", err) + } + _ = inletB.Close() + time.Sleep(tailFor) // second rebalance folds B's partitions back into A + close(stopA) + wgA.Wait() + if err := subA.Close(); err != nil { + t.Fatalf("member A Close() error = %v", err) + } + + got := log.snapshot() + if len(got) != records { + t.Fatalf("delivered %d distinct records, want %d", len(got), records) + } + var dupes []string + for k, n := range got { + if n != 1 { + dupes = append(dupes, fmt.Sprintf("%s×%d", k, n)) + } + } + if len(dupes) > 0 { + t.Fatalf("rebalance produced duplicate deliveries: %v", dupes) + } +} + +// mustSubscribe opens an Inlet and a group subscription against broker, +// failing the test on error. +func mustSubscribe(t *testing.T, broker, group, topic string) (source.Subscription, *kafkasource.Inlet) { + t.Helper() + inlet, err := kafkasource.New( + kafkasource.WithSeedBrokers(broker), + kafkasource.WithClientID(group+"-"+fmt.Sprint(time.Now().UnixNano())), + kafkasource.WithClientOptions( + kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()), + ), + ) + if err != nil { + t.Fatalf("New() error = %v", err) + } + sub, err := inlet.Subscribe(context.Background(), source.SubscribeConfig{ + Topics: []string{topic}, + Group: group, + }) + if err != nil { + t.Fatalf("Subscribe() error = %v", err) + } + return sub, inlet +} diff --git a/source/kafka/kafka.go b/source/kafka/kafka.go index 6361fac..e6344c2 100644 --- a/source/kafka/kafka.go +++ b/source/kafka/kafka.go @@ -23,7 +23,10 @@ // offset can pass the nacked record — a documented best-effort divergence, // not an in-session one. // - Term produces the record to the configured dead-letter topic, then marks -// it for commit so it is not re-read. +// it for commit so it is not re-read — except on a transactional +// subscription, where Term reports [ErrTermInsideTransaction]: route poison +// through Begin's transaction instead, producing to the dead-letter topic +// via the handed [source.Tx] so DLQ write and offset commit are atomic. // - InProgress is a no-op: Kafka has no per-message ack deadline to extend. // - Manual is a no-op: the handler settled the record itself through // [source.Message.As] and the underlying *kgo.Client. @@ -90,6 +93,17 @@ var ErrNoDLQTopic = errors.New("source/kafka: term requested but no dead-letter // with errors.Is. var ErrNoCommittedOffsets = errors.New("source/kafka: lag requires at least one committed offset") +// ErrTermInsideTransaction reports that a handler returned [source.Term] (or +// [source.Reject]) against a transactional subscription outside +// [source.Transactional.Begin]. A DLQ produce there would write outside the +// open EOS session and break atomicity with the consumed offset. On a +// transactional subscription, route poison through Begin instead: inside fn, +// produce the rejected record to the dead-letter topic via the handed +// [source.Tx] (a ProducedRecord with the DLQ topic), then return nil — the +// DLQ record and the input offset then commit as one atomic unit. Match it +// with errors.Is. +var ErrTermInsideTransaction = errors.New("source/kafka: dead-letter via term is not supported outside Begin on a transactional subscription") + // errTransactionalSingleSubscribe reports a second [Inlet.Subscribe] on a // transactional inlet. The exactly-once session backing a transactional inlet // fences a single consumer, so only one subscription per transactional inlet is diff --git a/source/kafka/subscription.go b/source/kafka/subscription.go index 030c84c..9fb4568 100644 --- a/source/kafka/subscription.go +++ b/source/kafka/subscription.go @@ -292,6 +292,15 @@ func (s *subscription) Settle(ctx context.Context, m source.Message, r source.Re return s.requeueWithDelay(ctx, rec, r.Requeue) case source.ActionTerm: + // On a transactional subscription the DLQ write must join the open + // EOS transaction; a direct Term here would produce outside it and + // break atomicity with the offset mark. Route poison through Begin's + // fn instead: produce to the dead-letter topic via the handed + // [source.Tx] and return success, committing DLQ record and consumed + // offset as one unit. + if s.transactSess != nil { + return fmt.Errorf("source/kafka: term: %w", ErrTermInsideTransaction) + } // Produce to the dead-letter topic, then mark so it is not re-read. if err := s.deadLetter(ctx, rec, r); err != nil { return err diff --git a/source/memsource/memsource_test.go b/source/memsource/memsource_test.go index 764f491..7f4f9e3 100644 --- a/source/memsource/memsource_test.go +++ b/source/memsource/memsource_test.go @@ -175,3 +175,45 @@ func memsourceLedgerWith(t *testing.T, results ...source.Result) *memsource.Ledg }) return h.Ledger() } + +// TestMemsourceDuplicateSettleIsIdempotent pins duplicate-settle semantics for +// the in-memory adapter: a second Settle for an already-settled message is +// safe — it records another ledger entry (every settle decision is logged) but +// never corrupts the in-flight accounting, so the subscription still drains. +func TestMemsourceDuplicateSettleIsIdempotent(t *testing.T) { + t.Parallel() + + in := memsource.New(memsource.WithMessages(memsource.Msg{Key: "k", Value: []byte("v")})) + sub, err := in.Subscribe(context.Background(), source.SubscribeConfig{}) + if err != nil { + t.Fatal(err) + } + m, err := sub.Next(context.Background()) + if err != nil { + t.Fatal(err) + } + for i := range 2 { + if err := sub.Settle(context.Background(), m, source.Ack()); err != nil { + t.Fatalf("Settle #%d error = %v, want nil", i+1, err) + } + } + if got := in.Ledger().Counts(); got.Acked != 2 { + t.Fatalf("ledger acks = %d, want 2 (each settle recorded, duplicates harmless)", got.Acked) + } + if err := sub.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + drained := make(chan error, 1) + go func() { + _, err := sub.Next(context.Background()) + drained <- err + }() + select { + case err := <-drained: + if !errors.Is(err, source.ErrDrained) { + t.Fatalf("Next after duplicate settles + Close = %v, want ErrDrained", err) + } + case <-time.After(2 * time.Second): + t.Fatal("subscription did not drain after duplicate settles (in-flight accounting corrupted)") + } +} From b2acb91b1ac5f9d188febd507454699ac23a1874 Mon Sep 17 00:00:00 2001 From: Joshua Temple Date: Sat, 22 Aug 2026 23:33:20 -0400 Subject: [PATCH 2/2] fix(source/kafka): commit the integration module's go.mod and go.sum The nested module's manifest was created after the checkpoint branch's initial commit and missed the git add, so CI's source/kafka/integration leg had no module definition. tidy run via GOFLAGS=-tags=integration so the build-tag-gated testcontainers/redpanda/kadm imports keep their requirements. --- source/kafka/integration/go.mod | 86 ++++++++++++++++++ source/kafka/integration/go.sum | 149 ++++++++++++++++++++++++++++++++ 2 files changed, 235 insertions(+) create mode 100644 source/kafka/integration/go.mod create mode 100644 source/kafka/integration/go.sum diff --git a/source/kafka/integration/go.mod b/source/kafka/integration/go.mod new file mode 100644 index 0000000..ede34d7 --- /dev/null +++ b/source/kafka/integration/go.mod @@ -0,0 +1,86 @@ +module github.com/stablekernel/crucible/source/kafka/integration + +// Test-only module: the RedPanda end-to-end leg lives here so the +// testcontainers-go, redpanda, and kadm dependencies never enter +// source/kafka's own go.mod (and therefore no downstream module graph). +// It is excluded from go.work, built with GOWORK=off, and run with +// `go test -tags integration ./...` from this directory (Docker required; +// tests skip cleanly when a daemon is unreachable). + +go 1.25.11 + +toolchain go1.26.4 + +require ( + github.com/stablekernel/crucible/source v0.0.0 + github.com/stablekernel/crucible/source/kafka v0.0.0-00010101000000-000000000000 + github.com/testcontainers/testcontainers-go v0.42.0 + github.com/testcontainers/testcontainers-go/modules/redpanda v0.42.0 + github.com/twmb/franz-go v1.21.2 + github.com/twmb/franz-go/pkg/kadm v1.11.0 +) + +require ( + dario.cat/mergo v1.0.2 // indirect + github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c // indirect + github.com/Microsoft/go-winio v0.6.2 // indirect + github.com/cenkalti/backoff/v4 v4.3.0 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/containerd/errdefs v1.0.0 // indirect + github.com/containerd/errdefs/pkg v0.3.0 // indirect + github.com/containerd/log v0.1.0 // indirect + github.com/containerd/platforms v0.2.1 // indirect + github.com/cpuguy83/dockercfg v0.3.2 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/distribution/reference v0.6.0 // indirect + github.com/docker/go-connections v0.6.0 // indirect + github.com/docker/go-units v0.5.0 // indirect + github.com/ebitengine/purego v0.10.0 // indirect + github.com/felixge/httpsnoop v1.0.4 // indirect + github.com/go-logr/logr v1.4.3 // indirect + github.com/go-logr/stdr v1.2.2 // indirect + github.com/go-ole/go-ole v1.2.6 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/klauspost/compress v1.18.6 // indirect + github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect + github.com/magiconair/properties v1.8.10 // indirect + github.com/moby/docker-image-spec v1.3.1 // indirect + github.com/moby/go-archive v0.2.0 // indirect + github.com/moby/moby/api v1.54.1 // indirect + github.com/moby/moby/client v0.4.0 // indirect + github.com/moby/patternmatcher v0.6.1 // indirect + github.com/moby/sys/sequential v0.6.0 // indirect + github.com/moby/sys/user v0.4.0 // indirect + github.com/moby/sys/userns v0.1.0 // indirect + github.com/moby/term v0.5.2 // indirect + github.com/opencontainers/go-digest v1.0.0 // indirect + github.com/opencontainers/image-spec v1.1.1 // indirect + github.com/pierrec/lz4/v4 v4.1.26 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect + github.com/shirou/gopsutil/v4 v4.26.3 // indirect + github.com/sirupsen/logrus v1.9.4 // indirect + github.com/stablekernel/crucible/telemetry v0.0.0 // indirect + github.com/stretchr/testify v1.11.1 // indirect + github.com/tklauser/go-sysconf v0.3.16 // indirect + github.com/tklauser/numcpus v0.11.0 // indirect + github.com/twmb/franz-go/pkg/kmsg v1.13.1 // indirect + github.com/yusufpapurcu/wmi v1.2.4 // indirect + go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 // indirect + go.opentelemetry.io/otel v1.41.0 // indirect + go.opentelemetry.io/otel/metric v1.41.0 // indirect + go.opentelemetry.io/otel/trace v1.41.0 // indirect + golang.org/x/crypto v0.51.0 // indirect + golang.org/x/mod v0.33.0 // indirect + golang.org/x/sys v0.44.0 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) + +// Both imported packages are local modules: the stdlib-only core (two levels +// up) and the franz-go-backed adapter (one level up, its own module). +replace github.com/stablekernel/crucible/source => ../../ + +replace github.com/stablekernel/crucible/source/kafka => ../ + +replace github.com/stablekernel/crucible/telemetry => ../../../telemetry diff --git a/source/kafka/integration/go.sum b/source/kafka/integration/go.sum new file mode 100644 index 0000000..c32579a --- /dev/null +++ b/source/kafka/integration/go.sum @@ -0,0 +1,149 @@ +dario.cat/mergo v1.0.2 h1:85+piFYR1tMbRrLcDwR18y4UKJ3aH1Tbzi24VRW1TK8= +dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA= +github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6 h1:He8afgbRMd7mFxO99hRNu+6tazq8nFF9lIwo9JFroBk= +github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8= +github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c h1:udKWzYgxTojEKWjV8V+WSxDXJ4NFATAsZjh8iIbsQIg= +github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= +github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= +github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= +github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= +github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG8PI= +github.com/containerd/errdefs v1.0.0/go.mod h1:+YBYIdtsnF4Iw6nWZhJcqGSg/dwvV7tyJ/kCkyJ2k+M= +github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151Xdx3ZPPE= +github.com/containerd/errdefs/pkg v0.3.0/go.mod h1:NJw6s9HwNuRhnjJhM7pylWwMyAkmCQvQ4GpJHEqRLVk= +github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I= +github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3EhrzVo= +github.com/containerd/platforms v0.2.1 h1:zvwtM3rz2YHPQsF2CHYM8+KtB5dvhISiXh5ZpSBQv6A= +github.com/containerd/platforms v0.2.1/go.mod h1:XHCb+2/hzowdiut9rkudds9bE5yJ7npe7dG/wG+uFPw= +github.com/cpuguy83/dockercfg v0.3.2 h1:DlJTyZGBDlXqUZ2Dk2Q3xHs/FtnooJJVaad2S9GKorA= +github.com/cpuguy83/dockercfg v0.3.2/go.mod h1:sugsbF4//dDlL/i+S+rtpIWp+5h0BHJHfjj5/jFyUJc= +github.com/creack/pty v1.1.24 h1:bJrF4RRfyJnbTJqzRLHzcGaZK1NeM5kTC9jGgovnR1s= +github.com/creack/pty v1.1.24/go.mod h1:08sCNb52WyoAwi2QDyzUCTgcvVFhUzewun7wtTfvcwE= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk= +github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E= +github.com/docker/go-connections v0.6.0 h1:LlMG9azAe1TqfR7sO+NJttz1gy6KO7VJBh+pMmjSD94= +github.com/docker/go-connections v0.6.0/go.mod h1:AahvXYshr6JgfUJGdDCs2b5EZG/vmaMAntpSFH5BFKE= +github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4= +github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= +github.com/ebitengine/purego v0.10.0 h1:QIw4xfpWT6GWTzaW5XEKy3HXoqrJGx1ijYHzTF0/ISU= +github.com/ebitengine/purego v0.10.0/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ= +github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= +github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= +github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY= +github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0= +github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao= +github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ81pIr0yLvtUWk2if982qA3F3QD6H4= +github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I= +github.com/magiconair/properties v1.8.10 h1:s31yESBquKXCV9a/ScB3ESkOjUYYv+X0rg8SYxI99mE= +github.com/magiconair/properties v1.8.10/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0= +github.com/mdelapenya/tlscert v0.2.0 h1:7H81W6Z/4weDvZBNOfQte5GpIMo0lGYEeWbkGp5LJHI= +github.com/mdelapenya/tlscert v0.2.0/go.mod h1:O4njj3ELLnJjGdkN7M/vIVCpZ+Cf0L6muqOG4tLSl8o= +github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= +github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo= +github.com/moby/go-archive v0.2.0 h1:zg5QDUM2mi0JIM9fdQZWC7U8+2ZfixfTYoHL7rWUcP8= +github.com/moby/go-archive v0.2.0/go.mod h1:mNeivT14o8xU+5q1YnNrkQVpK+dnNe/K6fHqnTg4qPU= +github.com/moby/moby/api v1.54.1 h1:TqVzuJkOLsgLDDwNLmYqACUuTehOHRGKiPhvH8V3Nn4= +github.com/moby/moby/api v1.54.1/go.mod h1:+RQ6wluLwtYaTd1WnPLykIDPekkuyD/ROWQClE83pzs= +github.com/moby/moby/client v0.4.0 h1:S+2XegzHQrrvTCvF6s5HFzcrywWQmuVnhOXe2kiWjIw= +github.com/moby/moby/client v0.4.0/go.mod h1:QWPbvWchQbxBNdaLSpoKpCdf5E+WxFAgNHogCWDoa7g= +github.com/moby/patternmatcher v0.6.1 h1:qlhtafmr6kgMIJjKJMDmMWq7WLkKIo23hsrpR3x084U= +github.com/moby/patternmatcher v0.6.1/go.mod h1:hDPoyOpDY7OrrMDLaYoY3hf52gNCR/YOUYxkhApJIxc= +github.com/moby/sys/sequential v0.6.0 h1:qrx7XFUd/5DxtqcoH1h438hF5TmOvzC/lspjy7zgvCU= +github.com/moby/sys/sequential v0.6.0/go.mod h1:uyv8EUTrca5PnDsdMGXhZe6CCe8U/UiTWd+lL+7b/Ko= +github.com/moby/sys/user v0.4.0 h1:jhcMKit7SA80hivmFJcbB1vqmw//wU61Zdui2eQXuMs= +github.com/moby/sys/user v0.4.0/go.mod h1:bG+tYYYJgaMtRKgEmuueC0hJEAZWwtIbZTB+85uoHjs= +github.com/moby/sys/userns v0.1.0 h1:tVLXkFOxVu9A64/yh59slHVv9ahO9UIev4JZusOLG/g= +github.com/moby/sys/userns v0.1.0/go.mod h1:IHUYgu/kao6N8YZlp9Cf444ySSvCmDlmzUcYfDHOl28= +github.com/moby/term v0.5.2 h1:6qk3FJAFDs6i/q3W/pQ97SX192qKfZgGjCQqfCJkgzQ= +github.com/moby/term v0.5.2/go.mod h1:d3djjFCrjnB+fl8NJux+EJzu0msscUP+f8it8hPkFLc= +github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= +github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= +github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040= +github.com/opencontainers/image-spec v1.1.1/go.mod h1:qpqAh3Dmcf36wStyyWU+kCeDgrGnAve2nCC8+7h8Q0M= +github.com/pierrec/lz4/v4 v4.1.26 h1:GrpZw1gZttORinvzBdXPUXATeqlJjqUG/D87TKMnhjY= +github.com/pierrec/lz4/v4 v4.1.26/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 h1:o4JXh1EVt9k/+g42oCprj/FisM4qX9L3sZB3upGN2ZU= +github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= +github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= +github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= +github.com/shirou/gopsutil/v4 v4.26.3 h1:2ESdQt90yU3oXF/CdOlRCJxrP+Am1aBYubTMTfxJ1qc= +github.com/shirou/gopsutil/v4 v4.26.3/go.mod h1:LZ6ewCSkBqUpvSOf+LsTGnRinC6iaNUNMGBtDkJBaLQ= +github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w= +github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g= +github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4= +github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/testcontainers/testcontainers-go v0.42.0 h1:He3IhTzTZOygSXLJPMX7n44XtK+qhjat1nI9cneBbUY= +github.com/testcontainers/testcontainers-go v0.42.0/go.mod h1:vZjdY1YmUA1qEForxOIOazfsrdyORJAbhi0bp8plN30= +github.com/testcontainers/testcontainers-go/modules/redpanda v0.42.0 h1:mcPLvf3rzvvwG46i4jcyOk7/KcapE/IKnBgwo27k68M= +github.com/testcontainers/testcontainers-go/modules/redpanda v0.42.0/go.mod h1:Yq0WrUIIsMkJoZ2DadCJ1Zq4RFhjPPpSNYHpjRrGfDU= +github.com/tklauser/go-sysconf v0.3.16 h1:frioLaCQSsF5Cy1jgRBrzr6t502KIIwQ0MArYICU0nA= +github.com/tklauser/go-sysconf v0.3.16/go.mod h1:/qNL9xxDhc7tx3HSRsLWNnuzbVfh3e7gh/BmM179nYI= +github.com/tklauser/numcpus v0.11.0 h1:nSTwhKH5e1dMNsCdVBukSZrURJRoHbSEQjdEbY+9RXw= +github.com/tklauser/numcpus v0.11.0/go.mod h1:z+LwcLq54uWZTX0u/bGobaV34u6V7KNlTZejzM6/3MQ= +github.com/twmb/franz-go v1.21.2 h1:WrvV/spF48JzcRylqDQy02Vm6V6W4lhtD9Y4BOYNMu4= +github.com/twmb/franz-go v1.21.2/go.mod h1:rfoMTnVk7107fhTGxfEKIHP/e7tPe6oyij/ywzO0czk= +github.com/twmb/franz-go/pkg/kadm v1.11.0 h1:FfeWJ0qadntFpAcQt8JzNXW4dijjytZNLrzJuzzzuxA= +github.com/twmb/franz-go/pkg/kadm v1.11.0/go.mod h1:qrhkdH+SWS3ivmbqOgHbpgVHamhaKcjH0UM+uOp0M1A= +github.com/twmb/franz-go/pkg/kmsg v1.13.1 h1:fG5kItwysTk5UXqVwb64EpQEy3TydF3vYYK21nUQ+bI= +github.com/twmb/franz-go/pkg/kmsg v1.13.1/go.mod h1:+DPt4NC8RmI6hqb8G09+3giKObE6uD2Eya6CfqBpeJY= +github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0= +github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 h1:sbiXRNDSWJOTobXh5HyQKjq6wUC5tNybqjIqDpAY4CU= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0/go.mod h1:69uWxva0WgAA/4bu2Yy70SLDBwZXuQ6PbBpbsa5iZrQ= +go.opentelemetry.io/otel v1.41.0 h1:YlEwVsGAlCvczDILpUXpIpPSL/VPugt7zHThEMLce1c= +go.opentelemetry.io/otel v1.41.0/go.mod h1:Yt4UwgEKeT05QbLwbyHXEwhnjxNO6D8L5PQP51/46dE= +go.opentelemetry.io/otel/metric v1.41.0 h1:rFnDcs4gRzBcsO9tS8LCpgR0dxg4aaxWlJxCno7JlTQ= +go.opentelemetry.io/otel/metric v1.41.0/go.mod h1:xPvCwd9pU0VN8tPZYzDZV/BMj9CM9vs00GuBjeKhJps= +go.opentelemetry.io/otel/sdk v1.35.0 h1:iPctf8iprVySXSKJffSS79eOjl9pvxV9ZqOWT0QejKY= +go.opentelemetry.io/otel/sdk v1.35.0/go.mod h1:+ga1bZliga3DxJ3CQGg3updiaAJoNECOgJREo9KHGQg= +go.opentelemetry.io/otel/sdk/metric v1.35.0 h1:1RriWBmCKgkeHEhM7a2uMjMUfP7MsOF5JpUCaEqEI9o= +go.opentelemetry.io/otel/sdk/metric v1.35.0/go.mod h1:is6XYCUMpcKi+ZsOvfluY5YstFnhW0BidkR+gL+qN+w= +go.opentelemetry.io/otel/trace v1.41.0 h1:Vbk2co6bhj8L59ZJ6/xFTskY+tGAbOnCtQGVVa9TIN0= +go.opentelemetry.io/otel/trace v1.41.0/go.mod h1:U1NU4ULCoxeDKc09yCWdWe+3QoyweJcISEVa1RBzOis= +golang.org/x/crypto v0.51.0 h1:IBPXwPfKxY7cWQZ38ZCIRPI50YLeevDLlLnyC5wRGTI= +golang.org/x/crypto v0.51.0/go.mod h1:8AdwkbraGNABw2kOX6YFPs3WM22XqI4EXEd8g+x7Oc8= +golang.org/x/mod v0.33.0 h1:tHFzIWbBifEmbwtGz65eaWyGiGZatSrT9prnU8DbVL8= +golang.org/x/mod v0.33.0/go.mod h1:swjeQEj+6r7fODbD2cqrnje9PnziFuw4bmLbBZFrQ5w= +golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.44.0 h1:ildZl3J4uzeKP07r2F++Op7E9B29JRUy+a27EibtBTQ= +golang.org/x/sys v0.44.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/term v0.43.0 h1:S4RLU2sB31O/NCl+zFN9Aru9A/Cq2aqKpTZJ6B+DwT4= +golang.org/x/term v0.43.0/go.mod h1:lrhlHNdQJHO+1qVYiHfFKVuVioJIheAc3fBSMFYEIsk= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gotest.tools/v3 v3.5.2 h1:7koQfIKdy+I8UTetycgUqXWSDwpgv193Ka+qRsmBY8Q= +gotest.tools/v3 v3.5.2/go.mod h1:LtdLGcnqToBH83WByAAi/wiwSFCArdFIUV/xxN4pcjA= +pgregory.net/rapid v1.2.0 h1:keKAYRcjm+e1F0oAuU5F5+YPAWcyxNNRK2wud503Gnk= +pgregory.net/rapid v1.2.0/go.mod h1:PY5XlDGj0+V1FCq0o192FdRhpKHGTRIWBgqjDBTrq04=