Un-confuse the "active"<->"stopped" relationship in Pub/Sub consumer. (#4562)

I messed this up during #4549, and as a result, there is no
lease maintenance in the `0.29.3` release.

Yay Friday releases!

Author: dhermes

packages/google-cloud-pubsub/google/cloud/pubsub_v1/subscriber/policy/base.py

@@ -173,15 +173,15 @@ def ack(self, ack_id, time_to_ack=None, byte_size=None):
         if time_to_ack is not None:
             self.histogram.add(int(time_to_ack))
 
-        # Send the request to ack the message.
-        # However, if the consumer is inactive, then queue the ack_id here
-        # instead; it will be acked as part of the initial request when the
-        # consumer is started again.
         if self._consumer.stopped.is_set():
+            # If the consumer is inactive, then queue the ack_id here; it
+            # will be acked as part of the initial request when the consumer
+            # is started again.
+            self._ack_on_resume.add(ack_id)
+        else:
+            # Send the request to ack the message.
             request = types.StreamingPullRequest(ack_ids=[ack_id])
             self._consumer.send_request(request)
-        else:
-            self._ack_on_resume.add(ack_id)
 
         # Remove the message from lease management.
         self.drop(ack_id=ack_id, byte_size=byte_size)
@@ -311,7 +311,7 @@ def maintain_leases(self):
         """
         while True:
             # Sanity check: Should this infinite loop quit?
-            if not self._consumer.stopped.is_set():
+            if self._consumer.stopped.is_set():
                 _LOGGER.debug('Consumer inactive, ending lease maintenance.')
                 return
 
@@ -326,7 +326,7 @@ def maintain_leases(self):
             # because it is more efficient to make a single request.
             ack_ids = list(self.managed_ack_ids)
             _LOGGER.debug('Renewing lease for %d ack IDs.', len(ack_ids))
-            if ack_ids and self._consumer.stopped.is_set():
+            if ack_ids and not self._consumer.stopped.is_set():
                 request = types.StreamingPullRequest(
                     modify_deadline_ack_ids=ack_ids,
                     modify_deadline_seconds=[p99] * len(ack_ids),

packages/google-cloud-pubsub/tests/unit/pubsub_v1/subscriber/test_policy_base.py

@@ -85,7 +85,7 @@ def test_subscription():
 
 def test_ack():
     policy = create_policy()
-    policy._consumer.stopped.set()
+    policy._consumer.stopped.clear()
     with mock.patch.object(policy._consumer, 'send_request') as send_request:
         policy.ack('ack_id_string', 20)
         send_request.assert_called_once_with(types.StreamingPullRequest(
@@ -97,7 +97,7 @@ def test_ack():
 
 def test_ack_no_time():
     policy = create_policy()
-    policy._consumer.stopped.set()
+    policy._consumer.stopped.clear()
     with mock.patch.object(policy._consumer, 'send_request') as send_request:
         policy.ack('ack_id_string')
         send_request.assert_called_once_with(types.StreamingPullRequest(
@@ -110,6 +110,7 @@ def test_ack_paused():
     policy = create_policy()
     policy._paused = True
     consumer = policy._consumer
+    consumer.stopped.set()
 
     with mock.patch.object(consumer, 'resume') as resume:
         policy.ack('ack_id_string')
@@ -206,20 +207,21 @@ def test_modify_ack_deadline():
 
 def test_maintain_leases_inactive_consumer():
     policy = create_policy()
-    policy._consumer.stopped.clear()
+    policy._consumer.stopped.set()
     assert policy.maintain_leases() is None
 
 
 def test_maintain_leases_ack_ids():
     policy = create_policy()
-    policy._consumer.stopped.set()
+    policy._consumer.stopped.clear()
     policy.lease('my ack id', 50)
 
     # Mock the sleep object.
     with mock.patch.object(time, 'sleep', autospec=True) as sleep:
         def trigger_inactive(seconds):
             assert 0 < seconds < 10
-            policy._consumer.stopped.clear()
+            policy._consumer.stopped.set()
+
         sleep.side_effect = trigger_inactive
 
         # Also mock the consumer, which sends the request.
@@ -234,11 +236,12 @@ def trigger_inactive(seconds):
 
 def test_maintain_leases_no_ack_ids():
     policy = create_policy()
-    policy._consumer.stopped.set()
+    policy._consumer.stopped.clear()
     with mock.patch.object(time, 'sleep', autospec=True) as sleep:
         def trigger_inactive(seconds):
             assert 0 < seconds < 10
-            policy._consumer.stopped.clear()
+            policy._consumer.stopped.set()
+
         sleep.side_effect = trigger_inactive
         policy.maintain_leases()
         sleep.assert_called()