writer.WriteMessages(...) method generates "Denied Operation = Describe" events on non-related topics

[chinniehendrix]

Describe the bug

When producing keyed messages to a specific topic using the writer.WriteMessages(...) method with mutual TLS and ACLs enabled, the client tries to access other topics (for which it does not have permissions) generating lots of "Denied Operation = Describe" events. This behavior, when shared simultaneously by hundreds of clients, can have a significant impact on the performance of the Kafka cluster.

Kafka Version

• What version(s) of Kafka are you testing against?
  v2.4.1

• What version of kafka-go are you using?
  v0.4.33

To Reproduce

All resources can be found in this public repository: https://github.com/chinniehendrix/local-kafka-poc/tree/feat/demo-kafka-go-describe-issue/strimzi/tls-clients

package main

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"fmt"
	"io/ioutil"
	"time"

	log "github.com/sirupsen/logrus"

	"github.com/segmentio/kafka-go"
)

func tlsConfig() *tls.Config {
	userCert, errCert := ioutil.ReadFile("/client-certs/user.crt")
	if errCert != nil {
		log.Fatal(errCert)
	}

	userKey, errKey := ioutil.ReadFile("/client-certs/user.key")
	if errKey != nil {
		log.Fatal(errKey)
	}

	caPEM, errCA := ioutil.ReadFile("/ca-certs/ca.crt")
	if errCA != nil {
		log.Fatal(errCA)
	}

	cert, err := tls.X509KeyPair(userCert, userKey)
	if err != nil {
		log.Fatal(err.Error())
	}

	caCertPool := x509.NewCertPool()
	caCertPool.AppendCertsFromPEM([]byte(caPEM))

	config := &tls.Config{
		MinVersion:   tls.VersionTLS12,
		Certificates: []tls.Certificate{cert},
		RootCAs:      caCertPool,
	}

	fmt.Println("Returning TLS config")
	return config
}

func main() {
	now := time.Now()
	fmt.Println("Go client started")
	fmt.Println(now)
	logger := log.New()

	w := &kafka.Writer{
		Addr:        kafka.TCP("my-cluster-kafka-brokers:9093"),
		Topic:       "my-topic",
		Compression: kafka.Lz4,
		Transport: &kafka.Transport{
			TLS: tlsConfig(),
		},
		//Balancer: &kafka.Hash{},
		Logger: logger,
	}

	for i := 0; i < 1; i++ {
		key := fmt.Sprintf("Key-%d", i)

		kafkaMessage := &kafka.Message{
			Key:   []byte(key),
			Value: []byte(fmt.Sprintf("This is message %d", i)),
		}

		err := w.WriteMessages(context.Background(), *kafkaMessage)
		if err != nil {
			fmt.Println(err)
		} else {
			fmt.Println("produced", key)
		}

	}

	if err := w.Close(); err != nil {
		log.Fatal("failed to close writer:", err)
	} else {
		log.Info("Successfully closed writer")
	}

	time.Sleep(3600 * time.Second)
}

Expected Behavior

The client should produce one message to topic "my-topic" and close the connection.

Observed Behavior

The client produces one message to topic "my-topic" and tries to access topic "my-topic-02" to which it does not have permissions, repeatedly causing "Denied Operation = Describe" events on the Kafka cluster (as shown below).

...
my-cluster-kafka-0 kafka 2022-08-02 13:10:18,014 INFO Principal = User:CN=test-producer-01 is Denied Operation = Describe from host = 172.17.0.8 on resource = Topic:LITERAL:__consumer_offsets (kafka.authorizer.logger) [data-plane-kafka-request-handler-7]
my-cluster-kafka-0 kafka 2022-08-02 13:10:18,015 INFO Principal = User:CN=test-producer-01 is Denied Operation = Describe from host = 172.17.0.8 on resource = Topic:LITERAL:my-topic-02 (kafka.authorizer.logger) [data-plane-kafka-request-handler-7]
my-cluster-kafka-0 kafka 2022-08-02 13:10:18,715 INFO Principal = User:CN=test-producer-01 is Denied Operation = Describe from host = 172.17.0.8 on resource = Topic:LITERAL:__consumer_offsets (kafka.authorizer.logger) [data-plane-kafka-request-handler-6]
my-cluster-kafka-0 kafka 2022-08-02 13:10:18,715 INFO Principal = User:CN=test-producer-01 is Denied Operation = Describe from host = 172.17.0.8 on resource = Topic:LITERAL:my-topic-02 (kafka.authorizer.logger) [data-plane-kafka-request-handler-6]
my-cluster-kafka-0 kafka 2022-08-02 13:10:21,331 INFO Principal = User:CN=test-producer-01 is Denied Operation = Describe from host = 172.17.0.8 on resource = Topic:LITERAL:__consumer_offsets (kafka.authorizer.logger) [data-plane-kafka-request-handler-4]
my-cluster-kafka-0 kafka 2022-08-02 13:10:21,331 INFO Principal = User:CN=test-producer-01 is Denied Operation = Describe from host = 172.17.0.8 on resource = Topic:LITERAL:my-topic-02 (kafka.authorizer.logger) [data-plane-kafka-request-handler-4]
my-cluster-kafka-0 kafka 2022-08-02 13:10:25,922 INFO Principal = User:CN=test-producer-01 is Denied Operation = Describe from host = 172.17.0.8 on resource = Topic:LITERAL:__consumer_offsets (kafka.authorizer.logger) [data-plane-kafka-request-handler-6]
my-cluster-kafka-0 kafka 2022-08-02 13:10:25,922 INFO Principal = User:CN=test-producer-01 is Denied Operation = Describe from host = 172.17.0.8 on resource = Topic:LITERAL:my-topic-02 (kafka.authorizer.logger) [data-plane-kafka-request-handler-6]

Additional Context

The described behavior does NOT occur when using kafka.Conn.WriteMessages(...) with a Conn obtained via the DialLeader API.

	conn, err := dialer.DialLeader(context.Background(), "tcp", "my-cluster-kafka-brokers:9093", topic, partition)
	if err != nil {
		log.Fatal("failed to dial leader:", err)
	}

	conn.SetWriteDeadline(time.Now().Add(10 * time.Second))

	for i := 0; i < 1000; i++ {
		key := fmt.Sprintf("Key-%d", i)

		kafkaMessage := &kafka.Message{
			Key:   []byte(key),
			Value: []byte(fmt.Sprintf("This is message %d", i)),
		}

		_, err := conn.WriteMessages(*kafkaMessage)

		if err != nil {
			fmt.Println(err)
		} else {
			fmt.Println("produced", key)
		}

	}

	if err := conn.Close(); err != nil {
		log.Fatal("failed to close writer:", err)
	}

Also, this behavior was not (yet) reproduced on any other version of Kafka than v2.4.1

[achille-roussel]

Hello @chinniehendrix, thanks for reporting the issue!

I believe this is due to the use of Metadata requests without topic list for discovery in the kafka.Transport here.

To address this issue, we should lazily detect the list of topics that the transport has been used with (either from Produce of Fetch requests) and watch these topics alone instead of all topics.

[AlexeiBlue]

Hi,

We get the same issue, it first came to our attention when the team that looks after our Kafka clusters reported logs filling up with Operation Denied requests on all topics from our apps ACL.

I haven't had much time to debug the issue but I suspect @achille-roussel is correct.

[AlexeiBlue]

Having manually set the topic name in the meta-request I can confirm that @achille-roussel is correct. It's a rubbish fix that allows me to set that topic name via the context for now which isn't brilliant, obviously it would be great to get the configured topic name for the writer but I'm not sure of all the underlaying structure so probably not the best person to raise a PR for this.

[achille-roussel]

The kafka.Transport object might be shared by multiple clients, writers, and readers, which is why it would be ideal to lazily discover the list of topics that are in use by the application and perform metadata requests for those only.

An alternative approach would be to configure a dedicated transport on the writer, and add configuration to the kafka.Transport type to explicitly list the topics that it should perform metadata requests for. This would obviously be less automation, it's my belief that the other approach I described above would create a better user experience.

Let me know what you think!

[paulo]

hey @achille-roussel! I'm seeing the same behavior from our kafka clients, so I'm hoping this issue is not too stale.

An alternative approach would be to configure a dedicated transport on the writer and add configuration to the kafka.Transport type to explicitly list the topics that it should perform metadata requests for. This would obviously be less automation, it's my belief that the other approach I described above would create a better user experience.

Sounds like maybe this would be the more straightforward solution, and it allows for some finetuning from people using the lib? I think both of your approaches sound good, but maybe they can co-exist.

Is there any chance this will be worked on in the near future? I don't mind submitting a PR if not 😄

[mancusi]

We're also running into this issue

[mhmtszr]

@achille-roussel #1131 I tried to implement an alternative approach that you mentioned could you review it?

[rhansen2]

Fixed via #1131