kafkareceiver: don't restart when marking errors (#40140)

#### Description

If processing of a message fails but is still marked as consumed due to
the `message_marking` configuration, log an error and continue instead
of restarting the consumer. Restarting will not affect the outcome since
the message will not be reprocessed due to being marked.

#### Link to tracking issue

Fixes #39909

#### Testing

Added unit tests.

#### Documentation

None

Author: axw

.chloggen/kafkareceiver-messagemarking.yaml

@@ -0,0 +1,31 @@
+# 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: Don't restart the Kafka consumer on failed errors when message marking is enabled for them.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [39909]
+
+# (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: |
+  The Kafka consumer will now only be restarted due to failed messages if
+  both `message_marking::after` is true, and `message_marking::on_error` is false.
+  In this case the message will not be marked as consumed, and the consumer will
+  restart with an error.
+
+# 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/kafka_receiver.go

@@ -528,10 +528,19 @@ func (c *consumerGroupHandler) handleMessage(
 			c.logger.Info("Stop error backoff because the configured max_elapsed_time is reached",
 				zap.Duration("max_elapsed_time", c.backOff.MaxElapsedTime))
 		}
-		if c.messageMarking.After && c.messageMarking.OnError {
-			session.MarkMessage(message, "")
+		if c.messageMarking.After && !c.messageMarking.OnError {
+			// Only return an error if messages are marked after successful processing.
+			return err
 		}
-		return err
+		// We're either marking messages as consumed ahead of time (disregarding outcome),
+		// or after processing but including errors. Either way we should not return an error,
+		// as that will restart the consumer unnecessarily.
+		c.logger.Error("failed to consume message, skipping due to message_marking config",
+			zap.Error(err),
+			zap.String("topic", message.Topic),
+			zap.Int32("partition", claim.Partition()),
+			zap.Int64("offset", message.Offset),
+		)
 	}
 	if c.backOff != nil {
 		c.resetBackoff()

receiver/kafkareceiver/kafka_receiver_test.go

@@ -292,10 +292,7 @@ func TestReceiver_InternalTelemetry(t *testing.T) {
 
 	// After receiving messages, the internal metrics should be updated.
 	metadatatest.AssertEqualKafkaReceiverPartitionStart(t, tel, []metricdata.DataPoint[int64]{{
-		// 2 because:
-		//  - the initial open
-		//  - the invalid message causes the consumer to restart, closing the partition
-		Value:      2,
+		Value:      1,
 		Attributes: attribute.NewSet(attribute.String("name", set.ID.Name())),
 	}}, metricdatatest.IgnoreTimestamp())
 
@@ -312,19 +309,17 @@ func TestReceiver_InternalTelemetry(t *testing.T) {
 	err = r.Shutdown(context.Background())
 	require.NoError(t, err)
 	metadatatest.AssertEqualKafkaReceiverPartitionClose(t, tel, []metricdata.DataPoint[int64]{{
-		// 2 because:
-		//  - the invalid message causes the consumer to restart, closing the partition
-		//  - it re-acquires the partition, but then shutting down closes the partition again
-		Value: 2,
+		Value: 1,
 		Attributes: attribute.NewSet(
 			attribute.String("name", set.ID.Name()),
 		),
 	}}, metricdatatest.IgnoreTimestamp())
 
 	observedErrorLogs := observedLogs.FilterLevelExact(zapcore.ErrorLevel)
 	logEntries := observedErrorLogs.All()
-	assert.Len(t, logEntries, 1)
+	assert.Len(t, logEntries, 2)
 	assert.Equal(t, "failed to unmarshal message", logEntries[0].Message)
+	assert.Equal(t, "failed to consume message, skipping due to message_marking config", logEntries[1].Message)
 
 	metadatatest.AssertEqualKafkaReceiverCurrentOffset(t, tel, []metricdata.DataPoint[int64]{{
 		Value: 4, // offset of the final message
@@ -345,6 +340,104 @@ func TestReceiver_InternalTelemetry(t *testing.T) {
 	}}, metricdatatest.IgnoreTimestamp())
 }
 
+func TestReceiver_MessageMarking(t *testing.T) {
+	t.Parallel()
+	for name, testcase := range map[string]struct {
+		markAfter  bool
+		markErrors bool
+
+		errorShouldRestart bool
+	}{
+		"mark_before": {
+			markAfter: false,
+		},
+		"mark_after_success": {
+			markAfter:          true,
+			errorShouldRestart: true,
+		},
+		"mark_after_all": {
+			markAfter:  true,
+			markErrors: true,
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+			kafkaClient, receiverConfig := mustNewFakeCluster(t, kfake.SeedTopics(1, "otlp_spans"))
+
+			// Send some invalid data to the otlp_spans topic so unmarshaling fails,
+			// and then send some valid data to show that the invalid data does not
+			// block the consumer.
+			traces := testdata.GenerateTraces(1)
+			data, err := (&ptrace.ProtoMarshaler{}).MarshalTraces(traces)
+			require.NoError(t, err)
+			results := kafkaClient.ProduceSync(context.Background(),
+				&kgo.Record{Topic: "otlp_spans", Value: []byte("junk")},
+				&kgo.Record{Topic: "otlp_spans", Value: data},
+			)
+			require.NoError(t, results.FirstErr())
+
+			var calls atomic.Int64
+			consumer := newTracesConsumer(func(_ context.Context, received ptrace.Traces) error {
+				calls.Add(1)
+				return ptracetest.CompareTraces(traces, received)
+			})
+
+			// Only mark messages after consuming, including for errors.
+			receiverConfig.MessageMarking.After = testcase.markAfter
+			receiverConfig.MessageMarking.OnError = testcase.markErrors
+			set, tel, observedLogs := mustNewSettings(t)
+			f := NewFactory()
+			r, err := f.CreateTraces(context.Background(), set, receiverConfig, consumer)
+			require.NoError(t, err)
+			require.NoError(t, r.Start(context.Background(), componenttest.NewNopHost()))
+			t.Cleanup(func() {
+				assert.NoError(t, r.Shutdown(context.Background()))
+			})
+
+			if testcase.errorShouldRestart {
+				// Verify that the consumer restarts at least once.
+				assert.Eventually(t, func() bool {
+					m, err := tel.GetMetric("otelcol_kafka_receiver_partition_start")
+					require.NoError(t, err)
+
+					dataPoints := m.Data.(metricdata.Sum[int64]).DataPoints
+					assert.Len(t, dataPoints, 1)
+					return dataPoints[0].Value > 5
+				}, time.Second, 100*time.Millisecond, "unmarshal error should restart consumer")
+
+				// The invalid message should block the consumer.
+				assert.Zero(t, calls.Load())
+
+				observedErrorLogs := observedLogs.FilterLevelExact(zapcore.ErrorLevel)
+				logEntries := observedErrorLogs.All()
+				require.NotEmpty(t, logEntries)
+				for _, entry := range logEntries {
+					assert.Equal(t, "failed to unmarshal message", entry.Message)
+				}
+			} else {
+				assert.Eventually(t, func() bool {
+					return calls.Load() == 1
+				}, time.Second, 100*time.Millisecond, "unmarshal error should not block consumption")
+
+				// Verify that the consumer did not restart.
+				metadatatest.AssertEqualKafkaReceiverPartitionStart(t, tel, []metricdata.DataPoint[int64]{{
+					Value:      1,
+					Attributes: attribute.NewSet(attribute.String("name", set.ID.Name())),
+				}}, metricdatatest.IgnoreTimestamp())
+
+				observedErrorLogs := observedLogs.FilterLevelExact(zapcore.ErrorLevel)
+				logEntries := observedErrorLogs.All()
+				require.Len(t, logEntries, 2)
+				assert.Equal(t, "failed to unmarshal message", logEntries[0].Message)
+				assert.Equal(t,
+					"failed to consume message, skipping due to message_marking config",
+					logEntries[1].Message,
+				)
+			}
+		})
+	}
+}
+
 func TestNewLogsReceiver(t *testing.T) {
 	t.Parallel()
 	kafkaClient, receiverConfig := mustNewFakeCluster(t, kfake.SeedTopics(1, "otlp_logs"))