Raise timeout for unfinished producer futures at deadline

kafka/producer/future.py

@@ -100,8 +100,8 @@ def get(self, timeout=None):
         # multiple times, so each rebind wakes us and we re-wait on the
         # (possibly new) _produce_future.
         timer = Timer(timeout * 1000 if timeout is not None else None)
-        while not self.is_done and not timer.expired:
-            if not self._produce_future.wait(timer.timeout_secs):
+        while not self.is_done:
+            if timer.expired or not self._produce_future.wait(timer.timeout_secs):
                 raise Errors.KafkaTimeoutError(
                     "Timeout after waiting for %s secs." % (timeout,))
         if self.failed():

test/producer/test_sender.py

@@ -925,6 +925,25 @@ def get_in_thread():
         assert result_holder[0] is not None
         assert result_holder[0].offset == 42
 
+    def test_get_times_out_when_rebind_wakes_at_deadline(self, mocker):
+        from kafka.producer.future import FutureProduceResult
+
+        tp = TopicPartition('foo', 0)
+        old_pf = FutureProduceResult(tp)
+        new_pf = FutureProduceResult(tp)
+        future = FutureRecordMetadata(old_pf, 0, 1000, None, 3, 5, -1)
+        clock = mocker.patch('kafka.util.time.monotonic', return_value=100.0)
+
+        def rebind_at_deadline(timeout):
+            future.rebind(new_pf, 0)
+            clock.return_value += timeout
+            return True
+
+        mocker.patch.object(FutureProduceResult, 'wait', side_effect=rebind_at_deadline)
+        with pytest.raises(Errors.KafkaTimeoutError):
+            future.get(timeout=1)
+        assert not future.is_done
+
     def test_get_rewait_after_multiple_rebinds(self):
         """get() survives multiple rebinds (batch split more than once)."""
         import threading
@@ -1585,3 +1604,37 @@ def on_produce(api_key, api_version, correlation_id, request_bytes):
             assert captured.get('acks') == 0
         finally:
             producer.close(timeout=2)
+
+    @pytest.mark.parametrize("timeout", [0, 0.01])
+    @pytest.mark.parametrize("error_code", [0, Errors.TopicAuthorizationFailedError.errno])
+    def test_pending_send_get_times_out_without_completing(self, timeout, error_code):
+        broker = MockBroker(broker_version=self._API_VERSION)
+        broker.set_metadata(topics=[self._metadata_topic()])
+        broker.respond(ProduceRequest, self._produce_response(
+            self._API_VERSION, error_code=error_code))
+
+        producer = KafkaProducer(
+            kafka_client=broker.client_factory(),
+            bootstrap_servers=['%s:%d' % (broker.host, broker.port)],
+            api_version=self._API_VERSION,
+            enable_idempotence=False,
+            acks=1,
+            linger_ms=60000,
+            request_timeout_ms=5000,
+        )
+        try:
+            future = producer.send(self._TOPIC, value=b'payload', partition=0)
+            with pytest.raises(Errors.KafkaTimeoutError):
+                future.get(timeout=timeout)
+            assert not future.is_done
+
+            # A get timeout must leave the send pending. Flush the lingered
+            # batch, then observe its actual delivery result, even at zero timeout.
+            producer.flush(timeout=5)
+            if error_code:
+                with pytest.raises(Errors.TopicAuthorizationFailedError):
+                    future.get(timeout=0)
+            else:
+                assert future.get(timeout=0).offset == self._BASE_OFFSET
+        finally:
+            producer.close(timeout=2)