You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "anxkhn (via GitHub)" <gi...@apache.org> on 2026/06/29 08:28:35 UTC
[PR] Stop BigQueryStreamingBufferEmptySensor reporting empty too early [airflow]
anxkhn opened a new pull request, #69118:
URL: https://github.com/apache/airflow/pull/69118
BigQuery's `streamingBuffer` table metadata is eventually consistent: for
several seconds after a streaming insert the rows are physically in the buffer
but the metadata still reads absent. `BigQueryStreamingBufferEmptySensor`
(added in #66652) decided "buffer empty" from a single absent reading, so during
that window it reported empty too early and a downstream DML task (UPDATE /
DELETE / MERGE) hit the very `... would affect rows in the streaming buffer`
error the sensor exists to prevent.
This makes "empty" trustworthy by requiring `empty_confirmations` consecutive
empty readings (default `2`), each one `poke_interval` apart, before the sensor
succeeds. A non-empty reading resets the counter. The same rule is applied in
both the sync `poke()` path and the deferrable `BigQueryStreamingBufferEmptyTrigger`.
Why consecutive confirmations instead of a non-empty -> empty transition check
(as discussed in the issue thread): a transition check polls until timeout when
the buffer flushes between two polls, and never fires for a table that is
genuinely empty (no prior streaming insert). A bounded consecutive-empty count
spans the consistency window while still terminating in both of those cases.
The default of `2` forces one full `poke_interval` to elapse between the two
empty readings, comfortably covering the ~10-12s metadata-lag window, while
remaining configurable (validated `>= 1`).
The obsolete metadata-wait workaround in the system test
(`example_bigquery_streaming_buffer_sensor.py`) is removed, as #66963 requested
once the sensor handles this itself.
closes: #66963
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: OpenCode (Claude Opus 4.8) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop BigQueryStreamingBufferEmptySensor reporting empty too early [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on PR #69118:
URL: https://github.com/apache/airflow/pull/69118#issuecomment-5270952692
> @shahar1 @SameerMesiah97 approved this after the branch was updated, and the checks are green. could you provide the final review or merge it?
@potiuk could you help with this?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop BigQueryStreamingBufferEmptySensor reporting empty too early [airflow]
Posted by "anxkhn (via GitHub)" <gi...@apache.org>.
anxkhn commented on PR #69118:
URL: https://github.com/apache/airflow/pull/69118#issuecomment-4998983132
hey @SameerMesiah97 , can you please re review?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop BigQueryStreamingBufferEmptySensor reporting empty too early [airflow]
Posted by "anxkhn (via GitHub)" <gi...@apache.org>.
anxkhn commented on PR #69118:
URL: https://github.com/apache/airflow/pull/69118#issuecomment-5265470362
@shahar1 @sameermesiah97 approved this after the branch was updated, and the checks are green. could you provide the final review or merge it?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop BigQueryStreamingBufferEmptySensor reporting empty too early [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #69118:
URL: https://github.com/apache/airflow/pull/69118#discussion_r3501983398
##########
providers/google/tests/unit/google/cloud/sensors/test_bigquery.py:
##########
@@ -343,6 +344,38 @@ def test_poke_returns_false_when_buffer_present(self, mock_hook):
assert sensor.poke(mock.MagicMock()) is False
+ @mock.patch("airflow.providers.google.cloud.sensors.bigquery.BigQueryHook")
+ def test_poke_requires_consecutive_empty_confirmations(self, mock_hook):
+ sensor = _make_streaming_sensor(empty_confirmations=2)
+ mock_hook.return_value.get_client.return_value.get_table.return_value = mock.MagicMock(
+ streaming_buffer=None
+ )
+
+ # A single absent reading is ambiguous (metadata lag), so it must not
+ # report empty; only the second consecutive empty reading does.
+ assert sensor.poke(mock.MagicMock()) is False
+ assert sensor.poke(mock.MagicMock()) is True
Review Comment:
I would parametrize this test to cover non-default configurations for `empty_confirmations`:
```
@pytest.mark.parametrize(
("empty_confirmations", "expected_results"),
[
pytest.param(2, [False, True], id="default"),
pytest.param(3, [False, False, True], id="three_confirmations"),
],
)
@mock.patch("airflow.providers.google.cloud.sensors.bigquery.BigQueryHook")
def test_poke_requires_consecutive_empty_confirmations(
self,
mock_hook,
empty_confirmations,
expected_results,
):
sensor = _make_streaming_sensor(empty_confirmations=empty_confirmations)
mock_hook.return_value.get_client.return_value.get_table.return_value = mock.MagicMock(
streaming_buffer=None
)
# A single absent reading is ambiguous (metadata lag), so the sensor only
# reports empty after the configured number of consecutive empty readings.
for expected in expected_results:
assert sensor.poke(mock.MagicMock()) is expected
```
##########
providers/google/src/airflow/providers/google/cloud/sensors/bigquery.py:
##########
@@ -333,20 +333,26 @@ class BigQueryStreamingBufferEmptySensor(BaseSensorOperator):
``UPDATE/MERGE/DELETE statement over table ... would affect rows in the
streaming buffer`` errors.
- .. warning::
- The sensor reads ``table.streaming_buffer`` from BigQuery's table
- metadata, which is eventually consistent. For a short window right
- after a streaming insert the buffer metadata is still absent, so the
- sensor may report the buffer empty before it actually is. Known
- limitation tracked at
- https://github.com/apache/airflow/issues/66963
+ The ``table.streaming_buffer`` metadata BigQuery exposes is eventually
+ consistent: for a short window right after a streaming insert the rows are
+ in the buffer but the metadata still reads absent. A single absent reading
+ is therefore ambiguous (truly empty vs. metadata lag), so the sensor only
+ reports empty after ``empty_confirmations`` consecutive empty readings, each
+ one ``poke_interval`` apart. This spans the eventual-consistency window
+ without hanging when the table is genuinely empty or the buffer flushes
+ between two pokes.
Review Comment:
I would keep the warning and change the this part of the docstring to this:
```
.. warning::
BigQuery's ``table.streaming_buffer`` metadata is eventually consistent.
The sensor mitigates this by requiring multiple consecutive empty
observations before reporting the buffer empty (configurable via
``empty_confirmations``), but this may delay success by up to one
``poke_interval`` compared to earlier releases.
```
We want to document suprising behaviour here.
##########
providers/google/src/airflow/providers/google/cloud/sensors/bigquery.py:
##########
@@ -426,4 +439,16 @@ def poke(self, context: Context) -> bool:
table = hook.get_client(project_id=self.project_id).get_table(table_ref)
except NotFound as err:
raise ValueError(f"Table {table_uri} not found") from err
- return table.streaming_buffer is None
+
+ if table.streaming_buffer is not None:
+ self._consecutive_empty = 0
+ return False
+
+ self._consecutive_empty += 1
+ self.log.info(
+ "Streaming buffer reported empty (%s/%s confirmations) for table: %s",
+ self._consecutive_empty,
+ self.empty_confirmations,
+ table_uri,
+ )
+ return self._consecutive_empty >= self.empty_confirmations
Review Comment:
nit: this could be more explicit:
```
if self._consecutive_empty >= self.empty_confirmations:
return True
return False
```
##########
providers/google/src/airflow/providers/google/cloud/sensors/bigquery.py:
##########
@@ -333,20 +333,26 @@ class BigQueryStreamingBufferEmptySensor(BaseSensorOperator):
``UPDATE/MERGE/DELETE statement over table ... would affect rows in the
streaming buffer`` errors.
- .. warning::
- The sensor reads ``table.streaming_buffer`` from BigQuery's table
- metadata, which is eventually consistent. For a short window right
- after a streaming insert the buffer metadata is still absent, so the
- sensor may report the buffer empty before it actually is. Known
- limitation tracked at
- https://github.com/apache/airflow/issues/66963
+ The ``table.streaming_buffer`` metadata BigQuery exposes is eventually
+ consistent: for a short window right after a streaming insert the rows are
+ in the buffer but the metadata still reads absent. A single absent reading
+ is therefore ambiguous (truly empty vs. metadata lag), so the sensor only
+ reports empty after ``empty_confirmations`` consecutive empty readings, each
+ one ``poke_interval`` apart. This spans the eventual-consistency window
+ without hanging when the table is genuinely empty or the buffer flushes
+ between two pokes.
:param project_id: Google Cloud project containing the table.
:param dataset_id: Dataset of the table to monitor.
:param table_id: Table to monitor.
:param gcp_conn_id: Airflow connection ID for GCP.
:param impersonation_chain: Optional service account to impersonate, or a
chained list of accounts. See the Google provider docs for details.
+ :param empty_confirmations: Number of consecutive empty readings (each
+ ``poke_interval`` apart) required before reporting the buffer empty.
+ Must be at least 1; values above 1 guard against BigQuery's
+ eventually-consistent streaming-buffer metadata reporting empty too
+ early after a streaming insert.
Review Comment:
This docstring is too long. I would do something like this instead:
```
:param empty_confirmations: Number of consecutive empty readings required
before considering the streaming buffer empty. Must be at least 1.
Defaults to 2.
```
It is good to document the defaults for users.
##########
providers/google/src/airflow/providers/google/cloud/triggers/bigquery.py:
##########
@@ -891,13 +891,21 @@ class BigQueryStreamingBufferEmptyTrigger(BaseTrigger):
Used by :class:`~airflow.providers.google.cloud.sensors.bigquery.BigQueryStreamingBufferEmptySensor`
in deferrable mode.
+ The ``streamingBuffer`` table metadata BigQuery returns is eventually
+ consistent, so a single absent reading can be a false "empty" right after a
+ streaming insert. The trigger therefore yields success only after
+ ``empty_confirmations`` consecutive empty polls, each ``poll_interval``
+ apart.
Review Comment:
This part you added belongs in the comment near the modifications you have made in the `run` method.
##########
providers/google/tests/unit/google/cloud/triggers/test_bigquery.py:
##########
@@ -1070,12 +1081,54 @@ def test_async_hook_receives_impersonation_chain(self, mock_hook_cls, streaming_
@pytest.mark.asyncio
@mock.patch(f"{_TRIGGER_PATH}._is_streaming_buffer_empty")
@mock.patch(f"{_TRIGGER_PATH}._get_async_hook")
- async def test_run_yields_success_when_buffer_empty(
- self, _mock_hook, mock_is_empty, streaming_buffer_trigger
- ):
+ async def test_run_yields_success_when_buffer_empty(self, _mock_hook, mock_is_empty):
+ trigger = BigQueryStreamingBufferEmptyTrigger(
+ project_id=TEST_GCP_PROJECT_ID,
+ dataset_id=TEST_DATASET_ID,
+ table_id=TEST_TABLE_ID,
+ gcp_conn_id=TEST_GCP_CONN_ID,
+ poll_interval=POLLING_PERIOD_SECONDS,
+ impersonation_chain=TEST_IMPERSONATION_CHAIN,
+ empty_confirmations=1,
+ )
mock_is_empty.return_value = True
+ actual = await trigger.run().asend(None)
+
+ table_uri = f"{TEST_GCP_PROJECT_ID}:{TEST_DATASET_ID}.{TEST_TABLE_ID}"
+ assert actual == TriggerEvent(
+ {"status": "success", "message": f"Streaming buffer is empty for table: {table_uri}"}
+ )
+
+ @pytest.mark.asyncio
+ @mock.patch("airflow.providers.google.cloud.triggers.bigquery.asyncio.sleep", new_callable=AsyncMock)
+ @mock.patch(f"{_TRIGGER_PATH}._is_streaming_buffer_empty")
+ @mock.patch(f"{_TRIGGER_PATH}._get_async_hook")
+ async def test_run_waits_for_consecutive_empty_confirmations(
+ self, _mock_hook, mock_is_empty, mock_sleep, streaming_buffer_trigger
+ ):
+ # A single empty reading must not yield success (default empty_confirmations=2);
+ # success only after the second consecutive empty poll.
+ mock_is_empty.side_effect = [True, True]
+ actual = await streaming_buffer_trigger.run().asend(None)
+
+ assert mock_is_empty.await_count == 2
+ table_uri = f"{TEST_GCP_PROJECT_ID}:{TEST_DATASET_ID}.{TEST_TABLE_ID}"
+ assert actual == TriggerEvent(
+ {"status": "success", "message": f"Streaming buffer is empty for table: {table_uri}"}
Review Comment:
I would parametrize this test to cover non-default configurations for `empty_confirmations`:
```
@pytest.mark.asyncio
@pytest.mark.parametrize(
("empty_confirmations", "side_effect", "expected_polls"),
[
pytest.param(2, [True, True], 2, id="default"),
pytest.param(3, [True, True, True], 3, id="three_confirmations"),
],
)
@mock.patch("airflow.providers.google.cloud.triggers.bigquery.asyncio.sleep", new_callable=AsyncMock)
@mock.patch(f"{_TRIGGER_PATH}._is_streaming_buffer_empty")
@mock.patch(f"{_TRIGGER_PATH}._get_async_hook")
async def test_run_waits_for_consecutive_empty_confirmations(
self,
_mock_hook,
mock_is_empty,
mock_sleep,
empty_confirmations,
side_effect,
expected_polls,
):
mock_is_empty.side_effect = side_effect
trigger = BigQueryStreamingBufferEmptyTrigger(
project_id=TEST_GCP_PROJECT_ID,
dataset_id=TEST_DATASET_ID,
table_id=TEST_TABLE_ID,
gcp_conn_id=TEST_GCP_CONN_ID,
empty_confirmations=empty_confirmations,
)
actual = await trigger.run().asend(None)
assert mock_is_empty.await_count == expected_polls
table_uri = f"{TEST_GCP_PROJECT_ID}:{TEST_DATASET_ID}.{TEST_TABLE_ID}"
assert actual == TriggerEvent(
{"status": "success", "message": f"Streaming buffer is empty for table: {table_uri}"}
)
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop BigQueryStreamingBufferEmptySensor reporting empty too early [airflow]
Posted by "anxkhn (via GitHub)" <gi...@apache.org>.
anxkhn commented on PR #69118:
URL: https://github.com/apache/airflow/pull/69118#issuecomment-4853586394
Thanks for the thorough review, all points addressed:
1. Kept the warning: the eventual-consistency note is now a `.. warning::` directive with your wording.
2. Shortened the `empty_confirmations` param docstring to the concise version (it states the default of 2).
3. Made `poke`'s return explicit (`if ...: return True` / `return False`).
4. Moved the eventual-consistency explanation out of the trigger class docstring and into a comment inside `run()`, next to the confirmation-counting logic.
5. Parametrized the trigger test over `empty_confirmations` 2 and 3, asserting it waits for exactly that many consecutive empty polls before yielding success.
6. Parametrized the sensor test the same way.
On the broader semantics point: you are right that the sensor now reports the stable state after N confirmations instead of the first observed reading, which can delay success by up to one `poke_interval` versus earlier releases. That is intentional (it closes the false-empty window) and is now called out in the `.. warning::` so it is not surprising. Ran ruff (check + format), mypy, and both edited test modules locally, all green.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop BigQueryStreamingBufferEmptySensor reporting empty too early [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #69118:
URL: https://github.com/apache/airflow/pull/69118#issuecomment-4830459773
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our [Contributors' Guide](https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:
- Pay attention to the quality of your code (ruff, mypy and type annotations). Our [prek-hooks]( https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst#prerequisites-for-prek-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/main/airflow-core/docs/howto/custom-operator.rst) Consider adding an example Dag that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/dev/breeze/doc/README.rst) for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#coding-style-and-best-practices).
- Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop BigQueryStreamingBufferEmptySensor reporting empty too early [airflow]
Posted by "anxkhn (via GitHub)" <gi...@apache.org>.
anxkhn commented on code in PR #69118:
URL: https://github.com/apache/airflow/pull/69118#discussion_r4050534060
##########
providers/google/src/airflow/providers/google/cloud/sensors/bigquery.py:
##########
@@ -426,4 +435,18 @@ def poke(self, context: Context) -> bool:
table = hook.get_client(project_id=self.project_id).get_table(table_ref)
except NotFound as err:
raise ValueError(f"Table {table_uri} not found") from err
- return table.streaming_buffer is None
+
+ if table.streaming_buffer is not None:
+ self._consecutive_empty = 0
+ return False
+
+ self._consecutive_empty += 1
+ self.log.info(
+ "Streaming buffer reported empty (%s/%s confirmations) for table: %s",
+ self._consecutive_empty,
+ self.empty_confirmations,
+ table_uri,
+ )
+ if self._consecutive_empty >= self.empty_confirmations:
+ return True
+ return False
Review Comment:
Addressed in 1a8aeabe6d. In reschedule mode, once the first empty observation is seen, `poke()` now completes the bounded confirmation sequence in the same process before returning. This prevents operator reconstruction from resetting the counter. A regression test exercises `BaseSensorOperator.execute()` with `mode="reschedule"` and verifies two polls occur without raising `AirflowRescheduleException`.
---
Drafted-by: OpenCode (GPT-5.6) (no human review before posting)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop BigQueryStreamingBufferEmptySensor reporting empty too early [airflow]
Posted by "anxkhn (via GitHub)" <gi...@apache.org>.
anxkhn commented on code in PR #69118:
URL: https://github.com/apache/airflow/pull/69118#discussion_r4050534060
##########
providers/google/src/airflow/providers/google/cloud/sensors/bigquery.py:
##########
@@ -426,4 +435,18 @@ def poke(self, context: Context) -> bool:
table = hook.get_client(project_id=self.project_id).get_table(table_ref)
except NotFound as err:
raise ValueError(f"Table {table_uri} not found") from err
- return table.streaming_buffer is None
+
+ if table.streaming_buffer is not None:
+ self._consecutive_empty = 0
+ return False
+
+ self._consecutive_empty += 1
+ self.log.info(
+ "Streaming buffer reported empty (%s/%s confirmations) for table: %s",
+ self._consecutive_empty,
+ self.empty_confirmations,
+ table_uri,
+ )
+ if self._consecutive_empty >= self.empty_confirmations:
+ return True
+ return False
Review Comment:
Addressed in 1a8aeabe6d. In reschedule mode, once the first empty observation is seen, `poke()` now completes the bounded confirmation sequence in the same process before returning. This prevents operator reconstruction from resetting the counter. A regression test exercises `BaseSensorOperator.execute()` with `mode="reschedule"` and verifies two polls occur without raising `AirflowRescheduleException`.\n\n---\nDrafted-by: OpenCode (GPT-5.6) (no human review before posting)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop BigQueryStreamingBufferEmptySensor reporting empty too early [airflow]
Posted by "anxkhn (via GitHub)" <gi...@apache.org>.
anxkhn commented on code in PR #69118:
URL: https://github.com/apache/airflow/pull/69118#discussion_r4050534060
##########
providers/google/src/airflow/providers/google/cloud/sensors/bigquery.py:
##########
@@ -426,4 +435,18 @@ def poke(self, context: Context) -> bool:
table = hook.get_client(project_id=self.project_id).get_table(table_ref)
except NotFound as err:
raise ValueError(f"Table {table_uri} not found") from err
- return table.streaming_buffer is None
+
+ if table.streaming_buffer is not None:
+ self._consecutive_empty = 0
+ return False
+
+ self._consecutive_empty += 1
+ self.log.info(
+ "Streaming buffer reported empty (%s/%s confirmations) for table: %s",
+ self._consecutive_empty,
+ self.empty_confirmations,
+ table_uri,
+ )
+ if self._consecutive_empty >= self.empty_confirmations:
+ return True
+ return False
Review Comment:
Addressed in c6003a2d6b. In reschedule mode, once the first empty observation is seen, `poke()` now completes the bounded confirmation sequence in the same process before returning. This prevents operator reconstruction from resetting the counter. A regression test exercises `BaseSensorOperator.execute()` with `mode="reschedule"` and verifies two polls occur without raising `AirflowRescheduleException`.
---
Drafted-by: OpenCode (GPT-5.6) (no human review before posting)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop BigQueryStreamingBufferEmptySensor reporting empty too early [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 commented on code in PR #69118:
URL: https://github.com/apache/airflow/pull/69118#discussion_r4050291019
##########
providers/google/src/airflow/providers/google/cloud/sensors/bigquery.py:
##########
@@ -426,4 +435,18 @@ def poke(self, context: Context) -> bool:
table = hook.get_client(project_id=self.project_id).get_table(table_ref)
except NotFound as err:
raise ValueError(f"Table {table_uri} not found") from err
- return table.streaming_buffer is None
+
+ if table.streaming_buffer is not None:
+ self._consecutive_empty = 0
+ return False
+
+ self._consecutive_empty += 1
+ self.log.info(
+ "Streaming buffer reported empty (%s/%s confirmations) for table: %s",
+ self._consecutive_empty,
+ self.empty_confirmations,
+ table_uri,
+ )
+ if self._consecutive_empty >= self.empty_confirmations:
+ return True
+ return False
Review Comment:
In mode="reschedule", Airflow creates a new sensor instance for every poke. That resets _consecutive_empty to 0, so with empty_confirmations=2 the counter never reaches 2 and the sensor keeps rescheduling until timeout.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop BigQueryStreamingBufferEmptySensor reporting empty too early [airflow]
Posted by "VladaZakharova (via GitHub)" <gi...@apache.org>.
VladaZakharova commented on code in PR #69118:
URL: https://github.com/apache/airflow/pull/69118#discussion_r4131756308
##########
providers/google/src/airflow/providers/google/cloud/sensors/bigquery.py:
##########
@@ -422,8 +432,28 @@ def poke(self, context: Context) -> bool:
)
hook = BigQueryHook(gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain)
- try:
- table = hook.get_client(project_id=self.project_id).get_table(table_ref)
- except NotFound as err:
- raise ValueError(f"Table {table_uri} not found") from err
- return table.streaming_buffer is None
+ client = hook.get_client(project_id=self.project_id)
Review Comment:
Keeping the confirmation loop inside poke() means mode="reschedule" occupies a worker and can bypass the normal sensor timeout check when the final poll succeeds. Could the empty-state/timestamp be persisted across actual reschedules, or could this mode use the trigger implementation instead? Please also add an execute-level timeout test; the current direct poke() test does not cover BaseSensorOperator semantics.
##########
providers/google/tests/system/google/cloud/bigquery/example_bigquery_streaming_buffer_sensor.py:
##########
@@ -109,19 +108,6 @@ def streaming_insert(ds: str | None = None) -> None:
rows=[{"value": 100, "ds": ds}],
fail_on_error=True,
)
Review Comment:
The updated system-test DAG inserts rows, runs the synchronous sensor, performs an update, and only then runs the deferrable sensor.
Therefore, only the synchronous sensor sees the “immediately after streaming insert” condition. Add another streaming insert or a separate table immediately before the deferrable senso
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop BigQueryStreamingBufferEmptySensor reporting empty too early [airflow]
Posted by "VladaZakharova (via GitHub)" <gi...@apache.org>.
VladaZakharova commented on PR #69118:
URL: https://github.com/apache/airflow/pull/69118#issuecomment-5867320061
Thank you for the PR! Can you please attach the screenshot of green system test for this operator?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org