ReadPartitions returns empty list on missing permissions instead of an error

[moolitayer]

Describe The Bug ReadPartitions returns an empty list when ACL permissions are missing to list topics (topic read). It takes a long time to figure out the reason for the empty list (e.g could be related to https://github.com/segmentio/kafka-go/issues/381)

Kafka Version

• What version(s) of Kafka are you testing against?
• What version of kafka-go are you using?

v0.4.35

To Reproduce

Resources to reproduce the behavior:

package main

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

func main() {
  d := kafka.NewDialer(params)
  conn, err := d.Dial("tcp", params.Host)
  if err != nil {
      logger.Fatal("dialer couldn't connect: ", err)
  }
  controllerConn, err := d.Dial("tcp", fmt.Sprintf("%s:%d", controller.Host, controller.Port))
  defer controllerConn.Close()
  partitions, err := controllerConn.ReadPartitions()	
}

partitions is an empty list.

After applying the following ACL using terraform the list is non empty:

resource "kafka_acl" "topic_read" {
  resource_name                = "*"
  resource_type                = "Topic"
  resource_pattern_type_filter = "Literal"
  acl_principal                = format("User:%d", "id key")
  acl_host                     = "*"
  acl_operation                = "Read"
  acl_permission_type          = "Allow"
}

Expected Behavior

An error is returned.

Observed Behavior

An empty list is returned.

[achille-roussel]

Hello @moolitayer, thanks for reporting the issue! Would you have spare cycles to submit a fix for this?

[moolitayer]

Hi @achille-roussel. I affraid not, any time soon, sorry.

[achille-roussel]

Relabeling since this is not technically a bug but rather we could improve the library's behavior.

A contribution will still be welcome!

[haochunchang]

Hi, I would like to contribute on this improvement.

To reproduce the behavior, one can modify the following line in docker-compose.yml to:

KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: 'false'

Then use the following main.go to see the number of partitions:

package main

import (
	"fmt"
	kafka "github.com/segmentio/kafka-go"
)

func main() {
	controllerConn, err := kafka.Dial("tcp", "localhost:9092")
	if err != nil {
		panic(err)
	}
	defer controllerConn.Close()

	partitions, err := controllerConn.ReadPartitions()
	if err != nil {
		panic(err)
	}
	fmt.Println("Number of partitions: ", len(partitions))
}

---

Current observations

• When KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: 'false'

  • After readResponse in the conn.go, the resulting metadataResponseV6 does not have any brokers and topics. And its ControllerID attribute is -1.
  • Response size: 40

• When KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: 'true'

  • The resulting metadataResponseV6 have brokers and topics information while ControllerID attribute is 1.
  • Response size: 285

Questions

• Can we use ControllerID to check if readPartitionResponse should return error for missing ACL?
• What error message should we use to indicate this kind of error?

Any suggestions or comments would be very appreciated!

[haochunchang]

Another question I would like to clarify: When ACL permissions are missing to list topics, usually means the current client has no explicit permission to list the topics, right? If so, I think it is reasonable to get an empty response for the client (since it does not have proper permission). But I am new to kafka, maybe I miss many things.