Connection closed when fetching partition info using topic-level OAuth token

[vitali-kr]

Description

When using OAuth-based authentication with Receiver role granted at Event Hub (topic) level and then fetching METADATA without specifying any topics connection is forcefully closed

This happens when using apache kafka-clients for Java and calling KafkaConsumer.partitionsFor(topic). Kafka client fetches broker information first and then requests partition info for specific topic. If account is not granted Receiver access at the Event Hub Namespace level then broker closes connection after first request.

2021-06-28 22:36:29.190 DEBUG 46624 --- [ main] org.apache.kafka.clients.NetworkClient : [Consumer clientId=consumer-anonymous.45adc0de-4669-43e1-961d-4f58b5cf6638-1, groupId=anonymous.45adc0de-4669-43e1-961d-4f58b5cf6638] Sending METADATA request with header RequestHeader(apiKey=METADATA, apiVersion=5, clientId=consumer-anonymous.45adc0de-4669-43e1-961d-4f58b5cf6638-1, correlationId=3) and timeout 30000 to node -1: {topics=[],allow_auto_topic_creation=false} 2021-06-28 22:36:29.190 DEBUG 46624 --- [ main] org.apache.kafka.clients.NetworkClient : [Consumer clientId=consumer-anonymous.45adc0de-4669-43e1-961d-4f58b5cf6638-1, groupId=anonymous.45adc0de-4669-43e1-961d-4f58b5cf6638] Sending METADATA request with header RequestHeader(apiKey=METADATA, apiVersion=5, clientId=consumer-anonymous.45adc0de-4669-43e1-961d-4f58b5cf6638-1, correlationId=0) and timeout 30000 to node -1: {topics=[{name=kcp_picking_communication}],allow_auto_topic_creation=false} 2021-06-28 22:36:30.897 DEBUG 46624 --- [ main] o.apache.kafka.common.network.Selector : [Consumer clientId=consumer-anonymous.45adc0de-4669-43e1-961d-4f58b5cf6638-1, groupId=anonymous.45adc0de-4669-43e1-961d-4f58b5cf6638] Connection with krogerCustomer-dev.servicebus.windows.net/52.167.106.29 disconnected ... java.io.IOException: Connection reset by peer at java.base/sun.nio.ch.FileDispatcherImpl.read0(Native Method) ~[na:na] at java.base/sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:39) ~[na:na] at java.base/sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:276) ~[na:na] at java.base/sun.nio.ch.IOUtil.read(IOUtil.java:245) ~[na:na]

This issue propagates down to inability to use Spring Cloud Stream with default Kafka binder.

How to reproduce

Setup service principal and grant Receiver role to a specific topic. Use apache's kafka-clients (tested with 2.6.0 and 2.8.0) with OAuth bearer token authentication outlined at https://github.com/Azure/azure-event-hubs-for-kafka/tree/master/tutorials/oauth/java/appsecret Change consumer code to

final Consumer<Long, String> consumer = new KafkaConsumer<>(consumerProperties);
List<PartitionInfo> partitionInfo = consumer.partitionsFor(TOPIC, Duration.ofSeconds(5));

Has it worked previously?

No

Checklist

IMPORTANT: We will close issues where the checklist has not been completed or where adequate information has not been provided.

Please provide the relevant information for the following items:

• [x] SDK (include version info): org.apache.kafka:kafka-clients2.6.0
• [x] Sample you're having trouble with: tutorials/oauth/java/appsecret
• [x] If using Apache Kafka Java clients or a framework that uses Apache Kafka Java clients, version: 2.6.0
• [x] Kafka client configuration: same as in tutorials/oauth/java/appsecret/consumer (do not include your connection string or SAS Key)
• [x] Namespace and EventHub/topic name krogerCustomer-dev.servicebus.windows.net/kcp_picking_communication
• [x] Consumer or producer failure Consumer
• [x] Timestamps in UTC 2021-06-29 05:36:30.897 UTC
• [x] group.id or client.id client.id=consumer-anonymous.45adc0de-4669-43e1-961d-4f58b5cf6638-1
• [x] Logs provided (with debug-level logging enabled if possible, e.g. log4j.rootLogger=DEBUG) or exception call stack
• [ ] Standalone repro <REPLACE with e.g., Willing/able to send scenario to repro issue>
• [x] Operating system: any
• [ ] Critical issue

If this is a question on basic functionality, please verify the following:

• [x] Port 9093 should not be blocked by firewall ("broker cannot be found" errors)
• [x] Pinging FQDN should return cluster DNS resolution (e.g. $ ping namespace.servicebus.windows.net returns ~ ns-eh2-prod-am3-516.cloudapp.net [13.69.64.0])
• [x] Namespace should be either Standard or Dedicated tier, not Basic (TopicAuthorization errors)

[laurenmcdonald-kr]

Hello! I also did some testing with this scenario, both with SAS keys and an OAuth identity (service principal). I am only noticing this behavior with OAuth. If an identity does not have namespace access (i.e. just Azure Event Hubs Data Receiver on a specific topic), the consumer will time out or disconnect from the broker with the following commands:

partitionsFor(topic) listTopics()

According to the Kafka API docs, these functions should return the topics / partitions for which an identity is authorized. If the user is not authorized to view the specific topic in partitionsFor, it should return an unauthorized exception, not a timeout / disconnect.

[serkantkaraca]

We are aware of the issue. A fix is being rolled out globally this week and next week. Let me know if you still hit that after then.