인프런 커뮤니티 질문&답변

안녕하세요 개발자님

KafkaTimeoutError: Failed to update metadata after 60.0 secs. 2025-03-21 21:28:43.328 | ERROR | kafka_communication:send_message:52 - KafkaTimeoutError: Failed to update metadata after 60.0 secs.

파이썬으로 작성된 코드를 받게 되어 개발을 진행 중인데 window 환경에서 ai프로그램을 돌려 린눅스 서버에 cnotainer 컨프런트 kafka 로 결과 값을 전송 하게 되어씁니다.

window 에서 ai프로그램이 결과 값을 리눅스 docker contianer로 전송 하는데 오래 걸리지만, kafka 로 전송이 되지만.,

window에 있던 ai 프고르램을 다른 리눅스 서버 2 번째 환경에서 실행시키고, 첫번째 kafka가 동작 중인 kafka에 결과 값을 전송 하면

KafkaTimeoutError: Failed to update metadata after 60.0 secs.

2025-03-21 21:28:43.328 | ERROR | kafka_communication:send_message:52 - KafkaTimeoutError: Failed to update metadata after 60.0 secs.

error 가 발생 합니다.

window에서는 안발생 했는데, 말이죠.,

ai 프로그램을 강제 종료 하고 error를 보면

^CTraceback (most recent call last):

File "/home/gpu-1/projects/argos/source/ai/ai-main/message_event_handler.py", line 174, in <module>

message_event_handler.handle_messages()

File "/home/gpu-1/projects/argos/source/ai/ai-main/message_event_handler.py", line 80, in handle_messages

msg = self.consumer.receive_message()

^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

File "/home/gpu-1/projects/argos/source/ai/ai-main/kafka_communication.py", line 83, in receive_message

message = next(self.consumer).value

^^^^^^^^^^^^^^^^^^^

File "/opt/miniconda3/envs/argos-ai-volov8/lib/python3.11/site-packages/kafka/consumer/group.py", line 1168, in next

return next(self._iterator)

^^^^^^^^^^^^^^^^^^^^

File "/opt/miniconda3/envs/argos-ai-volov8/lib/python3.11/site-packages/kafka/consumer/group.py", line 1140, in messagegenerator_v2

record_map = self.poll(timeout_ms=timeout_ms, update_offsets=False)

^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

File "/opt/miniconda3/envs/argos-ai-volov8/lib/python3.11/site-packages/kafka/consumer/group.py", line 679, in poll

records = self._poll_once(inner_timeout_ms(), max_records, update_offsets=update_offsets)

^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

File "/opt/miniconda3/envs/argos-ai-volov8/lib/python3.11/site-packages/kafka/consumer/group.py", line 722, in pollonce

self._client.poll(timeout_ms=inner_timeout_ms(self._coordinator.time_to_next_poll() * 1000))

File "/opt/miniconda3/envs/argos-ai-volov8/lib/python3.11/site-packages/kafka/client_async.py", line 683, in poll

self._poll(timeout / 1000)

File "/opt/miniconda3/envs/argos-ai-volov8/lib/python3.11/site-packages/kafka/client_async.py", line 742, in _poll

if conn.send_pending_requests_v2():

^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

File "/opt/miniconda3/envs/argos-ai-volov8/lib/python3.11/site-packages/kafka/conn.py", line 1039, in send_pending_requests_v2

total_bytes = self._send_bytes(self._send_buffer)

^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

File "/opt/miniconda3/envs/argos-ai-volov8/lib/python3.11/site-packages/kafka/conn.py", line 683, in sendbytes

sent_bytes = self._sock.send(data[total_sent:])

^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

이렇게 나오고 producer 설정은

class MessageProducer:

def init(self, broker):

self.broker = broker

# self.topic = topic

self.producer = self.init()

def init(self):

while True:

try:

producer = KafkaProducer(

bootstrap_servers=self.broker,

value_serializer=lambda x: json.dumps(x).encode('utf-8'),

acks=0,

request_timeout_ms=1200000,

retry_backoff_ms=3000,

batch_size=16384, #

metadata_max_age_ms=3000000,

api_version=(2, 5, 0),

retries=3

)

break

except Exception as e:

print(e)

logger.error(e)

logger.error("Retry connecting to broker...")

sleep(2)

logger.info("Connected to kafka broker !")

return producer

`

def send_message(self, topic, msg):

try:

print(f'topic: {topic}')

future = self.producer.send(topic, value=msg)

self.producer.flush()

future.get(timeout=None)

return {'status_code': 200, 'error': None}

except Exception as e:

print(e)

logger.error(e)

return e

입니다.

window에서 리눅스로 옴겼더니 왜 그런걸까요?
설정 부분 추가 및 변경 해야 하는 부분이 어디 일까요?

client에서 설정을 수정 해야 하나여ㅛ, kafka server.properties 에서 어디를 수정 해야 하나요?

producer로 kafak로 부토ㅓ matadata를 기다려야 하나요?