[8.17](backport #42401) Handle authorization errors in Kafka output (#42844)

* Handle authorization errors in Kafka output (#42401)

When there is an authorisation error in the Kafka output, the events
are dropped and an error message is logged.

(cherry picked from commit 5720300)

* Update CHANGELOG.next.asciidoc

* Fix changelog

---------

Co-authored-by: Tiago Queiroz <[email protected]>
Co-authored-by: Olga Naydyonock <[email protected]>
Co-authored-by: Khushi Jain <[email protected]>

Author: 4 people

CHANGELOG.next.asciidoc

@@ -77,6 +77,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Support Elastic Agent control protocol chunking support {pull}37343[37343]
 - Lower logging level to debug when attempting to configure beats with unknown fields from autodiscovered events/environments {pull}[37816][37816]
 - Set timeout of 1 minute for FQDN requests {pull}37756[37756]
+- The Kafka output now drops events when there is an authorization error. {issue}42343[42343] {pull}42401[42401]
 - 'add_cloud_metadata' processor - improve AWS provider HTTP client overriding to support custom certificate bundle handling {pull}44189[44189]
 - The Elasticsearch output now correctly applies exponential backoff when being throttled by 429s ("too many requests") from Elasticsarch. {issue}36926[36926] {pull}45073[45073]
 

libbeat/outputs/kafka/client.go

@@ -70,6 +70,20 @@ type msgRef struct {
 
 var (
 	errNoTopicsSelected = errors.New("no topic could be selected")
+
+	// authErrors are authentication/authorisation errors that will cause
+	// the event to be dropped
+	authErrors = []error{
+		sarama.ErrTopicAuthorizationFailed,
+		sarama.ErrGroupAuthorizationFailed,
+		sarama.ErrClusterAuthorizationFailed,
+		// I believe those are handled before the connection is
+		// stabilised, however we also handle them here just in
+		// case
+		sarama.ErrUnsupportedSASLMechanism,
+		sarama.ErrIllegalSASLState,
+		sarama.ErrSASLAuthenticationFailed,
+	}
 )
 
 func newKafkaClient(
@@ -368,6 +382,10 @@ func (r *msgRef) fail(msg *message, err error) {
 			len(msg.key)+len(msg.value))
 		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)
+
 	case errors.Is(err, breaker.ErrBreakerOpen):
 		// Add this message to the failed list, but don't overwrite r.err since
 		// all the breaker error means is "there were a lot of other errors".
@@ -425,3 +443,13 @@ func (c *client) Test(d testing.Driver) {
 	}
 
 }
+
+func isAuthError(err error) bool {
+	for _, e := range authErrors {
+		if errors.Is(err, e) {
+			return true
+		}
+	}
+
+	return false
+}

libbeat/tests/integration/kafka_test.go

@@ -87,3 +87,42 @@ func TestKafkaOutputCanConnectAndPublish(t *testing.T) {
 		10*time.Second,
 		"did not find finished batch log")
 }
+
+func TestAuthorisationErrors(t *testing.T) {
+	leader := sarama.NewMockBroker(t, 1)
+	defer leader.Close()
+
+	// The mock broker must respond to a single metadata request.
+	metadataResponse := new(sarama.MetadataResponse)
+	metadataResponse.AddBroker(leader.Addr(), leader.BrokerID())
+	metadataResponse.AddTopicPartition(kafkaTopic, 0, leader.BrokerID(), nil, nil, nil, sarama.ErrNoError)
+	leader.Returns(metadataResponse)
+
+	authErrors := []sarama.KError{
+		sarama.ErrTopicAuthorizationFailed,
+		sarama.ErrGroupAuthorizationFailed,
+		sarama.ErrClusterAuthorizationFailed,
+	}
+
+	// The mock broker must return one produce response per error we want
+	// to test. If less calls are made, the test will fail
+	for _, err := range authErrors {
+		producerResponse := new(sarama.ProduceResponse)
+		producerResponse.AddTopicPartition(kafkaTopic, 0, err)
+		leader.Returns(producerResponse)
+	}
+
+	// Start mockbeat with the appropriate configuration.
+	mockbeat := NewBeat(t, "mockbeat", "../../libbeat.test")
+	mockbeat.WriteConfigFile(fmt.Sprintf(kafkaCfg, kafkaTopic, kafkaVersion, leader.Addr()))
+	mockbeat.Start()
+
+	// Wait for mockbeat to log each of the errors.
+	for _, err := range authErrors {
+		t.Log("waiting for:", err)
+		mockbeat.WaitForLogs(
+			fmt.Sprintf("Kafka (topic=test_topic): authorisation error: %s", err),
+			10*time.Second,
+			"did not find error log: %s", err)
+	}
+}