Consumer/channel does not reauthenticate based on SaslAuthenticateResponse
- 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
Evaluación
Este issue todavía no se ha evaluado.