Skip to content

Commit

Permalink
fix: Stop using api_core default timeouts in publish since they are
Browse files Browse the repository at this point in the history
broken
  • Loading branch information
mukund-ananthu committed Jan 11, 2025
1 parent f648f65 commit c527c2e
Show file tree
Hide file tree
Showing 2 changed files with 22 additions and 2 deletions.
8 changes: 7 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 ExponentialTimeout

from google.pubsub_v1.types import pubsub as pubsub_gapic_types

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

timeout: "OptionalTimeout" = gapic_v1.method.DEFAULT # use api_core default
# Use ExponentialTimeout 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" = ExponentialTimeout(
initial=5, maximum=60, multiplier=1.3, deadline=600
)
(
"Timeout settings for message publishing by the client. It should be "
"compatible with :class:`~.pubsub_v1.types.TimeoutType`."
Expand Down
16 changes: 15 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 ExponentialTimeout

from google.cloud.pubsub_v1 import publisher
from google.cloud.pubsub_v1 import types
Expand Down Expand Up @@ -646,12 +648,15 @@ def test_publish_new_batch_needed(creds):

# Actually mock the batch class now.
batch_class = mock.Mock(spec=(), return_value=batch2)

client._set_batch_class(batch_class)

# Publish a message.
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 +665,17 @@ 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, ExponentialTimeout)
assert math.isclose(commit_timeout_arg._initial, 5) is True
assert math.isclose(commit_timeout_arg._maximum, 60) is True
assert math.isclose(commit_timeout_arg._multiplier, 1.3) is True
assert math.isclose(commit_timeout_arg._deadline, 600) is True
# ExponentialTimeout(
# initial=5, maximum=60, multiplier=1.3, deadline=600
# ),
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 c527c2e

Please sign in to comment.