[kafka] Handle configuration errors

libbeat/outputs/kafka/client.go

@@ -385,13 +385,15 @@ func (r *msgRef) fail(msg *message, err error) {
 	case errors.Is(err, sarama.ErrInvalidMessage):
 		r.client.log.Errorf("Kafka (topic=%v): dropping invalid message", msg.topic)
 		r.client.observer.PermanentErrors(1)
-
 	case errors.Is(err, sarama.ErrMessageSizeTooLarge) || errors.Is(err, sarama.ErrInvalidMessageSize):
 		r.client.log.Errorf("Kafka (topic=%v): dropping too large message of size %v.",
 			msg.topic,
 			len(msg.key)+len(msg.value))
 		r.client.observer.PermanentErrors(1)
-
+	// drop event if it exceeds size larger than max_message_bytes
+	case strings.Contains(err.Error(), "Attempt to produce message larger than configured Producer.MaxMessageBytes"):
+		r.client.log.Errorf("Kafka (topic=%v): dropping message as it exceeds max_mesage_bytes:", msg.topic)
+		r.client.observer.PermanentErrors(1)
 	case isAuthError(err):
 		r.client.log.Errorf("Kafka (topic=%v): authorisation error: %s", msg.topic, err)
 		r.client.observer.PermanentErrors(1)
@@ -404,7 +406,7 @@ func (r *msgRef) fail(msg *message, err error) {
 	default:
 		r.failed = append(r.failed, msg.data)
 		if r.err == nil {
-			// Don't overwrite an existing error. This way at tne end of the batch
+			// Don't overwrite an existing error. This way at the end of the batch
 			// we report the first error that we saw, rather than the last one.
 			r.err = err
 		}

libbeat/outputs/kafka/config.go

@@ -176,7 +176,7 @@
 
 	if c.Compression == "gzip" {
 		lvl := c.CompressionLevel
 		if lvl != sarama.CompressionLevelDefault && !(0 <= lvl && lvl <= 9) {
 			return fmt.Errorf("compression_level must be between 0 and 9")
 		}
 	}
@@ -185,6 +185,10 @@
 		return errors.New("either 'topic' or 'topics' must be defined")
 	}
 
+	if len(c.Headers) != 0 && c.Version < kafka.Version("0.11") {
+		return errors.New("including headers is not supported for kafka versions < 0.11")
+	}
+
 	// When running under Elastic-Agent we do not support dynamic topic
 	// selection, so `topics` is not supported and `topic` is treated as an
 	// plain string
@@ -238,7 +242,7 @@
 		k.Net.SASL.Enable = true
 		k.Net.SASL.Mechanism = sarama.SASLTypeGSSAPI
 		k.Net.SASL.GSSAPI = sarama.GSSAPIConfig{
 			AuthType:           int(config.Kerberos.AuthType),
 			KeyTabPath:         config.Kerberos.KeyTabPath,
 			KerberosConfigPath: config.Kerberos.ConfigPath,
 			ServiceName:        config.Kerberos.ServiceName,
@@ -266,12 +270,12 @@
 		k.Producer.MaxMessageBytes = *config.MaxMessageBytes
 	}
 	if config.RequiredACKs != nil {
 		k.Producer.RequiredAcks = sarama.RequiredAcks(*config.RequiredACKs)
 	}
 
 	compressionMode, ok := compressionModes[strings.ToLower(config.Compression)]
 	if !ok {
 		return nil, fmt.Errorf("Unknown compression mode: '%v'", config.Compression)
 	}
 	k.Producer.Compression = compressionMode
 
@@ -300,7 +304,7 @@
 
 	version, ok := config.Version.Get()
 	if !ok {
 		return nil, fmt.Errorf("Unknown/unsupported kafka version: %v", config.Version)
 	}
 	k.Version = version
 
@@ -332,7 +336,7 @@
 		// compute 'base' duration for exponential backoff
 		dur := cfg.Max
 		if retries < maxBackoffRetries {
 			dur = time.Duration(uint64(cfg.Init) * uint64(1<<retries))
 		}
 
 		// apply about equaly distributed jitter in second half of the interval, such that the wait