kafkareceiver with franz go enabled leads to panic for `OnPartitionLost`

[lahsivjar]

Component(s)

receiver/kafka

What happened?

Description

Panic being observed in logs for the receiver when UseFranzGo feature flag is turned on.

Steps to Reproduce

N/A

Expected Result

No panic

Actual Result

Panicing in partition lost callback

Collector version

de465d2

Environment information

Environment

OS: (e.g., "Ubuntu 20.04")
Compiler(if manually compiled): (e.g., "go 14.2")

OpenTelemetry Collector configuration

    receivers:
      kafka:
        brokers:
          - <redacted>
        tls:
          <redacted>
        logs:
          topic: <redacted>
          encoding: "otlp_proto"
        metrics:
          topic: <redacted>
          encoding: "otlp_proto"
        traces:
          topic: <redacted>
          encoding: "otlp_proto"
        group_id: <redacted>
        client_id: <redacted>
        session_timeout: "45s"
        heartbeat_interval: "5s"
        min_fetch_size: 819200
        max_fetch_size: 5e+07
        default_fetch_size: 5e+07
        max_fetch_wait: "10s"
        autocommit:
          enable: true
          interval: "1s"
        message_marking:
          after: true
          on_error: true
        error_backoff:
          enabled: true
          initial_interval: "100ms"
          max_interval: "5s"
          max_elapsed_time: "30s"
          multiplier: 2
          randomization_factor: 0.5


    connectors:
      signaltometrics:
        <redacted>

    exporters:
      kafka:
        brokers:
          - <redacted>
        metrics:
          encoding: "otlp_proto"
        sending_queue:
          enabled: false
        tls:
          <redacted>
        client_id: <redacted>

    service:
      extensions: []
      telemetry:
        <redacted>


      pipelines:
        traces:
          receivers: [kafka]
          processors: []
          exporters: [signaltometrics]
        metrics:
          receivers: [kafka]
          processors: []
          exporters: [signaltometrics]
        logs:
          receivers: [kafka]
          processors: []
          exporters: [signaltometrics]
        metrics/apm:
          receivers: [signaltometrics]
          processors: []
          exporters: [kafka]

The collector is running with the args:

            - '--feature-gates=exporter.kafkaexporter.UseFranzGo,receiver.kafkareceiver.UseFranzGo'

Log output

panic: runtime error: invalid memory address or nil pointer dereference [signal SIGSEGV: segmentation violation code=0x1 addr=0x28 pc=0x212bdc8] goroutine 5854 [running]: github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver.(*franzConsumer).lost(0x4000769080, {0x4e3fce0, 0x4000546780}, 0x1ec37a8?, 0x40010600c0, 0x0) github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver@v0.128.1-0.20250623222603-de465d285ea1/consumer_franz.go:382 +0x318 github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver.(*franzConsumer).Start.func1({0x4e3fce0?, 0x4000546780?}, 0x4000d10540?, 0x35?) github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver@v0.128.1-0.20250623222603-de465d285ea1/consumer_franz.go:134 +0x44 github.com/twmb/franz-go/pkg/kgo.(*consumer).initGroup.func2({0x4e3fce0, 0x4000546780}, 0x40004d9008, 0x40010e3d10) github.com/twmb/franz-go@v1.18.1/pkg/kgo/consumer_group.go:342 +0x320 github.com/twmb/franz-go/pkg/kgo.(*groupConsumer).revoke(0x400089f400, 0x0, 0x40010e3d10, 0x0) github.com/twmb/franz-go@v1.18.1/pkg/kgo/consumer_group.go:716 +0x904 github.com/twmb/franz-go/pkg/kgo.(*assignRevokeSession).prerevoke.func1() github.com/twmb/franz-go@v1.18.1/pkg/kgo/consumer_group.go:776 +0x84 created by github.com/twmb/franz-go/pkg/kgo.(*assignRevokeSession).prerevoke in goroutine 128 github.com/twmb/franz-go@v1.18.1/pkg/kgo/consumer_group.go:773 +0x84

Additional context

No response

[github-actions]

Pinging code owners:

• receiver/kafka: @pavolloffay @MovieStoreGuy @axw

See Adding Labels via Comments if you do not have permissions to add labels yourself.

[VihasMakwana]

@lahsivjar Thanks for reporting this!

Can you please attach your collector configuration as well?

[lahsivjar]

Looking into this further, it seems that we also need to protect against multiple calls for OnPartitionRevoked in addition to calls for OnPartitionLost. This is because, based on my debugging, for cooperative consumers when topics are deleted then OnPartitionRevoked can be called couple of times:

github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver.(*franzConsumer).lost(0x40003ef340, {0x4e3f380, 0x400017c910}, 0x1ec2e88?, 0x400079abd0, 0x0)
	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver@v0.128.1-0.20250623222603-de465d285ea1/consumer_franz.go:392 +0x2d8
github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver.(*franzConsumer).Start.func1({0x4e3f380?, 0x400017c910?}, 0x4000bba860?, 0x19?)
	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver@v0.128.1-0.20250623222603-de465d285ea1/consumer_franz.go:135 +0x44
github.com/twmb/franz-go/pkg/kgo.(*consumer).initGroup.func2({0x4e3f380, 0x400017c910}, 0x40007a4008, 0x400079a900)
	github.com/twmb/franz-go@v1.18.1/pkg/kgo/consumer_group.go:342 +0x320
github.com/twmb/franz-go/pkg/kgo.(*groupConsumer).revoke(0x4000894dc0, 0x1, 0x400079a900, 0x0)
	github.com/twmb/franz-go@v1.18.1/pkg/kgo/consumer_group.go:716 +0x904
github.com/twmb/franz-go/pkg/kgo.(*assignRevokeSession).revoke.func1()
	github.com/twmb/franz-go@v1.18.1/pkg/kgo/consumer_group.go:813 +0x78
created by github.com/twmb/franz-go/pkg/kgo.(*assignRevokeSession).revoke in goroutine 241
	github.com/twmb/franz-go@v1.18.1/pkg/kgo/consumer_group.go:810 +0x7c

The first call (stack above) is made due to the group session being revoked for the cooperative consumer at the end of the heartbeat loop. This call is made with a nil map for the lost (https://github.com/twmb/franz-go/blob/6ea03e3a3810f0b0db1ee02c6c3abaf79ff7f8d8/pkg/kgo/consumer_group.go#L813) which is later created dynamically (ref). On our end, in the kafkareceiver, we delete the assignments at this point for the lost/deleted topic.

github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver.(*franzConsumer).lost(0x40003ef340, {0x4e3f380, 0x400017c910}, 0x1ec2e88?, 0x400070b980, 0x0)
	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver@v0.128.1-0.20250623222603-de465d285ea1/consumer_franz.go:392 +0x2d8
github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver.(*franzConsumer).Start.func1({0x4e3f380?, 0x400017c910?}, 0x4000bba860?, 0x19?)
	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver@v0.128.1-0.20250623222603-de465d285ea1/consumer_franz.go:135 +0x44
github.com/twmb/franz-go/pkg/kgo.(*consumer).initGroup.func2({0x4e3f380, 0x400017c910}, 0x40007a4008, 0x400070b7a0)
	github.com/twmb/franz-go@v1.18.1/pkg/kgo/consumer_group.go:342 +0x320
github.com/twmb/franz-go/pkg/kgo.(*groupConsumer).revoke(0x4000894dc0, 0x0, 0x400070b7a0, 0x0)
	github.com/twmb/franz-go@v1.18.1/pkg/kgo/consumer_group.go:716 +0x904
github.com/twmb/franz-go/pkg/kgo.(*assignRevokeSession).prerevoke.func1()
	github.com/twmb/franz-go@v1.18.1/pkg/kgo/consumer_group.go:776 +0x84
created by github.com/twmb/franz-go/pkg/kgo.(*assignRevokeSession).prerevoke in goroutine 95
	github.com/twmb/franz-go@v1.18.1/pkg/kgo/consumer_group.go:773 +0x84

The second call (stack above) is a prerevoke call which the consumer group makes to revoke the last session (ref). At this point, the previous assignment is diffed with the current assignement and a map of lost partitions is created which is then revoked. It is a bit strange at this point why the topic deletion didn't update the previous assignement as that would have fixed the lost map. For now, I think our code should protect against both these cases: OnPartitionLost calling the callback even when no joining happens as well as OnPartitionRevoked being called more than one time with the same lost/deleted topic.