Fix Async GCSObjectsWithPrefixExistenceSensor xcom push

airflow/providers/google/cloud/sensors/gcs.py

@@ -363,6 +363,8 @@ def execute(self, context: Context):
                     ),
                     method_name="execute_complete",
                 )
+            else:
+                return self._matches
 
     def execute_complete(self, context: dict[str, Any], event: dict[str, str | list[str]]) -> str | list[str]:
         """Return immediately and rely on trigger to throw a success event. Callback for the trigger."""

tests/providers/google/cloud/sensors/test_gcs.py

@@ -429,6 +429,20 @@ def test_gcs_object_prefix_existence_sensor_finish_before_deferred(self, mock_de
         task.execute(mock.MagicMock())
         assert not mock_defer.called
 
+    @mock.patch("airflow.providers.google.cloud.sensors.gcs.GCSHook")
+    def test_xcom_value_when_poke_success(self, mock_hook):
+        mock_hook.return_value.list.return_value = ["test.txt"]
+        task = GCSObjectsWithPrefixExistenceSensor(
+            task_id="task-id",
+            bucket=TEST_BUCKET,
+            prefix=TEST_PREFIX,
+            google_cloud_conn_id=TEST_GCP_CONN_ID,
+            impersonation_chain=TEST_IMPERSONATION_CHAIN,
+            deferrable=True,
+        )
+        responses = task.execute(None)
+        assert responses == ["test.txt"]
+
 
 class TestGCSObjectsWithPrefixExistenceAsyncSensor:
     OPERATOR = GCSObjectsWithPrefixExistenceSensor(