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

fix: don't assert on unordered publishes after publish error #49

Merged
pradn merged 4 commits into from
Mar 23, 2020
Merged

fix: don't assert on unordered publishes after publish error #49

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
bc5129e
fix: allow unordered publishes after publish error
pradn Mar 20, 2020
e87903a
Merge branch 'master' into fix_assert
pradn Mar 20, 2020
c3b3496
Clarify comment / retrigger checks
pradn Mar 23, 2020
c6d4283
Merge branch 'master' into fix_assert
pradn Mar 23, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions google/cloud/pubsub_v1/publisher/_batch/thread.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -369,3 +369,6 @@ def publish(self, message):
self.commit()

return future

def _set_status(self, status):
self._status = status
6 changes: 6 additions & 0 deletions google/cloud/pubsub_v1/publisher/_sequencer/unordered_sequencer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,12 @@ def commit(self):
if self._current_batch:
self._current_batch.commit()

# At this point, we lose track of the old batch, but we don't
# care since we just committed it.
# Setting this to None guarantees the next publish() creates a new
# batch.
self._current_batch = None

def unpause(self):
""" Not relevant for this class. """
raise NotImplementedError
Expand Down
22 changes: 22 additions & 0 deletions tests/unit/pubsub_v1/publisher/sequencer/test_unordered_sequencer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
from google.auth import credentials
from google.cloud.pubsub_v1 import publisher
from google.cloud.pubsub_v1 import types
from google.cloud.pubsub_v1.publisher._batch import base
from google.cloud.pubsub_v1.publisher._sequencer import unordered_sequencer


Expand Down Expand Up @@ -102,3 +103,24 @@ def test_publish_batch_full():
future = sequencer.publish(message)
batch.publish.assert_called_once_with(message)
assert future is not None


def test_publish_after_batch_error():
client = create_client()
message = create_message()
batch = mock.Mock(spec=client._batch_class)

sequencer = unordered_sequencer.UnorderedSequencer(client, "topic_name")
sequencer._set_batch(batch)

sequencer.commit()
batch.commit.assert_called_once()

# Simulate publish RPC failing.
batch._set_status(base.BatchStatus.ERROR)

# Will create a new batch since the old one has been committed. The fact
# that the old batch errored should not matter in the publish of the next
# message.
future = sequencer.publish(message)
assert future is not None