# Consuming with Airflow ConsumeFromTopicOperator

**URL:** <https://forum.confluent.io/t/consuming-with-airflow-consumefromtopicoperator/10858>\
**Category:** Non-Java Clients\
**Created:** [11 June 2024 17:58 UTC](https://forum.confluent.io/t/consuming-with-airflow-consumefromtopicoperator/10858 "2024-06-11T17:58:33Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![mattmasuda](https://avatars.discourse-cdn.com/v4/letter/m/e47774/32.png) [@mattmasuda](https://forum.confluent.io/u/mattmasuda)\
**Post date:** [11 June 2024 17:58 UTC](https://forum.confluent.io/t/consuming-with-airflow-consumefromtopicoperator/10858/1 "2024-06-11T17:58:33Z")

</div>

Has anyone had success using Apache Airflow’s ConsumeFromTopicOperator to consume messages from a Confluent Cloud cluster? We are trying to get a basic consumer running in our Airflow environment but we keep running into this error: ‘Subscribed topic not available: orders: Broker: Topic authorization failed’.

Our Kafka config looks like this:

```auto
Connection(
    conn_id="kafka_conn_id",
	conn_type="Kafka",
	extra= json.dumps({
		"bootstrap.servers":<BOOTSTRAP_ADDRESS>,
		"security.protocol":"SASL_SSL",
		"sasl.mechanisms":"PLAIN",
		"sasl.username":<API_KEY>,
		"sasl.password":<API_SECRET>,
		"group.id":"my-group",
		"auto.offset.reset":"earliest",
		"enable.auto.commit":False,
	})
)

```

We have double checked our topic ACLs, tried with multiple topics, and tried using an admin’s API key which should have unrestricted access. All of those attempts resulted in the same topic authorization error. Any other ideas would be greatly appreciated.

---

<div class="post-metadata">

**Author:** ![cmeans](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/cmeans/32/3963_2.png) [@cmeans](https://forum.confluent.io/u/cmeans)\
**Post date:** [24 August 2024 02:10 UTC](https://forum.confluent.io/t/consuming-with-airflow-consumefromtopicoperator/10858/2 "2024-08-24T02:10:44Z")

</div>

I’ve just been working on a local PoC, but I think the correct “auto.offset.reset” value you want is “beginning”, not “earliest” (I too was using that and it wasn’t working as desired).

For your particular error though, I would guess you’ve not created the allow READ ACL for the topic “orders” and/or the group name “my-group”.