aio-libs / aio-libs/aiokafka

Consumer/channel does not reauthenticate based on SaslAuthenticateResponse

Abierto
#1,080 1 comentario 4 reacciones 0 asignados Ver en GitHub
Lenguaje dominante
Python
Estrellas
1.4k
Forks
269
Merge medio
1 d 1 h
PR fusionados (30 d)
6

Descripción

**Describe the bug**
When a consumer's connection outlives its SASL authentication (in this case an OAuth token), then consumer will fail to fetch new records because the broker no longer accepts any requests other than `SaslHandshakeRequest` and `SaslAuthenticateRequest`. See [KIP-368](https://cwiki.apache.org/confluence/display/KAFKA/KIP-368%3A+Allow+SASL+Connections+to+Periodically+Re-Authenticate) for more details.

As an example, if tokens have a valid duration of 5 minutes, the consumer will run fine for 5 minutes after which it will begin failing when it sends new requests. For example, this `HeartbeatRequest` for the group coordinator fails because the channel's session is expired:
```
ERROR:aiokafka.consumer.group_coordinator:Error sending HeartbeatRequest_v1 to node 2 [KafkaConnectionError: Connection at my.redacted-kafka.broker closed] -- marking coordinator dead
```

I suspect this is also an issue for admin and producer clients.

**Expected behaviour**
I expect the consumer to reauthenticate before its session expires.

**Environment (please complete the following information):**
- aiokafka version (`python -c "import aiokafka; print(aiokafka.__version__)"`): 0.12.0
- Kafka Broker version (`kafka-topics.sh --version`): 3.7.0
- Other information (Confluent Cloud version, etc.): Self-hosted Kafka with Strimzi OAuth authentication plugin

**Reproducible example**

The problem can be reproduced with a simple consumer such as this together with a Kafka broker configured to use secured OAuth SASL authentication for example using Strimzi. If it could be of value, I will try to create a Dockerfile to setup such a broker.

```python
import asyncio
from aiokafka import AIOKafkaConsumer
from aiokafka.abc import AbstractTokenProvider
from azure.identity import DefaultAzureCredential

class KafkaOAuthBearerTokenProvider(AbstractTokenProvider):
async def token(self):
# logic for fetching token
...
async def consume():
# Kafka consumer configuration with SASL authentication
consumer = AIOKafkaConsumer(
'sometopic',
bootstrap_servers='my.redacted-kafka.broker,my.redacted-kafka.broker2,my.redacted-kafka.broker3',
auto_offset_reset='earliest',
security_protocol='SASL_PLAINTEXT',
sasl_mechanism='OAUTHBEARER',
sasl_oauth_token_provider= KafkaOAuthBearerTokenProvider(),
group_id= "someconsumergroup"
)
# Start the consumer
await consumer.start()
try:
# Consume messages
async for msg in consumer:
print(f"Consumed message: {msg.value.decode('utf-8')} from partition: {msg.partition} and offset: {msg.offset} and header: {msg.headers}")
finally:
# Ensure the consumer is stopped gracefully
await consumer.stop()
```

Guía de contribución

Abrir la guía de contribución

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.