E575 feat(pubsub): ordering keys (#26) · googleapis/python-pubsub@cc3093a · GitHub
[go: up one dir, main page]
Skip to content
This repository was archived by the owner on Mar 9, 2026. It is now read-only.

Commit cc3093a

Browse files
pradnpradn
authored
feat(pubsub): ordering keys (#26)
1 parent da96526 commit cc3093a

25 files changed

+2406
-327
lines changed

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

Lines changed: 25 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
from __future__ import absolute_import
1616

1717
import abc
18+
import enum
1819

1920
import six
2021

@@ -134,6 +135,18 @@ def will_accept(self, message):
134135
# Okay, everything is good.
135136
return True
136137

138+
def cancel(self, cancellation_reason):
139+
"""Complete pending futures with an exception.
140+
141+
This method must be called before publishing starts (ie: while the
142+
batch is still accepting messages.)
143+
144+
Args:
145+
cancellation_reason (BatchCancellationReason): The reason why this
146+
batch has been cancelled.
147+
"""
148+
raise NotImplementedError
149+
137150
@abc.abstractmethod
138151
def publish(self, message):
139152
"""Publish a single message.
@@ -154,16 +167,21 @@ def publish(self, message):
154167
raise NotImplementedError
155168

156169

157-
class BatchStatus(object):
158-
"""An enum-like class representing valid statuses for a batch.
159-
160-
It is acceptable for a class to use a status that is not on this
161-
class; this represents the list of statuses where the existing
162-
library hooks in functionality.
163-
"""
170+
class BatchStatus(str, enum.Enum):
171+
"""An enum-like class representing valid statuses for a batch."""
164172

165173
ACCEPTING_MESSAGES = "accepting messages"
166174
STARTING = "starting"
167175
IN_PROGRESS = "in progress"
168176
ERROR = "error"
169177
SUCCESS = "success"
178+
179+
180+
class BatchCancellationReason(str, enum.Enum):
181+
"""An enum-like class representing reasons why a batch was cancelled."""
182+
183+
PRIOR_ORDERED_MESSAGE_FAILED = (
184+
"Batch cancelled because prior ordered message for the same key has "
185+
"failed. This batch has been cancelled to avoid out-of-order publish."
186+
)
187+
CLIENT_STOPPED = "Batch cancelled because the publisher client has been stopped."

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

Lines changed: 64 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -62,15 +62,23 @@ class Batch(base.Batch):
6262
settings (~.pubsub_v1.types.BatchSettings): The settings for batch
6363
publishing. These should be considered immutable once the batch
6464
has been opened.
65-
autocommit (bool): Whether to autocommit the batch when the time
66-
has elapsed. Defaults to True unless ``settings.max_latency`` is
67-
inf.
65+
batch_done_callback (Callable[[bool], Any]): Callback called when the
66+
response for a batch publish has been received. Called with one
67+
boolean argument: successfully published or a permanent error
68+
occurred. Temporary errors are not surfaced because they are retried
69+
at a lower level.
70+
commit_when_full (bool): Whether to commit the batch when the batch
71+
is full.
6872
"""
6973

70-
def __init__(self, client, topic, settings, autocommit=True):
74+
def __init__(
75+
self, client, topic, settings, batch_done_callback=None, commit_when_full=True
76+
):
7177
self._client = client
7278
self._topic = topic
7379
self._settings = settings
80+
self._batch_done_callback = batch_done_callback
81+
self._commit_when_full = commit_when_full
7482

7583
self._state_lock = threading.Lock()
7684
# These members are all communicated between threads; ensure that
@@ -87,15 +95,6 @@ def __init__(self, client, topic, settings, autocommit=True):
8795
self._base_request_size = types.PublishRequest(topic=topic).ByteSize()
8896
self._size = self._base_request_size
8997

90-
# If max latency is specified, start a thread to monitor the batch and
91-
# commit when the max latency is reached.
92-
self._thread = None
93-
if autocommit and self.settings.max_latency < float("inf"):
94-
self._thread = threading.Thread(
95-
name="Thread-MonitorBatchPublisher", target=self.monitor
96-
)
97-
self._thread.start()
98-
9998
@staticmethod
10099
def make_lock():
101100
"""Return a threading lock.
@@ -148,6 +147,27 @@ def status(self):
148147
"""
149148
return self._status
150149

150+
def cancel(self, cancellation_reason):
151+
"""Complete pending futures with an exception.
152+
153+
This method must be called before publishing starts (ie: while the
154+
batch is still accepting messages.)
155+
156+
Args:
157+
cancellation_reason (BatchCancellationReason): The reason why this
158+
batch has been cancelled.
159+
"""
160+
161+
with self._state_lock:
162+
assert (
163+
self._status == base.BatchStatus.ACCEPTING_MESSAGES
164+
), "Cancel should not be called after sending has started."
165+
166+
exc = RuntimeError(cancellation_reason.value)
167+
for future in self._futures:
168+
future.set_exception(exc)
169+
self._status = base.BatchStatus.ERROR
170+
151171
def commit(self):
152172
"""Actually publish all of the messages on the active batch.
153173
@@ -162,6 +182,7 @@ def commit(self):
162182
If the current batch is **not** accepting messages, this method
163183
does nothing.
164184
"""
185+
165186
# Set the status to "starting" synchronously, to ensure that
166187
# this batch will necessarily not accept new messages.
167188
with self._state_lock:
@@ -170,7 +191,11 @@ def commit(self):
170191
else:
171192
return
172193

173-
# Start a new thread to actually handle the commit.
194+
self._start_commit_thread()
195+
196+
def _start_commit_thread(self):
197+
"""Start a new thread to actually handle the commit."""
198+
174199
commit_thread = threading.Thread(
175200
name="Thread-CommitBatchPublisher", target=self._commit
176201
)
@@ -195,7 +220,10 @@ def _commit(self):
195220
# If, in the intervening period between when this method was
196221
# called and now, the batch started to be committed, or
197222
# completed a commit, then no-op at this point.
198-
_LOGGER.debug("Batch is already in progress, exiting commit")
223+
_LOGGER.debug(
224+
"Batch is already in progress or has been cancelled, "
225+
"exiting commit"
226+
)
199227
return
200228

201229
# Once in the IN_PROGRESS state, no other thread can publish additional
@@ -215,16 +243,24 @@ def _commit(self):
215243
# Log how long the underlying request takes.
216244
start = time.time()
217245

246+
batch_transport_succeeded = True
218247
try:
248+
# Performs retries for errors defined in retry_codes.publish in the
249+
# publisher_client_config.py file.
219250
response = self._client.api.publish(self._topic, self._messages)
220251
except google.api_core.exceptions.GoogleAPIError as exc:
221-
# We failed to publish, set the exception on all futures and
222-
# exit.
252+
# We failed to publish, even after retries, so set the exception on
253+
# all futures and exit.
223254
self._status = base.BatchStatus.ERROR
224255

225256
for future in self._futures:
226257
future.set_exception(exc)
227258

259+
batch_transport_succeeded = False
260+
if self._batch_done_callback is not None:
261+
# Failed to publish batch.
262+
self._batch_done_callback(batch_transport_succeeded)
263+
228264
_LOGGER.exception("Failed to publish %s messages.", len(self._futures))
229265
return
230266

@@ -250,26 +286,17 @@ def _commit(self):
250286
for future in self._futures:
251287
future.set_exception(exception)
252288

289+
# Unknown error -> batch failed to be correctly transported/
290+
batch_transport_succeeded = False
291+
253292
_LOGGER.error(
254293
"Only %s of %s messages were published.",
255294
len(response.message_ids),
256295
len(self._futures),
257296
)
258297

259-
def monitor(self):
260-
"""Commit this batch after sufficient time has elapsed.
261-
262-
This simply sleeps for ``self.settings.max_latency`` seconds,
263-
and then calls commit unless the batch has already been committed.
264-
"""
265-
# NOTE: This blocks; it is up to the calling code to call it
266-
# in a separate thread.
267-
268-
# Sleep for however long we should be waiting.
269-
time.sleep(self.settings.max_latency)
270-
271-
_LOGGER.debug("Monitor is waking up")
272-
return self._commit()
298+
if self._batch_done_callback is not None:
299+
self._batch_done_callback(batch_transport_succeeded)
273300

274301
def publish(self, message):
275302
"""Publish a single message.
@@ -294,13 +321,18 @@ def publish(self, message):
294321
pubsub_v1.publisher.exceptions.MessageTooLargeError: If publishing
295322
the ``message`` would exceed the max size limit on the backend.
296323
"""
324+
297325
# Coerce the type, just in case.
298326
if not isinstance(message, types.PubsubMessage):
299327
message = types.PubsubMessage(**message)
300328

301329
future = None
302330

303331
with self._state_lock:
332+
assert (
333+
self._status != base.BatchStatus.ERROR
334+
), "Publish after stop() or publish error."
335+
304336
if not self.will_accept(message):
305337
return future
306338

@@ -333,7 +365,7 @@ def publish(self, message):
4A96 333365

334366
# Try to commit, but it must be **without** the lock held, since
335367
# ``commit()`` will try to obtain the lock.
336-
if overflow:
368+
if self._commit_when_full and overflow:
337369
self.commit()
338370

339371
return future

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

Whitespace-only changes.

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

C12A
Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
1+
# Copyright 2019, Google LLC All rights reserved.
2+
#
3+
# Licensed under the Apache License, Version 2.0 (the "License");
4+
# you may not use this file except in compliance with the License.
5+
# You may obtain a copy of the License at
6+
#
7+
# http://www.apache.org/licenses/LICENSE-2.0
8+
#
9+
# Unless required by applicable law or agreed to in writing, software
10+
# distributed under the License is distributed on an "AS IS" BASIS,
11+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
# See the License for the specific language governing permissions and
13+
# limitations under the License.
14+
15+
from __future__ import absolute_import
16+
17+
import abc
18+
19+
import six
20+
21+
22+
@six.add_metaclass(abc.ABCMeta)
23+
class Sequencer(object):
24+
"""The base class for sequencers for Pub/Sub publishing. A sequencer
25+
sequences messages to be published.
26+
"""
27+
28+
@staticmethod
29+
@abc.abstractmethod
30+
def is_finished(self):
31+
""" Whether the sequencer is finished and should be cleaned up.
32+
33+
Returns:
34+
bool: Whether the sequencer is finished and should be cleaned up.
35+
"""
36+
raise NotImplementedError
37+
38+
@staticmethod
39+
@abc.abstractmethod
40+
def unpause(self, message):
41+
""" Unpauses this sequencer.
42+
43+
Raises:
44+
RuntimeError:
45+
If called when the sequencer has not been paused.
46+
"""
47+
raise NotImplementedError
48+
49+
@staticmethod
50+
@abc.abstractmethod
51+
def publish(self, message):
52+
""" Publish message for this ordering key.
53+
54+
Args:
55+
message (~.pubsub_v1.types.PubsubMessage): The Pub/Sub message.
56+
57+
Returns:
58+
A class instance that conforms to Python Standard library's
59+
:class:`~concurrent.futures.Future` interface (but not an
60+
instance of that class). The future might return immediately with a
61+
`pubsub_v1.publisher.exceptions.PublishToPausedOrderingKeyException`
62+
if the ordering key is paused. Otherwise, the future tracks the
63+
lifetime of the message publish.
64+
65+
Raises:
66+
RuntimeError:
67+
If called after this sequencer has been stopped, either by
68+
a call to stop() or after all batches have been published.
69+
"""
70+
raise NotImplementedError

0 commit comments

Comments
 (0)
0