xBis7 commented on code in PR #73276:
URL: https://github.com/apache/airflow/pull/73276#discussion_r4270767629
##########
providers/apache/kafka/tests/integration/apache/kafka/plugins/test_event_producer.py:
##########
Review Comment:
Since now messages are coming from different topics, it would make sense to
have an extra assertion, that if the message starts with `dag_run.`, then
`message.topic()` is `DAG_RUN_TOPIC` and similarly if the message starts with
`task_instance.`, then `message.topic()` is `TASK_INSTANCE_TOPIC`.
##########
providers/apache/kafka/tests/integration/apache/kafka/plugins/test_event_producer.py:
##########
@@ -107,12 +109,12 @@ def setup_class(cls):
cls._admin = KafkaAdminClientHook(kafka_config_id=cls.KAFKA_CONFIG_ID)
# Tuple positions are (name, partition, replication).
- cls._admin.create_topic([(cls.TOPIC, 1, 1)])
+ cls._admin.create_topic([(cls.DAGRUN_TOPIC, 1, 1),
(cls.TASK_INSTANCE_TOPIC, 1, 1)])
@classmethod
def teardown_class(cls):
try:
- cls._admin.delete_topic([cls.TOPIC])
+ cls._admin.delete_topic([cls.DAGRUN_TOPIC,
cls.TASK_INSTANCE_TOPIC])
time.sleep(2) # let the broker finish the async delete
except Exception as exc:
log.warning("teardown: failed to delete topic %r: %s", cls.TOPIC,
exc)
Review Comment:
This is still referencing `cls.TOPIC`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
