Skip to content

Commit

Permalink
fix: Stop using api_core default timeouts in publish since they are b…
Browse files Browse the repository at this point in the history
…roken (#1326)
  • Loading branch information
mukund-ananthu authored Jan 24, 2025
1 parent 8adb133 commit ba2c2ee
Show file tree
Hide file tree
Showing 2 changed files with 14 additions and 2 deletions.
6 changes: 5 additions & 1 deletion google/cloud/pubsub_v1/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
from google.protobuf import timestamp_pb2

from google.api_core.protobuf_helpers import get_messages
from google.api_core.timeout import ConstantTimeout

from google.pubsub_v1.types import pubsub as pubsub_gapic_types

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

timeout: "OptionalTimeout" = gapic_v1.method.DEFAULT # use api_core default
# Use ConstantTimeout instead of api_core default because the default
# value results in retries with zero deadline.
# Refer https://github.com/googleapis/python-api-core/issues/654
timeout: "OptionalTimeout" = ConstantTimeout(60)
(
"Timeout settings for message publishing by the client. It should be "
"compatible with :class:`~.pubsub_v1.types.TimeoutType`."
Expand Down
10 changes: 9 additions & 1 deletion tests/unit/pubsub_v1/publisher/test_publisher_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
import sys

import grpc
import math

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

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

call_args = batch_class.call_args

# Check the mocks.
batch_class.assert_called_once_with(
client=mock.ANY,
Expand All @@ -660,8 +664,12 @@ def test_publish_new_batch_needed(creds):
batch_done_callback=None,
commit_when_full=True,
commit_retry=gapic_v1.method.DEFAULT,
commit_timeout=gapic_v1.method.DEFAULT,
commit_timeout=mock.ANY,
)
commit_timeout_arg = call_args[1]["commit_timeout"]
assert isinstance(commit_timeout_arg, ConstantTimeout)
assert math.isclose(commit_timeout_arg._timeout, 60) is True

message_pb = gapic_types.PubsubMessage(data=b"foo", attributes={"bar": "baz"})
wrapper = PublishMessageWrapper(message=message_pb)
batch1.publish.assert_called_once_with(wrapper)
Expand Down

0 comments on commit ba2c2ee

Please sign in to comment.