diff --git a/providers/google/docs/operators/cloud/bigquery.rst b/providers/google/docs/operators/cloud/bigquery.rst index 41c49dd5e0311..292a745669360 100644 --- a/providers/google/docs/operators/cloud/bigquery.rst +++ b/providers/google/docs/operators/cloud/bigquery.rst @@ -518,6 +518,28 @@ Also you can use deferrable mode in this operator if you would like to free up t :start-after: [START howto_sensor_bigquery_table_partition_async] :end-before: [END howto_sensor_bigquery_table_partition_async] +Check that the BigQuery Table Streaming Buffer is empty +"""""""""""""""""""""""""""""""""""""""""""""""""""""""" + +To check that the BigQuery streaming buffer of a table is empty you can use +:class:`~airflow.providers.google.cloud.sensors.bigquery.BigQueryStreamingBufferEmptySensor`. +This sensor is useful in ETL pipelines to ensure that recent streamed data has been fully +processed before continuing downstream tasks. + +.. exampleinclude:: /../../google/tests/system/google/cloud/bigquery/example_bigquery_sensors.py + :language: python + :dedent: 4 + :start-after: [START howto_sensor_bigquery_streaming_buffer_empty] + :end-before: [END howto_sensor_bigquery_streaming_buffer_empty] + +Also you can use deferrable mode in this operator if you would like to free up the worker slots while the sensor is running. + +.. exampleinclude:: /../../google/tests/system/google/cloud/bigquery/example_bigquery_sensors.py + :language: python + :dedent: 4 + :start-after: [START howto_sensor_bigquery_streaming_buffer_empty_defered] + :end-before: [END howto_sensor_bigquery_streaming_buffer_empty_defered] + Reference ^^^^^^^^^ diff --git a/providers/google/src/airflow/providers/google/cloud/sensors/bigquery.py b/providers/google/src/airflow/providers/google/cloud/sensors/bigquery.py index 5b891bb075c0c..23c035615cd66 100644 --- a/providers/google/src/airflow/providers/google/cloud/sensors/bigquery.py +++ b/providers/google/src/airflow/providers/google/cloud/sensors/bigquery.py @@ -28,6 +28,7 @@ from airflow.providers.common.compat.sdk import AirflowException, BaseSensorOperator, conf from airflow.providers.google.cloud.hooks.bigquery import BigQueryHook from airflow.providers.google.cloud.triggers.bigquery import ( + BigQueryStreamingBufferEmptyTrigger, BigQueryTableExistenceTrigger, BigQueryTablePartitionExistenceTrigger, ) @@ -88,9 +89,7 @@ def __init__( ) else: kwargs["poke_interval"] = 5 - super().__init__(**kwargs) - self.project_id = project_id self.dataset_id = dataset_id self.table_id = table_id @@ -256,3 +255,143 @@ def execute_complete(self, context: dict[str, Any], event: dict[str, str] | None message = "No event received in trigger callback" raise AirflowException(message) + + +class BigQueryStreamingBufferEmptySensor(BaseSensorOperator): + """ + Sensor for checking whether the streaming buffer in a BigQuery table is empty. + + The BigQueryStreamingBufferEmptySensor waits for the streaming buffer in a specified + BigQuery table to be empty before proceeding. It can be used in ETL pipelines to ensure + that recent streamed data has been processed before continuing downstream tasks. + + :ivar template_fields: Fields that can be templated in this operator. + :type template_fields: Sequence[str] + :ivar ui_color: Color of the operator in the Airflow UI. + :type ui_color: Str + :ivar project_id: The Google Cloud project ID where the BigQuery table resides. + :type project_id: Str + :ivar dataset_id: The ID of the dataset containing the BigQuery table. + :type dataset_id: Str + :ivar table_id: The ID of the BigQuery table to monitor. + :type table_id: Str + :ivar gcp_conn_id: The Airflow connection ID for GCP. Defaults to "google_cloud_default". + :type gcp_conn_id: Str + :ivar impersonation_chain: Optional array or string of service accounts to impersonate using short-term + credentials. If multiple accounts are provided, the service account must + grant the role `roles/iam.serviceAccountTokenCreator` on the next account + in the chain. + :type impersonation_chain: Str | Sequence[str] | None + :ivar deferrable: Indicates whether the operator supports deferrable execution. If True, the sensor + can defer instead of polling, leading to reduced resource use. + :type deferrable: Bool + """ + + template_fields: Sequence[str] = ( + "project_id", + "dataset_id", + "table_id", + "impersonation_chain", + ) + + ui_color = "#f0eee4" + + def __init__( + self, + *, + project_id: str, + dataset_id: str, + table_id: str, + gcp_conn_id: str = "google_cloud_default", + impersonation_chain: str | Sequence[str] | None = None, + deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False), + **kwargs, + ) -> None: + if deferrable and "poke_interval" not in kwargs: + kwargs["poke_interval"] = 30 + + super().__init__(**kwargs) + + self.project_id = project_id + self.dataset_id = dataset_id + self.table_id = table_id + self.gcp_conn_id = gcp_conn_id + self.impersonation_chain = impersonation_chain + self.deferrable = deferrable + + def execute(self, context: Context) -> None: + """ + Executes the operator logic taking into account the `deferrable` attribute. + If not deferrable, it uses the base class `execute` method; otherwise, it + sets up deferral with a specific trigger to wait until the BigQuery Streaming + Buffer is empty. + + :param context: The execution context provided by Airflow. It allows + access to metadata and runtime information of the task instance. + :type context: Context + :return: None. The method does not return anything but performs actions + relevant to the operator's execution. + """ + if not self.deferrable: + super().execute(context) + else: + if not self.poke(context=context): + self.defer( + timeout=timedelta(seconds=self.timeout), + trigger=BigQueryStreamingBufferEmptyTrigger( + project_id=self.project_id, + dataset_id=self.dataset_id, + table_id=self.table_id, + poll_interval=self.poke_interval, + gcp_conn_id=self.gcp_conn_id, + hook_params={ + "impersonation_chain": self.impersonation_chain, + }, + ), + method_name="execute_complete", + ) + + def execute_complete(self, context: dict[str, Any], event: dict[str, str] | None = None) -> str: + """ + Act as a callback for when the trigger fires - returns immediately. + + Relies on trigger to throw an exception, otherwise it assumes execution was successful. + """ + table_uri = f"{self.project_id}:{self.dataset_id}.{self.table_id}" + self.log.info("Checking streaming buffer state for table: %s", table_uri) + if event: + if event["status"] == "success": + return event["message"] + raise AirflowException(event["message"]) + + message = "No event received in trigger callback" + raise AirflowException(message) + + def poke(self, context: Context) -> bool: + """ + Check if the BigQuery streaming buffer is empty for the specified table. + + This method periodically checks the status of the BigQuery table's streaming buffer + to determine if it is empty. It is useful for ensuring that recent streamed data + has been fully processed before continuing with downstream tasks. + """ + table_uri = f"{self.project_id}:{self.dataset_id}.{self.table_id}" + self.log.info("Checking streaming buffer state for table: %s", table_uri) + + hook = BigQueryHook( + gcp_conn_id=self.gcp_conn_id, + impersonation_chain=self.impersonation_chain, + ) + try: + table = hook.get_table( + project_id=self.project_id, + dataset_id=self.dataset_id, + table_id=self.table_id, + ) + return table.get("streamingBuffer") is None + except Exception as err: + if "not found" in str(err): + raise AirflowException( + f"Table {self.project_id}.{self.dataset_id}.{self.table_id} not found" + ) from err + raise err diff --git a/providers/google/src/airflow/providers/google/cloud/triggers/bigquery.py b/providers/google/src/airflow/providers/google/cloud/triggers/bigquery.py index 70419e5b08d3f..b4b5fddd124a8 100644 --- a/providers/google/src/airflow/providers/google/cloud/triggers/bigquery.py +++ b/providers/google/src/airflow/providers/google/cloud/triggers/bigquery.py @@ -806,3 +806,158 @@ async def _partition_exists(self, hook: BigQueryAsyncHook, job_id: str | None, p if records: records = [row[0] for row in records] return self.partition_id in records + + +class BigQueryStreamingBufferEmptyTrigger(BaseTrigger): + """ + Trigger that periodically checks if a BigQuery table's streaming buffer is empty. + + This trigger continuously polls a BigQuery table to determine if its streaming buffer + has been fully processed and is now empty. It's particularly useful before running + DML operations (UPDATE, DELETE, MERGE) on tables populated via streaming inserts. + + :param project_id: The Google Cloud Project in which to look for the table. + :param dataset_id: The dataset ID of the table to check. + :param table_id: The table ID of the table to check. + :param gcp_conn_id: Reference to Google Cloud connection ID. + :param hook_params: Additional parameters for hook initialization. + :param poll_interval: Polling period in seconds to check the streaming buffer status. + :param impersonation_chain: Optional service account to impersonate using short-term + credentials, or chained list of accounts required to get the access_token + of the last account in the list, which will be impersonated in the request. + If set as a string, the account must grant the originating account + the Service Account Token Creator IAM role. + If set as a sequence, the identities from the list must grant + Service Account Token Creator IAM role to the directly preceding identity, with first + account from the list granting this role to the originating account. + """ + + def __init__( + self, + project_id: str, + dataset_id: str, + table_id: str, + gcp_conn_id: str, + hook_params: dict[str, Any], + poll_interval: float = 30.0, + impersonation_chain: str | Sequence[str] | None = None, + ): + super().__init__() + self.project_id = project_id + self.dataset_id = dataset_id + self.table_id = table_id + self.gcp_conn_id = gcp_conn_id + self.poll_interval = poll_interval + self.hook_params = hook_params + self.impersonation_chain = impersonation_chain + + def serialize(self) -> tuple[str, dict[str, Any]]: + """Serialize BigQueryStreamingBufferEmptyTrigger arguments and classpath.""" + return ( + "airflow.providers.google.cloud.triggers.bigquery.BigQueryStreamingBufferEmptyTrigger", + { + "project_id": self.project_id, + "dataset_id": self.dataset_id, + "table_id": self.table_id, + "gcp_conn_id": self.gcp_conn_id, + "poll_interval": self.poll_interval, + "hook_params": self.hook_params, + "impersonation_chain": self.impersonation_chain, + }, + ) + + def _get_async_hook(self) -> BigQueryTableAsyncHook: + """Get the async hook for BigQuery table operations.""" + return BigQueryTableAsyncHook( + gcp_conn_id=self.gcp_conn_id, + impersonation_chain=self.impersonation_chain, + ) + + async def run(self) -> AsyncIterator[TriggerEvent]: + """ + Continuously check if the streaming buffer is empty. + + Yields a TriggerEvent when the streaming buffer becomes empty or if an error occurs. + """ + try: + hook = self._get_async_hook() + while True: + self.log.info( + "Checking streaming buffer for table %s.%s.%s", + self.project_id, + self.dataset_id, + self.table_id, + ) + + is_buffer_empty = await self._is_streaming_buffer_empty( + hook=hook, + project_id=self.project_id, + dataset_id=self.dataset_id, + table_id=self.table_id, + ) + + if is_buffer_empty: + table_uri = f"{self.project_id}:{self.dataset_id}.{self.table_id}" + message = f"Streaming buffer is empty for table: {table_uri}" + self.log.info(message) + yield TriggerEvent( + { + "status": "success", + "message": message, + } + ) + return + else: + self.log.info( + "Streaming buffer still has data. Sleeping for %s seconds.", + self.poll_interval, + ) + await asyncio.sleep(self.poll_interval) + + except Exception as e: + self.log.exception( + "Exception occurred while checking streaming buffer for table %s.%s.%s", + self.project_id, + self.dataset_id, + self.table_id, + ) + yield TriggerEvent({"status": "error", "message": str(e)}) + + async def _is_streaming_buffer_empty( + self, + hook: BigQueryTableAsyncHook, + project_id: str, + dataset_id: str, + table_id: str, + ) -> bool: + """ + Check if the streaming buffer is empty for the specified table. + + :param hook: BigQueryTableAsyncHook instance for async operations. + :param project_id: The Google Cloud Project ID. + :param dataset_id: The dataset ID containing the table. + :param table_id: The table ID to check. + :return: True if streaming buffer is empty or doesn't exist, False otherwise. + """ + async with ClientSession() as session: + try: + client = await hook.get_table_client( + dataset=dataset_id, + table_id=table_id, + project_id=project_id, + session=session, + ) + response = await client.get() + + if not response: + # Table doesn't exist + raise AirflowException(f"Table {project_id}.{dataset_id}.{table_id} does not exist") + + # Check if streamingBuffer exists in the response + streaming_buffer = response.get("streamingBuffer") + return streaming_buffer is None + + except ClientResponseError as err: + if err.status == 404: + raise AirflowException(f"Table {project_id}.{dataset_id}.{table_id} not found") from err + raise err diff --git a/providers/google/tests/system/google/cloud/bigquery/example_bigquery_sensors.py b/providers/google/tests/system/google/cloud/bigquery/example_bigquery_sensors.py index 48a0780dbedbd..f56b674dcb1c8 100644 --- a/providers/google/tests/system/google/cloud/bigquery/example_bigquery_sensors.py +++ b/providers/google/tests/system/google/cloud/bigquery/example_bigquery_sensors.py @@ -32,6 +32,7 @@ BigQueryInsertJobOperator, ) from airflow.providers.google.cloud.sensors.bigquery import ( + BigQueryStreamingBufferEmptySensor, BigQueryTableExistenceSensor, BigQueryTablePartitionExistenceSensor, ) @@ -54,6 +55,11 @@ INSERT_ROWS_QUERY = f"INSERT {DATASET_NAME}.{TABLE_NAME} VALUES (42, '{{{{ ds }}}}')" +# Streaming INSERT, UPDATE, DELETE queries example +STREAMING_INSERT_QUERY = f"INSERT {DATASET_NAME}.{TABLE_NAME} VALUES (100, '{{{{ ds }}}}')" +STREAMING_UPDATE_QUERY = f"UPDATE {DATASET_NAME}.{TABLE_NAME} SET value = 200 WHERE value = 100" +STREAMING_DELETE_QUERY = f"DELETE FROM {DATASET_NAME}.{TABLE_NAME} WHERE value = 200" + SCHEMA = [ {"name": "value", "type": "INTEGER", "mode": "REQUIRED"}, {"name": "ds", "type": "DATE", "mode": "NULLABLE"}, @@ -108,6 +114,7 @@ project_id=PROJECT_ID, dataset_id=DATASET_NAME, table_id=TABLE_NAME, + deferrable=True, ) # [END howto_sensor_async_bigquery_table] @@ -149,9 +156,65 @@ project_id=PROJECT_ID, dataset_id=DATASET_NAME, table_id=TABLE_NAME, + deferrable=True, ) # [END howto_sensor_bigquery_table_partition_async] + # [START howto_sensor_bigquery_streaming_buffer_empty] + check_streaming_buffer_empty = BigQueryStreamingBufferEmptySensor( + task_id="check_streaming_buffer_empty", + project_id=PROJECT_ID, + dataset_id=DATASET_NAME, + table_id=TABLE_NAME, + poke_interval=30, + timeout=5400, # 90 minutes - Google Cloud flushes streaming buffer within 90 minutes + ) + # [END howto_sensor_bigquery_streaming_buffer_empty] + + # Streaming operations: INSERT, UPDATE, DELETE + # These operations write data to the streaming buffer before being flushed to persistent storage + stream_insert = BigQueryInsertJobOperator( + task_id="stream_insert", + configuration={ + "query": { + "query": STREAMING_INSERT_QUERY, + "useLegacySql": False, + } + }, + ) + + stream_update = BigQueryInsertJobOperator( + task_id="stream_update", + configuration={ + "query": { + "query": STREAMING_UPDATE_QUERY, + "useLegacySql": False, + } + }, + ) + + stream_delete = BigQueryInsertJobOperator( + task_id="stream_delete", + configuration={ + "query": { + "query": STREAMING_DELETE_QUERY, + "useLegacySql": False, + } + }, + ) + + # [START howto_sensor_bigquery_streaming_buffer_empty_defered] + check_streaming_buffer_empty_def = BigQueryStreamingBufferEmptySensor( + task_id="check_streaming_buffer_empty_def", + project_id=PROJECT_ID, + dataset_id=DATASET_NAME, + table_id=TABLE_NAME, + deferrable=True, + poke_interval=30, + timeout=5400, # 90 minutes - Google Cloud flushes streaming buffer within 90 minutes + ) + # [END howto_sensor_bigquery_streaming_buffer_empty_defered] + delete_dataset = BigQueryDeleteDatasetOperator( task_id="delete_dataset", dataset_id=DATASET_NAME, @@ -169,6 +232,11 @@ check_table_partition_exists_async, check_table_partition_exists_def, ] + >> [stream_insert, stream_update, stream_delete] + >> [ + check_streaming_buffer_empty, + check_streaming_buffer_empty_def, + ] >> delete_dataset ) diff --git a/providers/google/tests/unit/google/cloud/triggers/test_bigquery.py b/providers/google/tests/unit/google/cloud/triggers/test_bigquery.py index 22f4a17cf54bd..baae78606d074 100644 --- a/providers/google/tests/unit/google/cloud/triggers/test_bigquery.py +++ b/providers/google/tests/unit/google/cloud/triggers/test_bigquery.py @@ -888,3 +888,168 @@ def test_serialization_successfully(self): "poll_interval": POLLING_PERIOD_SECONDS, "hook_params": TEST_HOOK_PARAMS, } + + +@pytest.fixture +def streaming_buffer_trigger(): + return BigQueryStreamingBufferEmptyTrigger( + project_id=TEST_GCP_PROJECT_ID, + dataset_id=TEST_DATASET_ID, + table_id=TEST_TABLE_ID, + gcp_conn_id=TEST_GCP_CONN_ID, + hook_params=TEST_HOOK_PARAMS, + poll_interval=POLLING_PERIOD_SECONDS, + impersonation_chain=TEST_IMPERSONATION_CHAIN, + ) + + +class TestBigQueryStreamingBufferEmptyTrigger: + def test_serialization(self, streaming_buffer_trigger): + """Asserts that the trigger correctly serializes its arguments and classpath.""" + classpath, kwargs = streaming_buffer_trigger.serialize() + assert ( + classpath + == "airflow.providers.google.cloud.triggers.bigquery.BigQueryStreamingBufferEmptyTrigger" + ) + assert kwargs == { + "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, + "hook_params": TEST_HOOK_PARAMS, + "impersonation_chain": TEST_IMPERSONATION_CHAIN, + } + + @pytest.mark.asyncio + @mock.patch( + "airflow.providers.google.cloud.triggers.bigquery." + "BigQueryStreamingBufferEmptyTrigger._is_streaming_buffer_empty" + ) + @mock.patch( + "airflow.providers.google.cloud.triggers.bigquery.BigQueryStreamingBufferEmptyTrigger._get_sync_hook" + ) + async def test_run_buffer_empty_yields_success( + self, mock_get_hook, mock_is_empty, streaming_buffer_trigger + ): + """When buffer is empty, trigger yields a success TriggerEvent.""" + mock_is_empty.return_value = True + + generator = streaming_buffer_trigger.run() + actual = await generator.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." + "BigQueryStreamingBufferEmptyTrigger._is_streaming_buffer_empty" + ) + @mock.patch( + "airflow.providers.google.cloud.triggers.bigquery.BigQueryStreamingBufferEmptyTrigger._get_sync_hook" + ) + async def test_run_buffer_not_empty_keeps_polling( + self, mock_get_hook, mock_is_empty, streaming_buffer_trigger + ): + """When buffer is not empty, trigger keeps polling (does not yield immediately).""" + mock_is_empty.return_value = False + + task = asyncio.create_task(streaming_buffer_trigger.run().__anext__()) + await asyncio.sleep(0.5) + + # TriggerEvent was not returned yet + assert task.done() is False + asyncio.get_event_loop().stop() + + @pytest.mark.asyncio + @mock.patch( + "airflow.providers.google.cloud.triggers.bigquery." + "BigQueryStreamingBufferEmptyTrigger._is_streaming_buffer_empty" + ) + @mock.patch( + "airflow.providers.google.cloud.triggers.bigquery.BigQueryStreamingBufferEmptyTrigger._get_sync_hook" + ) + async def test_run_exception_yields_error_event( + self, mock_get_hook, mock_is_empty, streaming_buffer_trigger + ): + """When the hook raises an exception, trigger yields an error TriggerEvent.""" + mock_is_empty.side_effect = Exception("API failure") + + generator = streaming_buffer_trigger.run() + actual = await generator.asend(None) + + assert actual == TriggerEvent({"status": "error", "message": "API failure"}) + + @pytest.mark.asyncio + @mock.patch("airflow.providers.google.cloud.hooks.bigquery.BigQueryHook.get_client") + async def test_is_streaming_buffer_empty_true(self, mock_get_client, streaming_buffer_trigger): + """_is_streaming_buffer_empty returns True when streaming_buffer is None (SDK Table object).""" + bq_table = BQTable.from_api_repr(BQ_TABLE_RESOURCE_NO_BUFFER) + mock_get_client.return_value.get_table.return_value = bq_table + + hook = mock.MagicMock(spec=BigQueryHook) + hook.get_client = mock_get_client + result = await streaming_buffer_trigger._is_streaming_buffer_empty( + hook, TEST_GCP_PROJECT_ID, TEST_DATASET_ID, TEST_TABLE_ID + ) + assert result is True + mock_get_client.return_value.get_table.assert_called_once_with( + f"{TEST_GCP_PROJECT_ID}.{TEST_DATASET_ID}.{TEST_TABLE_ID}" + ) + + @pytest.mark.asyncio + @mock.patch("airflow.providers.google.cloud.hooks.bigquery.BigQueryHook.get_client") + async def test_is_streaming_buffer_empty_false(self, mock_get_client, streaming_buffer_trigger): + """_is_streaming_buffer_empty returns False when streaming_buffer is present (SDK Table object).""" + bq_table = BQTable.from_api_repr(BQ_TABLE_RESOURCE_WITH_BUFFER) + mock_get_client.return_value.get_table.return_value = bq_table + + hook = mock.MagicMock(spec=BigQueryHook) + hook.get_client = mock_get_client + result = await streaming_buffer_trigger._is_streaming_buffer_empty( + hook, TEST_GCP_PROJECT_ID, TEST_DATASET_ID, TEST_TABLE_ID + ) + assert result is False + mock_get_client.return_value.get_table.assert_called_once_with( + f"{TEST_GCP_PROJECT_ID}.{TEST_DATASET_ID}.{TEST_TABLE_ID}" + ) + + @pytest.mark.asyncio + @mock.patch("airflow.providers.google.cloud.hooks.bigquery.BigQueryHook.get_client") + async def test_is_streaming_buffer_empty_not_found_raises( + self, mock_get_client, streaming_buffer_trigger + ): + """_is_streaming_buffer_empty raises ValueError on google.api_core.exceptions.NotFound (404).""" + # Real 404 error message from BigQuery when table does not exist. + # Captured via: bq show student-00343:test.NonExistentTable + mock_get_client.return_value.get_table.side_effect = NotFound( + f"Not found: Table {TEST_GCP_PROJECT_ID}:{TEST_DATASET_ID}.NonExistentTable" + ) + + hook = mock.MagicMock(spec=BigQueryHook) + hook.get_client = mock_get_client + with pytest.raises(ValueError, match="not found"): + await streaming_buffer_trigger._is_streaming_buffer_empty( + hook, TEST_GCP_PROJECT_ID, TEST_DATASET_ID, TEST_TABLE_ID + ) + + @pytest.mark.asyncio + @mock.patch("airflow.providers.google.cloud.hooks.bigquery.BigQueryHook.get_client") + async def test_is_streaming_buffer_empty_forbidden_reraised( + self, mock_get_client, streaming_buffer_trigger + ): + """_is_streaming_buffer_empty re-raises non-NotFound exceptions (e.g. Forbidden 403).""" + mock_get_client.return_value.get_table.side_effect = Forbidden( + f"Access Denied: Table {TEST_GCP_PROJECT_ID}:{TEST_DATASET_ID}.{TEST_TABLE_ID}: " + "Permission bigquery.tables.get denied" + ) + + hook = mock.MagicMock(spec=BigQueryHook) + hook.get_client = mock_get_client + with pytest.raises(Forbidden): + await streaming_buffer_trigger._is_streaming_buffer_empty( + hook, TEST_GCP_PROJECT_ID, TEST_DATASET_ID, TEST_TABLE_ID + )