fix : error handling on message too large

Author: siddhanta-rath

publisher/kafka.go

@@ -63,19 +63,19 @@ 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))
-			logger.Errorf("produce to kafka failed : %v and string value : %s", err, err.Error())
 			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))
 
-			case "Local: Message size too large":
+			case "Broker: 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
+				logger.Errorf("produce to kafka failed due to: %v", err)
 			}
 			continue
 		}

publisher/kafka_test.go

@@ -97,13 +97,13 @@ func TestKafka_ProduceBulk(suite *testing.T) {
 			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) {
+		t.Run("Should return topic name when message size is too large", func(t *testing.T) {
 			client := &mockClient{}
-			client.On("Produce", mock.Anything, mock.Anything).Return(fmt.Errorf("Local: Message size too large")).Once()
+			client.On("Produce", mock.Anything, mock.Anything).Return(fmt.Errorf("Broker: 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)
+			assert.EqualError(t, err.(BulkError).Errors[0], "Broker: Message size too large "+topic)
 		})
 	})