aio-libs / aio-libs/aiokafka

[QUESTION] Unable to request metadata from node with id 0: Unable to update metadata from [0]

Open
#624 15 comments 0 reactions 0 assignees View on GitHub
question
Dominant language
Python
Stars
1.4k
Forks
269
Avg merge
1d 1h
Merged PRs (30d)
6

Description

Hello, I have problem with producer. I am using
```aiokafka==0.6.0 and kafka-2.5.0 (Commit:66563e712b0b9f84)```

First I connect and send data to kafka in AWS server successfully. Then I wait 10 minutes and get logs:

```
[MainThread][ERROR][aiokafka ]: Unable to request metadata from node with id 0:
[MainThread][ERROR][aiokafka ]: Unable to update metadata from [0]

```
and trying sending (same) data again it fails with :
```
[MainThread][WARNING][aiokafka.producer.sender]: Got error produce response: [Error 7] RequestTimedOutError
```

My setup is using FastApi, aiokafka and gunicorn (workers=4):

```
...

app = FastAPI()

ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH, cadata=cdata)
loop = asyncio.get_event_loop()
producer = AIOKafkaProducer(loop=loop, bootstrap_servers=bootstrap_servers,
security_protocol="SSL",
ssl_context=ssl_context,
value_serializer=orjson.dumps,
)

@app.on_event("startup")
async def startup_event():
logger.info('Startup')
await producer.start()

@app.on_event("shutdown")
async def shutdown_event():
logger.info('Shutdown')
await producer.stop()
logger.info('Poducer Flushing ....')
# producer.flush()
pass

# inject producer object to each request
@app.middleware("http")
async def kafka_middleware(request: Request, call_next):
global producer
request.state.producer = producer
response = await call_next(request)
return response

@app.get("/")
async def home(request: Request):
data = {"Hello": "World", "time": time.time()}
producer = request.state.producer
f = await producer.send('my-topic', data )
return data
```
I could initialize/connect AIOKafkaProducer at each request but I thing it is not right way.

Kafka server.properties file
```
############################# Server Basics #############################
broker.id=0

listeners=SSL_INTERNAL://:9094,SSL_EXTERNAL://:9093
advertised.listeners=SSL_INTERNAL://:9094,SSL_EXTERNAL://ec2-ip-address.eu-central-1.compute.amazonaws.com:9093

listener.security.protocol.map=SSL_INTERNAL:SSL,SSL_EXTERNAL:SSL
inter.broker.listener.name=SSL_INTERNAL

ssl.keystore.location=/path_to/keystore/kafka.keystore.jks
ssl.keystore.password=securepass
ssl.key.password=securepass
ssl.truststore.location=/path_to/truststore/kafka.truststore.jks
ssl.truststore.password=securepass
ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1
ssl.keystore.type=JKS
ssl.truststore.type=JKS
ssl.endpoint.identification.algorithm=

############################# Log Basics #############################
log.dirs=/var/log/kafka-logs
num.partitions=4

num.recovery.threads.per.data.dir=1

offsets.topic.replication.factor=4
transaction.state.log.replication.factor=4
transaction.state.log.min.isr=4
```
All other kafka and Zookeeper configs are defaults.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.