Updating kfake fails TestConsumerDelivery

[rubvs]

A new error is observed when updating franz-go and kfake, see Failing PR.

logger.go:146: 2025-05-28T10:39:54.773-0400     ERROR   1.kafka.partition       data loss: unable to process event      {"namespace": "name_space", "topic": "topic", "partition": 0, "error": "failed processing record", "offset": 0, "headers": {}}
logger.go:146: 2025-05-28T10:39:54.774-0400     ERROR   1.kafka.partition       data loss: unable to process event      {"namespace": "name_space", "topic": "topic", "partition": 0, "error": "failed processing record", "offset": 1, "headers": {}}
logger.go:146: 2025-05-28T10:39:54.774-0400     ERROR   1.kafka.partition       data loss: unable to process event      {"namespace": "name_space", "topic": "topic", "partition": 0, "error": "extra records are lost", "offset": 2, "headers": {}}
logger.go:146: 2025-05-28T10:39:54.775-0400     ERROR   1.kafka.partition       data loss: unable to process event      {"namespace": "name_space", "topic": "topic", "partition": 0, "error": "extra records are lost", "offset": 3, "headers": {}}
logger.go:146: 2025-05-28T10:39:54.775-0400     INFO    1.kafka.kafka   assigning partitions    {"namespace": "name_space", "why": "invalidating all assignments in LeaveGroup", "how": "unassigning everything", "input": ""}
logger.go:146: 2025-05-28T10:39:54.775-0400     INFO    1.kafka.kafka   heartbeat errored       {"namespace": "name_space", "group": "groupid", "err": "context canceled"}
logger.go:146: 2025-05-28T10:39:54.776-0400     INFO    1.kafka.kafka   assigning partitions    {"namespace": "name_space", "why": "clearing assignment at end of group management session", "how": "unassigning everything", "input": ""}
logger.go:146: 2025-05-28T10:39:54.776-0400     INFO    1.kafka.kafka   leaving group   {"namespace": "name_space", "group": "groupid", "member_id": "kgo-e4ee4cfac5d8b871e12f3743b8d22f21"}
consumer_test.go:381: got: 2 events errored, 0 processed, want: 2 errored, 26 processed
consumer_test.go:390:
        Error Trace:    /Users/rubenvanstaden/issue/apm_queue/661/apm-queue/kafka/consumer_test.go:927
                                                /Users/rubenvanstaden/issue/apm_queue/661/apm-queue/kafka/consumer_test.go:390
        Error:          Received unexpected error:
                        context canceled
        Test:           TestConsumerDelivery/30_produced_2_poll_AMOD

[rubvs]

Tested using a local Kafka broker with Redpanda.

• Terminal 1

> docker run --rm -it --name=source \
    -p 8081:8081 \
    -p 9092:9092 \
    -p 9644:9644 \
    redpandadata/redpanda redpanda start \
    --node-id 0 \
    --mode dev-container \
    --set "rpk.additional_start_flags=[--reactor-backend=epoll]" \
    --set redpanda.auto_create_topics_enabled=true \
    --kafka-addr 0.0.0.0:9092 \
    --advertise-kafka-addr localhost:9092 \
    --schema-registry-addr 0.0.0.0:8081

• Terminal 2

> go get -u github.com/twmb/franz-go/pkg/kfake
> go test -v -timeout=30s ./... -run TestConsumerDelivery

?       github.com/elastic/apm-queue/v2 [no test files]
?       github.com/elastic/apm-queue/v2/cmd/queuebench  [no test files]
=== RUN   TestConsumerDelivery
=== RUN   TestConsumerDelivery/12_produced_1_poll_ALOD
    consumer_test.go:338:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:925
                                                        /tmp/apm-queue/kafka/consumer_test.go:338
                Error:          Received unexpected error:
                                UNKNOWN_TOPIC_OR_PARTITION: This server does not host this topic-partition.
                Test:           TestConsumerDelivery/12_produced_1_poll_ALOD
    consumer_test.go:338:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:927
                                                        /tmp/apm-queue/kafka/consumer_test.go:338
                Error:          Received unexpected error:
                                UNKNOWN_TOPIC_OR_PARTITION: This server does not host this topic-partition.
                Test:           TestConsumerDelivery/12_produced_1_poll_ALOD
    logger.go:146: 2025-06-14T20:13:58.080-0400 INFO    1.kafka.kafka   immediate metadata update triggered     {"namespace": "name_space", "why": "querying metadata for consumer initialization"}
    consumer_test.go:366: timed out waiting for consumer to process event
    logger.go:146: 2025-06-14T20:13:59.081-0400 INFO    1.kafka.kafka   assigning partitions    {"namespace": "name_space", "why": "invalidating all assignments in LeaveGroup", "how": "unassigning everything", "input": ""}
    logger.go:146: 2025-06-14T20:13:59.081-0400 INFO    1.kafka.kafka   tried to leave group but we have no member ID yet, returning early      {"namespace": "name_space", "group": "groupid"}
=== RUN   TestConsumerDelivery/1_produced_1_poll_ALOD
    consumer_test.go:338:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:925
                                                        /tmp/apm-queue/kafka/consumer_test.go:338
                Error:          Received unexpected error:
                                UNKNOWN_TOPIC_OR_PARTITION: This server does not host this topic-partition.
                Test:           TestConsumerDelivery/1_produced_1_poll_ALOD
    consumer_test.go:338:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:927
                                                        /tmp/apm-queue/kafka/consumer_test.go:338
                Error:          Received unexpected error:
                                UNKNOWN_TOPIC_OR_PARTITION: This server does not host this topic-partition.
                Test:           TestConsumerDelivery/1_produced_1_poll_ALOD
    logger.go:146: 2025-06-14T20:14:00.123-0400 INFO    1.kafka.kafka   immediate metadata update triggered     {"namespace": "name_space", "why": "querying metadata for consumer initialization"}
    consumer_test.go:366: timed out waiting for consumer to process event
    logger.go:146: 2025-06-14T20:14:01.124-0400 INFO    1.kafka.kafka   assigning partitions    {"namespace": "name_space", "why": "invalidating all assignments in LeaveGroup", "how": "unassigning everything", "input": ""}
    logger.go:146: 2025-06-14T20:14:01.125-0400 INFO    1.kafka.kafka   tried to leave group but we have no member ID yet, returning early      {"namespace": "name_space", "group": "groupid"}
=== RUN   TestConsumerDelivery/12_produced_10_poll_AMOD
    consumer_test.go:338:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:925
                                                        /tmp/apm-queue/kafka/consumer_test.go:338
                Error:          Received unexpected error:
                                UNKNOWN_TOPIC_OR_PARTITION: This server does not host this topic-partition.
                Test:           TestConsumerDelivery/12_produced_10_poll_AMOD
    consumer_test.go:338:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:927
                                                        /tmp/apm-queue/kafka/consumer_test.go:338
                Error:          Received unexpected error:
                                UNKNOWN_TOPIC_OR_PARTITION: This server does not host this topic-partition.
                Test:           TestConsumerDelivery/12_produced_10_poll_AMOD
    consumer_test.go:338:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:925
                                                        /tmp/apm-queue/kafka/consumer_test.go:338
                Error:          Received unexpected error:
                                UNKNOWN_TOPIC_OR_PARTITION: This server does not host this topic-partition.
                Test:           TestConsumerDelivery/12_produced_10_poll_AMOD
    consumer_test.go:338:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:927
                                                        /tmp/apm-queue/kafka/consumer_test.go:338
                Error:          Received unexpected error:
                                UNKNOWN_TOPIC_OR_PARTITION: This server does not host this topic-partition.
                Test:           TestConsumerDelivery/12_produced_10_poll_AMOD
panic: test timed out after 30s
        running tests:
                TestConsumerDelivery (30s)
                TestConsumerDelivery/12_produced_10_poll_AMOD (26s)

goroutine 329 [running]:
testing.(*M).startAlarm.func1()
        /usr/local/go/src/testing/testing.go:2484 +0x308
created by time.goFunc
        /usr/local/go/src/time/sleep.go:215 +0x38

goroutine 1 [chan receive]:
testing.(*T).Run(0x14000182a80, {0x1036b17aa?, 0x14000259b38?}, 0x103a5e928)
        /usr/local/go/src/testing/testing.go:1859 +0x388
testing.runTests.func1(0x14000182a80)
        /usr/local/go/src/testing/testing.go:2279 +0x40
testing.tRunner(0x14000182a80, 0x14000259c68)
        /usr/local/go/src/testing/testing.go:1792 +0xe4
testing.runTests(0x1400019e2a0, {0x104112de0, 0x21, 0x21}, {0x140001a72c0?, 0x7?, 0x104119da0?})
        /usr/local/go/src/testing/testing.go:2277 +0x3ec
testing.(*M).Run(0x140002af0e0)
        /usr/local/go/src/testing/testing.go:2142 +0x588
main.main()
        _testmain.go:111 +0x90

goroutine 35 [chan receive]:
testing.(*T).Run(0x14000182c40, {0x1036b5c56?, 0x14000198ef8?}, 0x140001d6c00)
        /usr/local/go/src/testing/testing.go:1859 +0x388
github.com/elastic/apm-queue/v2/kafka.TestConsumerDelivery(0x14000182c40)
        /tmp/apm-queue/kafka/consumer_test.go:283 +0x3c4
testing.tRunner(0x14000182c40, 0x103a5e928)
        /usr/local/go/src/testing/testing.go:1792 +0xe4
created by testing.(*T).Run in goroutine 1
        /usr/local/go/src/testing/testing.go:1851 +0x374

goroutine 192 [sync.WaitGroup.Wait]:
sync.runtime_SemacquireWaitGroup(0x140003149c0?)
        /usr/local/go/src/runtime/sema.go:110 +0x2c
sync.(*WaitGroup).Wait(0x140003149b0)
        /usr/local/go/src/sync/waitgroup.go:118 +0x70
github.com/twmb/franz-go/pkg/kgo.(*Client).ProduceSync(0x14000390a88, {0x103a71c10, 0x140003ba050}, {0x14000049af0, 0x1, 0x0?})
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/producer.go:325 +0x138
github.com/elastic/apm-queue/v2/kafka.produceRecord({0x103a71c10, 0x140003ba050}, {0x103a7d300, 0x140002eae00}, 0x14000390a88, 0x140002af400)
        /tmp/apm-queue/kafka/consumer_test.go:924 +0x64
github.com/elastic/apm-queue/v2/kafka.TestConsumerDelivery.func1(0x140002eae00)
        /tmp/apm-queue/kafka/consumer_test.go:338 +0x308
testing.tRunner(0x140002eae00, 0x140001d6c00)
        /usr/local/go/src/testing/testing.go:1792 +0xe4
created by testing.(*T).Run in goroutine 35
        /usr/local/go/src/testing/testing.go:1851 +0x374

goroutine 37 [select]:
github.com/twmb/franz-go/pkg/kgo.(*Client).updateMetadataLoop(0x140002baa88)
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/metadata.go:209 +0x6a8
created by github.com/twmb/franz-go/pkg/kgo.NewClient in goroutine 36
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/client.go:562 +0xa54

goroutine 38 [select]:
github.com/twmb/franz-go/pkg/kgo.(*Client).reapConnectionsLoop(0x140002baa88)
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/broker.go:617 +0xe8
created by github.com/twmb/franz-go/pkg/kgo.NewClient in goroutine 36
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/client.go:563 +0xa94

goroutine 39 [select]:
github.com/twmb/franz-go/pkg/kgo.(*Client).pushMetrics(0x140002baa88)
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/metrics_714.go:29 +0x90
created by github.com/twmb/franz-go/pkg/kgo.NewClient in goroutine 36
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/client.go:564 +0xad4

goroutine 113 [select]:
github.com/twmb/franz-go/pkg/kgo.(*Client).reapConnectionsLoop(0x1400040a008)
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/broker.go:617 +0xe8
created by github.com/twmb/franz-go/pkg/kgo.NewClient in goroutine 95
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/client.go:563 +0xa94

goroutine 96 [select]:
github.com/twmb/franz-go/pkg/kgo.(*Client).updateMetadataLoop(0x1400040a008)
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/metadata.go:209 +0x6a8
created by github.com/twmb/franz-go/pkg/kgo.NewClient in goroutine 95
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/client.go:562 +0xa54

goroutine 195 [select]:
github.com/twmb/franz-go/pkg/kgo.(*Client).pushMetrics(0x14000390a88)
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/metrics_714.go:29 +0x90
created by github.com/twmb/franz-go/pkg/kgo.NewClient in goroutine 192
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/client.go:564 +0xad4

goroutine 114 [select]:
github.com/twmb/franz-go/pkg/kgo.(*Client).pushMetrics(0x1400040a008)
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/metrics_714.go:29 +0x90
created by github.com/twmb/franz-go/pkg/kgo.NewClient in goroutine 95
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/client.go:564 +0xad4

goroutine 193 [select]:
github.com/twmb/franz-go/pkg/kgo.(*Client).updateMetadataLoop(0x14000390a88)
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/metadata.go:209 +0x6a8
created by github.com/twmb/franz-go/pkg/kgo.NewClient in goroutine 192
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/client.go:562 +0xa54

goroutine 194 [select]:
github.com/twmb/franz-go/pkg/kgo.(*Client).reapConnectionsLoop(0x14000390a88)
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/broker.go:617 +0xe8
created by github.com/twmb/franz-go/pkg/kgo.NewClient in goroutine 192
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/client.go:563 +0xa94

goroutine 285 [select]:
github.com/twmb/franz-go/pkg/kgo.(*Client).waitUnknownTopic(0x14000390a88, {0x103a71c10, 0x140003ba050}, {0x103a71c10, 0x140003ba050}, {0x1036adbed, 0x10}, 0x1400041b410)
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/producer.go:1073 +0x254
created by github.com/twmb/franz-go/pkg/kgo.(*Client).addUnknownTopicRecord in goroutine 192
        /Users/rubenvanstaden/go/pkg/mod/github.com/twmb/franz-go@v1.19.5/pkg/kgo/producer.go:1039 +0x2e0
FAIL    github.com/elastic/apm-queue/v2/kafka   30.450s
?       github.com/elastic/apm-queue/v2/metrictest      [no test files]
testing: warning: no tests to run
PASS
ok      github.com/elastic/apm-queue/v2/queuecontext    (cached) [no tests to run]
2025-06-14T20:13:57.024-0400    INFO    systemtest/main_test.go:40      skipping system tests, pass -run-system-tests to enable
ok      github.com/elastic/apm-queue/v2/systemtest      0.431s
FAIL

[rubvs]

Manually creating the topic then running the tests gives the following.

> rpk topic create name_space-topic
> go test -v -timeout=10s ./... -run TestConsumerDelivery

=== RUN   TestConsumerDelivery
=== RUN   TestConsumerDelivery/12_produced_1_poll_AMOD
    logger.go:146: 2025-06-14T20:51:16.702-0400 INFO    1.kafka.kafka   immediate metadata update triggered     {"namespace": "name_space", "why": "querying metadata for consumer initialization"}
    logger.go:146: 2025-06-14T20:51:16.705-0400 INFO    1.kafka.kafka   beginning to manage the group lifecycle {"namespace": "name_space", "group": "groupid"}
    logger.go:146: 2025-06-14T20:51:16.705-0400 INFO    1.kafka.kafka   joining group   {"namespace": "name_space", "group": "groupid"}
    logger.go:146: 2025-06-14T20:51:16.709-0400 INFO    1.kafka.kafka   join returned MemberIDRequired, rejoining with response's MemberID      {"namespace": "name_space", "group": "groupid", "member_id": "kgo-b1c7def1-dd2e-494a-9e90-7c2395f0674e"}
    logger.go:146: 2025-06-14T20:51:16.709-0400 INFO    1.kafka.kafka   joined, balancing group {"namespace": "name_space", "group": "groupid", "member_id": "kgo-b1c7def1-dd2e-494a-9e90-7c2395f0674e", "instance_id": "<nil>", "generation": 1, "balance_protocol": "cooperative-sticky", "leader": true}
    logger.go:146: 2025-06-14T20:51:16.709-0400 INFO    1.kafka.kafka   balancing group as leader       {"namespace": "name_space"}
    logger.go:146: 2025-06-14T20:51:16.709-0400 INFO    1.kafka.kafka   balance group member    {"namespace": "name_space", "id": "kgo-b1c7def1-dd2e-494a-9e90-7c2395f0674e", "interests": "interested topics: [name_space-topic], previously owned: "}
    logger.go:146: 2025-06-14T20:51:16.710-0400 INFO    1.kafka.kafka   balanced        {"namespace": "name_space", "plan": "kgo-b1c7def1-dd2e-494a-9e90-7c2395f0674e{name_space-topic[0]}"}
    logger.go:146: 2025-06-14T20:51:16.710-0400 INFO    1.kafka.kafka   syncing {"namespace": "name_space", "group": "groupid", "protocol_type": "consumer", "protocol": "cooperative-sticky"}
    logger.go:146: 2025-06-14T20:51:16.710-0400 INFO    1.kafka.kafka   synced  {"namespace": "name_space", "group": "groupid", "assigned": "name_space-topic[0]"}
    logger.go:146: 2025-06-14T20:51:16.710-0400 INFO    1.kafka.kafka   new group session begun {"namespace": "name_space", "group": "groupid", "added": "name_space-topic[0]", "lost": ""}
    logger.go:146: 2025-06-14T20:51:16.710-0400 INFO    1.kafka.kafka   beginning heartbeat loop        {"namespace": "name_space", "group": "groupid"}
    logger.go:146: 2025-06-14T20:51:16.711-0400 INFO    1.kafka.kafka   assigning partitions    {"namespace": "name_space", "why": "newly fetched offsets for group groupid", "how": "assigning everything new, keeping current assignment", "input": "name_space-topic[0{-2 e-1 ce0}]"}
    logger.go:146: 2025-06-14T20:51:16.715-0400 ERROR   1.kafka.partition       data loss: unable to process event      {"namespace": "name_space", "topic": "topic", "partition": 0, "error": "failed processing record", "offset": 0, "headers": {}}
    logger.go:146: 2025-06-14T20:51:16.716-0400 INFO    1.kafka.kafka   assigning partitions    {"namespace": "name_space", "why": "invalidating all assignments in LeaveGroup", "how": "unassigning everything", "input": ""}
    logger.go:146: 2025-06-14T20:51:16.716-0400 INFO    1.kafka.kafka   heartbeat errored       {"namespace": "name_space", "group": "groupid", "err": "context canceled"}
    logger.go:146: 2025-06-14T20:51:16.716-0400 INFO    1.kafka.kafka   assigning partitions    {"namespace": "name_space", "why": "clearing assignment at end of group management session", "how": "unassigning everything", "input": ""}
    logger.go:146: 2025-06-14T20:51:16.716-0400 INFO    1.kafka.kafka   leaving group   {"namespace": "name_space", "group": "groupid", "member_id": "kgo-b1c7def1-dd2e-494a-9e90-7c2395f0674e"}
    consumer_test.go:322: got: 1 events errored, 0 processed, want: 1 errored, 11 processed
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:868
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:870
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:868
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:870
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:868
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:870
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:868
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:870
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:868
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:870
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:868
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:870
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:868
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:870
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:868
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:870
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:868
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:870
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:868
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:870
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:868
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    consumer_test.go:331:
                Error Trace:    /tmp/apm-queue/kafka/consumer_test.go:870
                                                        /tmp/apm-queue/kafka/consumer_test.go:331
                Error:          Received unexpected error:
                                context canceled
                Test:           TestConsumerDelivery/12_produced_1_poll_AMOD
    logger.go:146: 2025-06-14T20:51:16.723-0400 INFO    2.kafka.kafka   immediate metadata update triggered     {"namespace": "name_space", "why": "querying metadata for consumer initialization"}
    logger.go:146: 2025-06-14T20:51:16.728-0400 INFO    2.kafka.kafka   beginning to manage the group lifecycle {"namespace": "name_space", "group": "groupid"}
    logger.go:146: 2025-06-14T20:51:16.728-0400 INFO    2.kafka.kafka   joining group   {"namespace": "name_space", "group": "groupid"}
    logger.go:146: 2025-06-14T20:51:16.732-0400 INFO    2.kafka.kafka   join returned MemberIDRequired, rejoining with response's MemberID      {"namespace": "name_space", "group": "groupid", "member_id": "kgo-eebbb980-fdf9-40c1-9744-2ad7d2ed605d"}
    logger.go:146: 2025-06-14T20:51:16.733-0400 INFO    2.kafka.kafka   joined, balancing group {"namespace": "name_space", "group": "groupid", "member_id": "kgo-eebbb980-fdf9-40c1-9744-2ad7d2ed605d", "instance_id": "<nil>", "generation": 3, "balance_protocol": "cooperative-sticky", "leader": true}
    logger.go:146: 2025-06-14T20:51:16.733-0400 INFO    2.kafka.kafka   balancing group as leader       {"namespace": "name_space"}
    logger.go:146: 2025-06-14T20:51:16.733-0400 INFO    2.kafka.kafka   balance group member    {"namespace": "name_space", "id": "kgo-eebbb980-fdf9-40c1-9744-2ad7d2ed605d", "interests": "interested topics: [name_space-topic], previously owned: "}
    logger.go:146: 2025-06-14T20:51:16.733-0400 INFO    2.kafka.kafka   balanced        {"namespace": "name_space", "plan": "kgo-eebbb980-fdf9-40c1-9744-2ad7d2ed605d{name_space-topic[0]}"}
    logger.go:146: 2025-06-14T20:51:16.733-0400 INFO    2.kafka.kafka   syncing {"namespace": "name_space", "group": "groupid", "protocol_type": "consumer", "protocol": "cooperative-sticky"}
    logger.go:146: 2025-06-14T20:51:16.733-0400 INFO    2.kafka.kafka   synced  {"namespace": "name_space", "group": "groupid", "assigned": "name_space-topic[0]"}
    logger.go:146: 2025-06-14T20:51:16.733-0400 INFO    2.kafka.kafka   new group session begun {"namespace": "name_space", "group": "groupid", "added": "name_space-topic[0]", "lost": ""}
    logger.go:146: 2025-06-14T20:51:16.734-0400 INFO    2.kafka.kafka   beginning heartbeat loop        {"namespace": "name_space", "group": "groupid"}
    logger.go:146: 2025-06-14T20:51:16.734-0400 INFO    2.kafka.kafka   assigning partitions    {"namespace": "name_space", "why": "newly fetched offsets for group groupid", "how": "assigning everything new, keeping current assignment", "input": "name_space-topic[0{1 e1 ce0}]"}
    consumer_test.go:349: timed out waiting for consumer to process event
    logger.go:146: 2025-06-14T20:51:22.724-0400 INFO    2.kafka.kafka   assigning partitions    {"namespace": "name_space", "why": "invalidating all assignments in LeaveGroup", "how": "unassigning everything", "input": ""}
    logger.go:146: 2025-06-14T20:51:22.724-0400 INFO    2.kafka.kafka   heartbeat errored       {"namespace": "name_space", "group": "groupid", "err": "context canceled"}
    logger.go:146: 2025-06-14T20:51:22.724-0400 INFO    2.kafka.kafka   assigning partitions    {"namespace": "name_space", "why": "clearing assignment at end of group management session", "how": "unassigning everything", "input": ""}
    logger.go:146: 2025-06-14T20:51:22.725-0400 INFO    2.kafka.kafka   leaving group   {"namespace": "name_space", "group": "groupid", "member_id": "kgo-eebbb980-fdf9-40c1-9744-2ad7d2ed605d"}
--- FAIL: TestConsumerDelivery (6.38s)
    --- FAIL: TestConsumerDelivery/12_produced_1_poll_AMOD (6.38s)
FAIL
FAIL    github.com/elastic/apm-queue/v2/kafka   6.867s
?       github.com/elastic/apm-queue/v2/metrictest      [no test files]
testing: warning: no tests to run
PASS
ok      github.com/elastic/apm-queue/v2/queuecontext    (cached) [no tests to run]
2025-06-14T20:51:16.136-0400    INFO    systemtest/main_test.go:40      skipping system tests, pass -run-system-tests to enable
ok      github.com/elastic/apm-queue/v2/systemtest      0.256s
FAIL

[simitt]

@rubvs are you looking for futher guidance on how to proceed here or do you have a path forward that you are approaching?