Skip to content

Commit ba2c2ee

Browse files
fix: Stop using api_core default timeouts in publish since they are broken (#1326)
1 parent 8adb133 commit ba2c2ee

File tree

2 files changed

+14
-2
lines changed

2 files changed

+14
-2
lines changed

google/cloud/pubsub_v1/types.py

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@
3535
from google.protobuf import timestamp_pb2
3636

3737
from google.api_core.protobuf_helpers import get_messages
38+
from google.api_core.timeout import ConstantTimeout
3839

3940
from google.pubsub_v1.types import pubsub as pubsub_gapic_types
4041

@@ -191,7 +192,10 @@ class PublisherOptions(NamedTuple):
191192
"an instance of :class:`google.api_core.retry.Retry`."
192193
)
193194

194-
timeout: "OptionalTimeout" = gapic_v1.method.DEFAULT # use api_core default
195+
# Use ConstantTimeout instead of api_core default because the default
196+
# value results in retries with zero deadline.
197+
# Refer https://github.com/googleapis/python-api-core/issues/654
198+
timeout: "OptionalTimeout" = ConstantTimeout(60)
195199
(
196200
"Timeout settings for message publishing by the client. It should be "
197201
"compatible with :class:`~.pubsub_v1.types.TimeoutType`."

tests/unit/pubsub_v1/publisher/test_publisher_client.py

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import sys
2020

2121
import grpc
22+
import math
2223

2324
# special case python < 3.8
2425
if sys.version_info.major == 3 and sys.version_info.minor < 8:
@@ -35,6 +36,7 @@
3536
from google.api_core import gapic_v1
3637
from google.api_core import retry as retries
3738
from google.api_core.gapic_v1.client_info import METRICS_METADATA_KEY
39+
from google.api_core.timeout import ConstantTimeout
3840

3941
from google.cloud.pubsub_v1 import publisher
4042
from google.cloud.pubsub_v1 import types
@@ -652,6 +654,8 @@ def test_publish_new_batch_needed(creds):
652654
future = client.publish(topic, b"foo", bar=b"baz")
653655
assert future is mock.sentinel.future
654656

657+
call_args = batch_class.call_args
658+
655659
# Check the mocks.
656660
batch_class.assert_called_once_with(
657661
client=mock.ANY,
@@ -660,8 +664,12 @@ def test_publish_new_batch_needed(creds):
660664
batch_done_callback=None,
661665
commit_when_full=True,
662666
commit_retry=gapic_v1.method.DEFAULT,
663-
commit_timeout=gapic_v1.method.DEFAULT,
667+
commit_timeout=mock.ANY,
664668
)
669+
commit_timeout_arg = call_args[1]["commit_timeout"]
670+
assert isinstance(commit_timeout_arg, ConstantTimeout)
671+
assert math.isclose(commit_timeout_arg._timeout, 60) is True
672+
665673
message_pb = gapic_types.PubsubMessage(data=b"foo", attributes={"bar": "baz"})
666674
wrapper = PublishMessageWrapper(message=message_pb)
667675
batch1.publish.assert_called_once_with(wrapper)

0 commit comments

Comments
 (0)