<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.6.1</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">The AwaitMessageTriggerFunctionSensor is showing some buggy behaviour.<br>
When consuming from a topic, it is correctly applying the apply_function in order to yield a TriggerEvent.</p>
<p dir="auto">However, it is consuming multiple messages at a time and not yielding a trigger for the correct amount of messages that would be eligble (return a value in the apply_function). The observed behaviour is as follows:</p>
<ul dir="auto">
<li>Sensor is deferred and messages start getting consumed</li>
<li>Multiple eligble messages trigger a single TriggerEvent instead of multiple TriggerEvents.</li>
<li>The sensor returns to a deferred state , repeating the cycle.</li>
</ul>
<p dir="auto">The event_triggered_function is being called correctly. However, due to the issue in consuming and correctly generating the appropriate TriggerEvents some of them are missed.</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">Each eligble message should create an individual TriggerEvent to be consumed by the event_triggered_function.</p>
<h3 dir="auto">How to reproduce</h3>
<ul dir="auto">
<li>Use a producer DAG to produce a set amount of messages on your kafka topic</li>
<li>Use a listener DAG to consume this topic, screening for eligble messages (apply_function) and use the event_trigger_function to monitor the amount of events that are being consumed.</li>
</ul>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Kubernetes cluster - Linux</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">apache-airflow-providers-apache-kafka==1.1.0</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Official Apache Airflow Helm Chart</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">helm chart version 1.9.0</p>
<h3 dir="auto">Anything else</h3>
<p dir="auto">Every time (independent of topic, message content, apply_function and event_triggered_function)</p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.3.0 (latest released)</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">upgraded from 2.2.5</p>
<p dir="auto">airflow db upgrade showed a list of messages inline with what is in the changelog:</p>
<p dir="auto">Database configuration moved to new section (<a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="1169624280" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/22284" data-hovercard-type="pull_request" data-hovercard-url="/apache/airflow/pull/22284/hovercard" href="https://github.com/apache/airflow/pull/22284">#22284</a>)</p>
<p dir="auto">The following configurations have been moved from [core] to the new [database] section. However when reading the new option, the old option will be checked to see if it exists. If it does a DeprecationWarning will be issued and the old option will be used instead.</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="sql_alchemy_conn
sql_engine_encoding
sql_engine_collation_for_ids
sql_alchemy_pool_enabled
sql_alchemy_pool_size
sql_alchemy_max_overflow
sql_alchemy_pool_recycle
sql_alchemy_pool_pre_ping
sql_alchemy_schema
sql_alchemy_connect_args
load_default_connections
max_db_retries"><pre class="notranslate"><code class="notranslate">sql_alchemy_conn
sql_engine_encoding
sql_engine_collation_for_ids
sql_alchemy_pool_enabled
sql_alchemy_pool_size
sql_alchemy_max_overflow
sql_alchemy_pool_recycle
sql_alchemy_pool_pre_ping
sql_alchemy_schema
sql_alchemy_connect_args
load_default_connections
max_db_retries
</code></pre></div>
<p dir="auto">so nice as I am I moved those lines from [core] to a new [database] section and airflow db upgrade did not mention them again so all was well.</p>
<p dir="auto">After that I started the services</p>
<p dir="auto">sudo systemctl start airflow-webserver.service<br>
sudo systemctl start airflow-scheduler.service</p>
<p dir="auto">but the webservice failed with this error:</p>
<p dir="auto">May 04 08:46:21 postest airflow[108983]: [2022-05-04 08:46:21 +0200] [108983] [ERROR] Exception in worker process<br>
May 04 08:46:21 postest airflow[108983]: [2022-05-04 08:46:21,254] {configuration.py:494} WARNING - section/key [core/sql_alchemy_conn] not found in config<br>
May 04 08:46:21 postest airflow[108982]: [2022-05-04 08:46:21 +0200] [108982] [INFO] Worker exiting (pid: 108982)<br>
May 04 08:46:21 postest airflow[108982]: airflow.exceptions.AirflowConfigException: section/key [core/sql_alchemy_conn] not found in config</p>
<p dir="auto">for now I duplicated the lines, so they are both in [core] and [database], so no big issue.</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">service should start and find the settings in the [database] section</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">upgrade from 2.2.5, move the lines in question to a new [database] section and start the webservice.</p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Ubuntu 20.04.4 LTS (Focal Fossa)</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">apache-airflow-providers-ftp==2.1.2<br>
apache-airflow-providers-http==2.1.2<br>
apache-airflow-providers-imap==2.2.3<br>
apache-airflow-providers-postgres==4.1.0<br>
apache-airflow-providers-sqlite==2.1.3</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Other</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Anything else</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>