Merge branch 'main' into main

Author: sv-splunk

.chloggen/40920-fix-franzgo-panic.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: kafkareceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Protect partition lost/revoke callback against unassigned groups
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [40920]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

receiver/kafkareceiver/consumer_franz.go

@@ -378,17 +378,25 @@ func (c *franzConsumer) lost(ctx context.Context, _ *kgo.Client,
 	for topic, partitions := range lost {
 		for _, partition := range partitions {
 			tp := topicPartition{topic: topic, partition: partition}
-			pc := c.assignments[tp]
-			delete(c.assignments, tp)
-			pc.cancel(errors.New(
-				"stopping processing: partition reassigned or lost",
-			))
-			wg.Add(1)
-			go func() {
-				defer wg.Done()
-				pc.wg.Wait()
-			}()
-			c.telemetryBuilder.KafkaReceiverPartitionClose.Add(context.Background(), 1)
+			// In some cases, it is possible for the `lost` to be called with
+			// no assignments. So, check if assignments exists first.
+			//
+			// - OnPartitionLost can be called without the group ever joining
+			// and getting assigned.
+			// - OnPartitionRevoked can be called multiple times for cooperative
+			// balancer on topic lost/deleted.
+			if pc, ok := c.assignments[tp]; ok {
+				delete(c.assignments, tp)
+				pc.cancel(errors.New(
+					"stopping processing: partition reassigned or lost",
+				))
+				wg.Add(1)
+				go func() {
+					defer wg.Done()
+					pc.wg.Wait()
+				}()
+				c.telemetryBuilder.KafkaReceiverPartitionClose.Add(context.Background(), 1)
+			}
 		}
 	}
 	if fatal {

receiver/kafkareceiver/consumer_franz_test.go

@@ -243,3 +243,29 @@ func TestConsumerShutdownNotStarted(t *testing.T) {
 			"kafka consumer: consumer isn't running")
 	}
 }
+
+func TestLost(t *testing.T) {
+	// It is possible that lost is called multiple times for the same partition
+	// or called with a topic/partition that hasn't been assigned. This test
+	// ensures that `lost` works without error in both cases.
+	_, cfg := mustNewFakeCluster(t, kfake.SeedTopics(1, "test"))
+	settings, _, _ := mustNewSettings(t)
+
+	consumeFn := func(component.Host, *receiverhelper.ObsReport, *metadata.TelemetryBuilder) (consumeMessageFunc, error) {
+		return func(_ context.Context, _ kafkaMessage, _ attribute.Set) error {
+			return nil
+		}, nil
+	}
+	c, err := newFranzKafkaConsumer(cfg, settings, []string{"test"}, consumeFn)
+	require.NoError(t, err)
+	require.NoError(t, c.Start(context.Background(), componenttest.NewNopHost()))
+	defer func() { require.NoError(t, c.Shutdown(context.Background())) }()
+
+	// Call lost couple of times for same partition
+	lostM := map[string][]int32{"test": {0}}
+	c.lost(context.Background(), nil, lostM, false)
+	c.lost(context.Background(), nil, lostM, false)
+
+	// Call lost for a topic and partition that was not assigned
+	c.lost(context.Background(), nil, map[string][]int32{"404": {0}}, true)
+}