Handle authorization errors in Kafka output (#42401) (#42411)

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)

Co-authored-by: Tiago Queiroz <[email protected]>

CHANGELOG.next.asciidoc

@@ -104,6 +104,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Allow network condition to handle field values that are arrays of IP addresses. {pull}41918[41918]
 - Fix a bug where log files are rotated on startup when interval is configured and rotateonstartup is disabled {issue}41894[41894] {pull}41895[41895]
 - Fix setting unique registry for non beat receivers {issue}42288[42288] {pull}42292[42292]
+- The Kafka output now drops events when there is an authorisation error {issue}42343[42343] {pull}42401[42401]
 
 *Auditbeat*
 

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)
+	}
+}