From e6871b25b2e311ca35371d19f4d40f3bcaafa6f5 Mon Sep 17 00:00:00 2001
From: William Barnhart <williambbarnhart@gmail.com>
Date: Sun, 19 Jul 2026 11:02:41 +0000
Subject: [PATCH] Fix #214: skip end-offset metric when highwater is unknown
 (None)

During a rebalance Consumer.highwater(tp) can return None before the end
offset is known.  getmany() passed that straight to
monitor.track_tp_end_offset(), and the Prometheus/Datadog/StatsD sensors
do float(offset) on it, raising
'TypeError: float() argument must be a string or a number, not NoneType'
inside _drain_messages and crashing the consumer.

Guard the call so the end-offset metric is only tracked once the highwater
is known.  This fixes every metric sensor at the single source instead of
patching each one.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL
---
 faust/transport/consumer.py           |  8 ++++++-
 tests/unit/transport/test_consumer.py | 31 +++++++++++++++++++++++++++
 2 files changed, 38 insertions(+), 1 deletion(-)

diff --git a/faust/transport/consumer.py b/faust/transport/consumer.py
index d5a82b1b8..02ecaa558 100644
--- a/faust/transport/consumer.py
+++ b/faust/transport/consumer.py
@@ -748,7 +748,13 @@ async def getmany(self, timeout: float) -> AsyncIterator[Tuple[TP, Message]]:
                     or tp in self._buffered_partitions
                 ):
                     highwater_mark = self.highwater(tp)
-                    self.app.monitor.track_tp_end_offset(tp, highwater_mark)
+                    # highwater() can return None during a rebalance before
+                    # the end offset is known.  Tracking it would crash the
+                    # metric sensors that do float(offset) (Prometheus,
+                    # Datadog, StatsD) and take down _drain_messages.  Skip
+                    # until the highwater is known.  See issue #214.
+                    if highwater_mark is not None:
+                        self.app.monitor.track_tp_end_offset(tp, highwater_mark)
                     # convert timestamp to seconds from int milliseconds.
                     yield tp, to_message(tp, record)
         else:
diff --git a/tests/unit/transport/test_consumer.py b/tests/unit/transport/test_consumer.py
index 070066984..dcd73f5ec 100644
--- a/tests/unit/transport/test_consumer.py
+++ b/tests/unit/transport/test_consumer.py
@@ -528,6 +528,37 @@ def to_message(tp, record):
             (TP2, "G"),
         ]
 
+    @pytest.mark.asyncio
+    async def test_getmany__highwater_none_not_tracked(self, *, consumer):
+        # Regression test for #214: highwater() can return None during a
+        # rebalance.  Passing it to track_tp_end_offset crashes metric
+        # sensors that do float(offset), so it must be skipped.
+        def to_message(tp, record):
+            return record
+
+        consumer._to_message = to_message
+        consumer.highwater = Mock(name="highwater", return_value=None)
+        consumer.app.monitor = Mock(name="monitor")
+        self._setup_records(consumer, active_partitions={TP1}, records={TP1: ["A"]})
+        consumer.flow_active = True
+
+        assert [a async for a in consumer.getmany(1.0)] == [(TP1, "A")]
+        consumer.app.monitor.track_tp_end_offset.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_getmany__highwater_tracked(self, *, consumer):
+        def to_message(tp, record):
+            return record
+
+        consumer._to_message = to_message
+        consumer.highwater = Mock(name="highwater", return_value=42)
+        consumer.app.monitor = Mock(name="monitor")
+        self._setup_records(consumer, active_partitions={TP1}, records={TP1: ["A"]})
+        consumer.flow_active = True
+
+        assert [a async for a in consumer.getmany(1.0)] == [(TP1, "A")]
+        consumer.app.monitor.track_tp_end_offset.assert_called_once_with(TP1, 42)
+
     @pytest.mark.asyncio
     async def test_getmany_buffered(self, *, consumer):
         def to_message(tp, record):