fix: handle message too large error on kafka client and emit metric

Author: siddhanta-rath

publisher/kafka.go

@@ -63,10 +63,17 @@ func (pr *Kafka) ProduceBulk(events []*pb.Event, connGroup string, deliveryChann
 		err := pr.kp.Produce(message, deliveryChannel)
 		if err != nil {
 			metrics.Increment("kafka_messages_delivered_total", fmt.Sprintf("success=false,conn_group=%s,event_type=%s", connGroup, event.Type))
-			if err.Error() == "Local: Unknown topic" {
+			switch err.Error() {
+			case "Local: Unknown topic":
 				errors[order] = fmt.Errorf("%v %s", err, topic)
-				metrics.Increment("kafka_unknown_topic_failure_total", fmt.Sprintf("topic=%s,event_type=%s,conn_group=%s", topic, event.Type, connGroup))
-			} else {
+				metrics.Increment("kafka_unknown_topic_failure_total",
+					fmt.Sprintf("topic=%s,event_type=%s,conn_group=%s", topic, event.Type, connGroup))
+
+			case "Local: Message size too large":
+				errors[order] = fmt.Errorf("%v %s", err, topic)
+				metrics.Increment("kafka_message_too_large_total",
+					fmt.Sprintf("topic=%s,event_type=%s,conn_group=%s", topic, event.Type, connGroup))
+			default:
 				errors[order] = err
 			}
 			continue

publisher/kafka_test.go

@@ -96,6 +96,15 @@ func TestKafka_ProduceBulk(suite *testing.T) {
 			err := kp.ProduceBulk([]*pb.Event{{EventBytes: []byte{}, Type: topic}}, "group1", make(chan kafka.Event, 2))
 			assert.EqualError(t, err.(BulkError).Errors[0], "Local: Unknown topic "+topic)
 		})
+
+		t.Run("Should return topic name when message size large is found", func(t *testing.T) {
+			client := &mockClient{}
+			client.On("Produce", mock.Anything, mock.Anything).Return(fmt.Errorf("Local: Message size too large")).Once()
+			kp := NewKafkaFromClient(client, 10, "%s")
+
+			err := kp.ProduceBulk([]*pb.Event{{EventBytes: []byte{}, Type: topic}}, "group1", make(chan kafka.Event, 2))
+			assert.EqualError(t, err.(BulkError).Errors[0], "Local: Message size too large "+topic)
+		})
 	})
 
 	suite.Run("MessageFailedToProduce", func(t *testing.T) {