Skip to content
This repository was archived by the owner on Mar 9, 2026. It is now read-only.

Commit 7597604

Browse files
plamutcguardia
plamut
and
cguardia
authored
feat: support customizable retry and timeout settings on the publisher client (#299)
* feat: allow retry and timeout settings on publisher client * build: update generated code and update doc * Propagate publish timeout to the RPC call * test publisher client * Fix timeout parameter type in docstrings * Apply timeout changes to async publisher client * Introduce TimeoutType type alias * Update PublisherOptions docs * Use type alias for timeout in upgrading guide * Widen timeout types in generated publisher clients * Use TimeoutType in handwritten code * Remove redundant backslash * Use DEFAULT as a defualt timeout in base sequencer * Do not accept plain None as a valid timeout Using no timeout is not a good idea, but if one really wants to, they can pass it in as ConstantTimeout(None). As a side effect, the logic of converting a constant into a COnstantTimeout instance can be removed, as this is already handled in api-core for int and float values. Co-authored-by: Carlos de la Guardia <[email protected]>
1 parent 691a3dd commit 7597604

File tree

15 files changed

+357
-78
lines changed

15 files changed

+357
-78
lines changed

‎UPGRADING.md‎

Lines changed: 42 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -100,7 +100,7 @@ specified by the API producer.
100100
*,
101101
project: str = None,
102102
retry: retries.Retry = gapic_v1.method.DEFAULT,
103-
timeout: float = None,
103+
timeout: google.pubsub_v1.types.TimeoutType = gapic_v1.method.DEFAULT,
104104
metadata: Sequence[Tuple[str, str]] = (),
105105
) -> pagers.ListTopicsPager:
106106
```
@@ -161,3 +161,44 @@ The publisher and subscriber clients cannot be constructed with `client_config`
161161
argument anymore. If you want to customize retry and timeout settings for a particular
162162
method, you need to do it upon method invocation by passing the custom `timeout` and
163163
`retry` arguments, respectively.
164+
165+
166+
## Custom Retry and Timeout settings for Publisher Client
167+
168+
The ``publisher_options`` parameter to the Publisher Client, as well as all of the
169+
client's methods, now accept custom retry and timeout settings:
170+
171+
```py
172+
custom_retry = api_core.retry.Retry(
173+
initial=0.250, # seconds (default: 0.1)
174+
maximum=90.0, # seconds (default: 60.0)
175+
multiplier=1.45, # default: 1.3
176+
deadline=300.0, # seconds (default: 60.0)
177+
predicate=api_core.retry.if_exception_type(
178+
api_core.exceptions.Aborted,
179+
api_core.exceptions.DeadlineExceeded,
180+
api_core.exceptions.InternalServerError,
181+
api_core.exceptions.ResourceExhausted,
182+
api_core.exceptions.ServiceUnavailable,
183+
api_core.exceptions.Unknown,
184+
api_core.exceptions.Cancelled,
185+
),
186+
)
187+
188+
custom_timeout=api_core.timeout.ExponentialTimeout(
189+
initial=1.0,
190+
maximum=10.0,
191+
multiplier=1.0,
192+
deadline=300.0,
193+
)
194+
195+
publisher = pubsub_v1.PublisherClient(
196+
publisher_options = pubsub_v1.types.PublisherOptions(
197+
retry=custom_retry,
198+
timeout=custom_timeout,
199+
),
200+
)
201+
```
202+
203+
The timeout can be either an instance of `google.api_core.timeout.ConstantTimeout`,
204+
or an instance of `google.api_core.timeout.ExponentialTimeout`, as in the example.

‎google/cloud/pubsub_v1/publisher/_batch/thread.py‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,9 @@ class Batch(base.Batch):
7373
commit_retry (Optional[google.api_core.retry.Retry]): Designation of what
7474
errors, if any, should be retried when commiting the batch. If not
7575
provided, a default retry is used.
76+
commit_timeout (:class:`~.pubsub_v1.types.TimeoutType`):
77+
The timeout to apply when commiting the batch. If not provided, a
78+
default timeout is used.
7679
"""
7780

7881
def __init__(
@@ -83,6 +86,7 @@ def __init__(
8386
batch_done_callback=None,
8487
commit_when_full=True,
8588
commit_retry=gapic_v1.method.DEFAULT,
89+
commit_timeout: gapic_types.TimeoutType = gapic_v1.method.DEFAULT,
8690
):
8791
self._client = client
8892
self._topic = topic
@@ -106,6 +110,7 @@ def __init__(
106110
self._size = self._base_request_size
107111

108112
self._commit_retry = commit_retry
113+
self._commit_timeout = commit_timeout
109114

110115
@staticmethod
111116
def make_lock():
@@ -261,7 +266,10 @@ def _commit(self):
261266
try:
262267
# Performs retries for errors defined by the retry configuration.
263268
response = self._client.api.publish(
264-
topic=self._topic, messages=self._messages, retry=self._commit_retry
269+
topic=self._topic,
270+
messages=self._messages,
271+
retry=self._commit_retry,
272+
timeout=self._commit_timeout,
265273
)
266274
except google.api_core.exceptions.GoogleAPIError as exc:
267275
# We failed to publish, even after retries, so set the exception on

‎google/cloud/pubsub_v1/publisher/_sequencer/base.py‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,9 @@
1616

1717
import abc
1818

19+
from google.api_core import gapic_v1
20+
from google.pubsub_v1 import types as gapic_types
21+
1922

2023
class Sequencer(metaclass=abc.ABCMeta):
2124
"""The base class for sequencers for Pub/Sub publishing. A sequencer
@@ -45,14 +48,21 @@ def unpause(self, message): # pragma: NO COVER
4548

4649
@staticmethod
4750
@abc.abstractmethod
48-
def publish(self, message, retry=None): # pragma: NO COVER
51+
def publish(
52+
self,
53+
message,
54+
retry=None,
55+
timeout: gapic_types.TimeoutType = gapic_v1.method.DEFAULT,
56+
): # pragma: NO COVER
4957
""" Publish message for this ordering key.
5058
5159
Args:
5260
message (~.pubsub_v1.types.PubsubMessage):
5361
The Pub/Sub message.
5462
retry (Optional[google.api_core.retry.Retry]):
5563
The retry settings to apply when publishing the message.
64+
timeout (:class:`~.pubsub_v1.types.TimeoutType`):
65+
The timeout to apply when publishing the message.
5666
5767
Returns:
5868
A class instance that conforms to Python Standard library's

‎google/cloud/pubsub_v1/publisher/_sequencer/ordered_sequencer.py‎

Lines changed: 21 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
from google.cloud.pubsub_v1.publisher import exceptions
2222
from google.cloud.pubsub_v1.publisher._sequencer import base as sequencer_base
2323
from google.cloud.pubsub_v1.publisher._batch import base as batch_base
24+
from google.pubsub_v1 import types as gapic_types
2425

2526

2627
class _OrderedSequencerStatus(str, enum.Enum):
@@ -226,13 +227,19 @@ def unpause(self):
226227
raise RuntimeError("Ordering key is not paused.")
227228
self._state = _OrderedSequencerStatus.ACCEPTING_MESSAGES
228229

229-
def _create_batch(self, commit_retry=gapic_v1.method.DEFAULT):
230+
def _create_batch(
231+
self,
232+
commit_retry=gapic_v1.method.DEFAULT,
233+
commit_timeout: gapic_types.TimeoutType = gapic_v1.method.DEFAULT,
234+
):
230235
""" Create a new batch using the client's batch class and other stored
231236
settings.
232237
233238
Args:
234239
commit_retry (Optional[google.api_core.retry.Retry]):
235240
The retry settings to apply when publishing the batch.
241+
commit_timeout (:class:`~.pubsub_v1.types.TimeoutType`):
242+
The timeout to apply when publishing the batch.
236243
"""
237244
return self._client._batch_class(
238245
client=self._client,
@@ -241,16 +248,24 @@ def _create_batch(self, commit_retry=gapic_v1.method.DEFAULT):
241248
batch_done_callback=self._batch_done_callback,
242249
commit_when_full=False,
243250
commit_retry=commit_retry,
251+
commit_timeout=commit_timeout,
244252
)
245253

246-
def publish(self, message, retry=gapic_v1.method.DEFAULT):
254+
def publish(
255+
self,
256+
message,
257+
retry=gapic_v1.method.DEFAULT,
258+
timeout: gapic_types.TimeoutType = gapic_v1.method.DEFAULT,
259+
):
247260
""" Publish message for this ordering key.
248261
249262
Args:
250263
message (~.pubsub_v1.types.PubsubMessage):
251264
The Pub/Sub message.
252265
retry (Optional[google.api_core.retry.Retry]):
253266
The retry settings to apply when publishing the message.
267+
timeout (:class:`~.pubsub_v1.types.TimeoutType`):
268+
The timeout to apply when publishing the message.
254269
255270
Returns:
256271
A class instance that conforms to Python Standard library's
@@ -287,13 +302,15 @@ def publish(self, message, retry=gapic_v1.method.DEFAULT):
287302
), "Publish is only allowed in accepting-messages state."
288303

289304
if not self._ordered_batches:
290-
new_batch = self._create_batch(commit_retry=retry)
305+
new_batch = self._create_batch(
306+
commit_retry=retry, commit_timeout=timeout
307+
)
291308
self._ordered_batches.append(new_batch)
292309

293310
batch = self._ordered_batches[-1]
294311
future = batch.publish(message)
295312
while future is None:
296-
batch = self._create_batch(commit_retry=retry)
313+
batch = self._create_batch(commit_retry=retry, commit_timeout=timeout)
297314
self._ordered_batches.append(batch)
298315
future = batch.publish(message)
299316

‎google/cloud/pubsub_v1/publisher/_sequencer/unordered_sequencer.py‎

Lines changed: 19 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
from google.api_core import gapic_v1
1616

1717
from google.cloud.pubsub_v1.publisher._sequencer import base
18+
from google.pubsub_v1 import types as gapic_types
1819

1920

2021
class UnorderedSequencer(base.Sequencer):
@@ -77,13 +78,19 @@ def unpause(self):
7778
""" Not relevant for this class. """
7879
raise NotImplementedError
7980

80-
def _create_batch(self, commit_retry=gapic_v1.method.DEFAULT):
81+
def _create_batch(
82+
self,
83+
commit_retry=gapic_v1.method.DEFAULT,
84+
commit_timeout: gapic_types.TimeoutType = gapic_v1.method.DEFAULT,
85+
):
8186
""" Create a new batch using the client's batch class and other stored
8287
settings.
8388
8489
Args:
8590
commit_retry (Optional[google.api_core.retry.Retry]):
8691
The retry settings to apply when publishing the batch.
92+
commit_timeout (:class:`~.pubsub_v1.types.TimeoutType`):
93+
The timeout to apply when publishing the batch.
8794
"""
8895
return self._client._batch_class(
8996
client=self._client,
@@ -92,16 +99,24 @@ def _create_batch(self, commit_retry=gapic_v1.method.DEFAULT):
9299
batch_done_callback=None,
93100
commit_when_full=True,
94101
commit_retry=commit_retry,
102+
commit_timeout=commit_timeout,
95103
)
96104

97-
def publish(self, message, retry=gapic_v1.method.DEFAULT):
105+
def publish(
106+
self,
107+
message,
108+
retry=gapic_v1.method.DEFAULT,
109+
timeout: gapic_types.TimeoutType = gapic_v1.method.DEFAULT,
110+
):
98111
""" Batch message into existing or new batch.
99112
100113
Args:
101114
message (~.pubsub_v1.types.PubsubMessage):
102115
The Pub/Sub message.
103116
retry (Optional[google.api_core.retry.Retry]):
104117
The retry settings to apply when publishing the message.
118+
timeout (:class:`~.pubsub_v1.types.TimeoutType`):
119+
The timeout to apply when publishing the message.
105120
106121
Returns:
107122
~google.api_core.future.Future: An object conforming to
@@ -119,7 +134,7 @@ def publish(self, message, retry=gapic_v1.method.DEFAULT):
119134
raise RuntimeError("Unordered sequencer already stopped.")
120135

121136
if not self._current_batch:
122-
newbatch = self._create_batch(commit_retry=retry)
137+
newbatch = self._create_batch(commit_retry=retry, commit_timeout=timeout)
123138
self._current_batch = newbatch
124139

125140
batch = self._current_batch
@@ -129,7 +144,7 @@ def publish(self, message, retry=gapic_v1.method.DEFAULT):
129144
future = batch.publish(message)
130145
# batch is full, triggering commit_when_full
131146
if future is None:
132-
batch = self._create_batch(commit_retry=retry)
147+
batch = self._create_batch(commit_retry=retry, commit_timeout=timeout)
133148
# At this point, we lose track of the old batch, but we don't
134149
# care since it's already committed (because it was full.)
135150
self._current_batch = batch

‎google/cloud/pubsub_v1/publisher/client.py‎

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -230,7 +230,13 @@ def resume_publish(self, topic, ordering_key):
230230
sequencer.unpause()
231231

232232
def publish(
233-
self, topic, data, ordering_key="", retry=gapic_v1.method.DEFAULT, **attrs
233+
self,
234+
topic,
235+
data,
236+
ordering_key="",
237+
retry=gapic_v1.method.DEFAULT,
238+
timeout: gapic_types.TimeoutType = gapic_v1.method.DEFAULT,
239+
**attrs
234240
):
235241
"""Publish a single message.
236242
@@ -269,6 +275,12 @@ def publish(
269275
retry (Optional[google.api_core.retry.Retry]): Designation of what
270276
errors, if any, should be retried. If `ordering_key` is specified,
271277
the total retry deadline will be changed to "infinity".
278+
If given, it overides any retry passed into the client through
279+
the ``publisher_options`` argument.
280+
timeout (:class:`~.pubsub_v1.types.TimeoutType`):
281+
The timeout for the RPC request. Can be used to override any timeout
282+
passed in through ``publisher_options`` when instantiating the client.
283+
272284
attrs (Mapping[str, str]): A dictionary of attributes to be
273285
sent as metadata. (These may be text strings or byte strings.)
274286
@@ -331,6 +343,12 @@ def publish(
331343
def on_publish_done(future):
332344
self._flow_controller.release(message)
333345

346+
if retry is gapic_v1.method.DEFAULT: # if custom retry not passed in
347+
retry = self.publisher_options.retry
348+
349+
if timeout is gapic_v1.method.DEFAULT: # if custom timeout not passed in
350+
timeout = self.publisher_options.timeout
351+
334352
with self._batch_lock:
335353
if self._is_stopped:
336354
raise RuntimeError("Cannot publish on a stopped publisher.")
@@ -347,7 +365,7 @@ def on_publish_done(future):
347365

348366
# Delegate the publishing to the sequencer.
349367
sequencer = self._get_or_create_sequencer(topic, ordering_key)
350-
future = sequencer.publish(message, retry=retry)
368+
future = sequencer.publish(message, retry=retry, timeout=timeout)
351369
future.add_done_callback(on_publish_done)
352370

353371
# Create a timer thread if necessary to enforce the batching

‎google/cloud/pubsub_v1/types.py‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import proto
2323

2424
from google.api import http_pb2
25+
from google.api_core import gapic_v1
2526
from google.iam.v1 import iam_policy_pb2
2627
from google.iam.v1 import policy_pb2
2728
from google.iam.v1.logging import audit_data_pb2
@@ -98,11 +99,13 @@ class LimitExceededBehavior(str, enum.Enum):
9899
# This class is used when creating a publisher client to pass in options
99100
# to enable/disable features.
100101
PublisherOptions = collections.namedtuple(
101-
"PublisherConfig", ["enable_message_ordering", "flow_control"]
102+
"PublisherOptions", ["enable_message_ordering", "flow_control", "retry", "timeout"]
102103
)
103104
PublisherOptions.__new__.__defaults__ = (
104105
False, # enable_message_ordering: False
105106
PublishFlowControl(), # default flow control settings
107+
gapic_v1.method.DEFAULT, # use default api_core value for retry
108+
gapic_v1.method.DEFAULT, # use default api_core value for timeout
106109
)
107110
PublisherOptions.__doc__ = "The options for the publisher client."
108111
PublisherOptions.enable_message_ordering.__doc__ = (
@@ -112,6 +115,14 @@ class LimitExceededBehavior(str, enum.Enum):
112115
"Flow control settings for message publishing by the client. By default "
113116
"the publisher client does not do any throttling."
114117
)
118+
PublisherOptions.retry.__doc__ = (
119+
"Retry settings for message publishing by the client. This should be "
120+
"an instance of :class:`google.api_core.retry.Retry`."
121+
)
122+
PublisherOptions.timeout.__doc__ = (
123+
"Timeout settings for message publishing by the client. It should be compatible "
124+
"with :class:`~.pubsub_v1.types.TimeoutType`."
125+
)
115126

116127
# Define the type class and default values for flow control settings.
117128
#

0 commit comments

Comments
 (0)