I am using airflow to consume topics from kafka using ConsumeFromTopicOperator. This is all working well: Airflow connects to kafka and my producer function can read all the topics and print successfully process and print to the screen. From here I can see that the producer function can send the processed data away from Airflow, to a file or another service, for example. What I can't seem to do is send any data back to airflow. The producer function does not seem to be able to push to xcom and it's return value doesn't go anywhere. How would you tell airflow that the data has been processed successfully or trigger another task or dag from the producer function? Am I missing something or is that really how ConsumeFromTopicOperator is supposed to work? It seems the most appropriate operator for reading all the Kafka topics but if it only works one way then it's no good for the rest of the dag. Example Code def consume_function(message): print("consuming topics...") message_details = json.loads(message.value()) message_subject = message_details["subject"] print(f"message subject is: {message_subject}") # so far so good, subject prints to screen task_instance.xcom_push(key="message_subject", value=message_subject) # can't talk to xcom return message_subject # return value disappears @dag( start_date=datetime(2024, 1, 23), schedule_interval="* * * * *", # every minute catchup=False, ) def read_topics(): # consume the topic details consume_topics = ConsumeFromTopicOperator( task_id="consume_topics", kafka_config_id="kafka_default", topics=[KAFKA_TOPIC], apply_function=consume_function, poll_timeout=20, max_messages=20, max_batch_size=20, ) Summary Tried to send data from ConsumeFromTopicOperator's apply_function downstream to the rest of the dag, but can't see how to achieve this... Continue reading...
