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

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

@@ -369,3 +369,6 @@ def publish(self, message):
             self.commit()
 
         return future
+
+    def _set_status(self, status):
+        self._status = status

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

@@ -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

tests/unit/pubsub_v1/publisher/sequencer/test_unordered_sequencer.py

@@ -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
 
 
@@ -102,3 +103,23 @@ 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 it errored should not matter in the publish of the next message.
+    future = sequencer.publish(message)
+    assert future is not None