fix: add warning to retry target to avoid incorrect usage (#543)

* fix: add warning to retry target to avoid incorrect use

* fix: add comment for predicate

* 🦉 Updates from OwlBot post-processor

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

---------

Co-authored-by: omair <omairn@google.com>
Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>

google/api_core/retry.py

@@ -62,6 +62,8 @@ def check_if_exists():
 import random
 import sys
 import time
+import inspect
+import warnings
 from typing import Any, Callable, TypeVar, TYPE_CHECKING
 
 import requests.exceptions
@@ -84,6 +86,7 @@ def check_if_exists():
 _DEFAULT_MAXIMUM_DELAY = 60.0  # seconds
 _DEFAULT_DELAY_MULTIPLIER = 2.0
 _DEFAULT_DEADLINE = 60.0 * 2.0  # seconds
+_ASYNC_RETRY_WARNING = "Using the synchronous google.api_core.retry.Retry with asynchronous calls may lead to unexpected results. Please use google.api_core.retry_async.AsyncRetry instead."
 
 
 def if_exception_type(
@@ -201,7 +204,10 @@ def retry_target(
 
     for sleep in sleep_generator:
         try:
-            return target()
+            result = target()
+            if inspect.isawaitable(result):
+                warnings.warn(_ASYNC_RETRY_WARNING)
+            return result
 
         # pylint: disable=broad-except
         # This function explicitly must deal with broad exceptions.

tests/unit/test_retry.py

@@ -129,6 +129,26 @@ def test_retry_target_non_retryable_error(utcnow, sleep):
     sleep.assert_not_called()
 
 
+@mock.patch("asyncio.sleep", autospec=True)
+@mock.patch(
+    "google.api_core.datetime_helpers.utcnow",
+    return_value=datetime.datetime.min,
+    autospec=True,
+)
+@pytest.mark.asyncio
+async def test_retry_target_warning_for_retry(utcnow, sleep):
+    predicate = retry.if_exception_type(ValueError)
+    target = mock.AsyncMock(spec=["__call__"])
+
+    with pytest.warns(Warning) as exc_info:
+        # Note: predicate is just a filler and doesn't affect the test
+        retry.retry_target(target, predicate, range(10), None)
+
+    assert len(exc_info) == 2
+    assert str(exc_info[0].message) == retry._ASYNC_RETRY_WARNING
+    sleep.assert_not_called()
+
+
 @mock.patch("time.sleep", autospec=True)
 @mock.patch("google.api_core.datetime_helpers.utcnow", autospec=True)
 def test_retry_target_deadline_exceeded(utcnow, sleep):