chore: sampling errors if no message in the batch succeeds

services/streammanager/kafka/kafkamanager.go

@@ -556,7 +556,16 @@
 ) {
 	start := now()
 	defer func() { kafkaStats.prepareBatchTime.SendTiming(since(start)) }()
+
 	var messages []client.Message
+	var errorSamples []error
+	addErrorSample := func(msg string, args ...any) {
+		err := fmt.Errorf(msg, args...)
+		if len(errorSamples) < 3 {
+			errorSamples = append(errorSamples, err)
+		}
+		pkgLogger.Error(err.Error())
+	}
 
 	for i, data := range batch {
 		topic, ok := data["topic"].(string)
@@ -567,39 +576,39 @@
 		message, ok := data["message"]
 		if !ok {
 			kafkaStats.missingMessage.Increment()
-			pkgLogger.Errorf("batch from topic %s is missing the message attribute", topic)
+			addErrorSample("batch from topic %s is missing the message attribute", topic)
 			continue
 		}
 		userID, ok := data["userId"].(string)
 		if !ok && !allowReqsWithoutUserIDAndAnonymousID.Load() {
 			kafkaStats.missingUserID.Increment()
-			pkgLogger.Errorf("batch from topic %s is missing the userId attribute", topic)
+			addErrorSample("batch from topic %s is missing the userId attribute", topic)
 			continue
 		}
 		marshalledMsg, err := json.Marshal(message)
 		if err != nil {
 			kafkaStats.jsonSerializationMsgErr.Increment()
-			pkgLogger.Errorf("unable to marshal message at index %d", i)
+			addErrorSample("unable to marshal message at index %d", i)
 			continue
 		}
 		codecs := p.getCodecs()
 		if len(codecs) > 0 {
 			schemaId, _ := data["schemaId"].(string)
 			if schemaId == "" {
 				kafkaStats.avroSerializationErr.Increment()
-				pkgLogger.Errorf("schemaId is not available for the event at index %d", i)
+				addErrorSample("schemaId is not available for the event at index %d", i)
 				continue
 			}
 			codec, ok := codecs[schemaId]
 			if !ok {
 				kafkaStats.avroSerializationErr.Increment()
-				pkgLogger.Errorf("unable to find schema with schemaId %q", schemaId)
+				addErrorSample("unable to find schema with schemaId %q", schemaId)
 				continue
 			}
 			marshalledMsg, err = serializeAvroMessage(schemaId, p.getEmbedAvroSchemaID(), marshalledMsg, *codec)
 			if err != nil {
 				kafkaStats.avroSerializationErr.Increment()
-				pkgLogger.Errorf(
+				addErrorSample(
 					"unable to serialize the event with schemaId %q at index %d: %s",
 					schemaId, i, err,
 				)
@@ -609,6 +618,9 @@
 		messages = append(messages, prepareMessage(topic, userID, marshalledMsg, timestamp))
 	}
 	if len(messages) == 0 {
+		if len(errorSamples) > 0 {
+			return nil, fmt.Errorf("unable to process any of the event in the batch: some errors are: %v", errorSamples)
+		}
 		return nil, fmt.Errorf("unable to process any of the event in the batch")
 	}
 	return messages, nil

services/streammanager/kafka/kafkamanager_test.go

@@ -536,7 +536,8 @@ func TestPrepareBatchOfMessages(t *testing.T) {
 		}}
 		batch, err := prepareBatchOfMessages(data, time.Now(), pm, "some-topic")
 		require.Equal(t, []client.Message(nil), batch)
-		require.Equal(t, fmt.Errorf("unable to process any of the event in the batch"), err)
+		require.Equal(t, fmt.Errorf("unable to process any of the event in the batch: "+
+			"some errors are: [batch from topic some-topic is missing the message attribute]"), err)
 	})
 
 	t.Run("with message and user id", func(t *testing.T) {
@@ -873,7 +874,8 @@ func TestSendBatchedMessage(t *testing.T) {
 		)
 		require.Equal(t, 400, sc)
 		require.Equal(t, "Failure", res)
-		require.Equal(t, "Error while preparing batched message: unable to process any of the event in the batch", err)
+		require.Equal(t, "Error while preparing batched message: unable to process any of the event in the batch: "+
+			"some errors are: [schemaId is not available for the event at index 0]", err)
 	})
 
 	t.Run("wrong codec", func(t *testing.T) {
@@ -903,7 +905,10 @@ func TestSendBatchedMessage(t *testing.T) {
 		)
 		require.Equal(t, 400, sc)
 		require.Equal(t, "Failure", res)
-		require.Equal(t, "Error while preparing batched message: unable to process any of the event in the batch", err)
+		require.Equal(t, "Error while preparing batched message: unable to process any of the event in the batch: "+
+			"some errors are: [unable to serialize the event with schemaId \"schemaId001\" at index 0: unable convert "+
+			"the event to native from textual, with error: cannot decode textual record \"kafkaAvroTest.myRecord\": "+
+			"cannot decode textual map: cannot determine codec: \"data\"]", err)
 	})
 
 	t.Run("unavailable schemaId", func(t *testing.T) {
@@ -933,7 +938,8 @@ func TestSendBatchedMessage(t *testing.T) {
 		)
 		require.Equal(t, 400, sc)
 		require.Equal(t, "Failure", res)
-		require.Equal(t, "Error while preparing batched message: unable to process any of the event in the batch", err)
+		require.Equal(t, "Error while preparing batched message: unable to process any of the event in the batch: "+
+			"some errors are: [unable to find schema with schemaId \"schemaId002\"]", err)
 	})
 }