admin管理员组

文章数量:1323729

KafkaError{code=_MSG_TIMED_OUT,val=-192,str="Local: Message timed out"}


Receuvung the errir when producing a message to a kafka topic from airflow task (python operator).

When producing the same message to the same topic from a simple python producer application (outside airflow) it works fine. But when producing from the airflow task it fails.

The airflow cluster successfuly listens to the Kafka brokers. Topic authentication and authorization is done.

from airflow import DAG
from airflow.operators.python import PythonOperator
from confluent_kafka import Producer

def produce_to_kafka(**context):
        # Kafka configuration
        KAFKA_CONFIG = {
            'bootstrap.servers': 'hostname:port',
            'security.protocol': 'ssl',
            "partitioner": "random",
            'ssl.ca.location': 'path.pem',
            'ssl.certificate.location':'path.cer', 
            'ssl.key.location':'path.key',
        }
        KAFKA_TOPIC = 'my_topic'
        
        producer = Producer(KAFKA_CONFIG)
        
        producer.produce(KAFKA_TOPIC, key='my_key', value='my_message', partition=1, callback=handle_error)
        producer.flush()


with DAG(
    dag_id='my_dag',
    schedule_interval='0 0 * * *',
    start_date=datetime(2025, 1, 1),
    catchup=False,
) as dag:

   

    
    send_to_kafka = PythonOperator(
        task_id='send_messages_to_kafka',
        python_callable=produce_to_kafka,
        provide_context=True,
    )

    send_to_kafka

I tried to create a simple producer python application outside of airflow with the same producer configs and it works fine.

本文标签: