This is an automated email from the ASF dual-hosted git repository.
Abacn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 8364da70a4e Handle empty batches in Python worker counters (#38748)
8364da70a4e is described below
commit 8364da70a4e4d29f5110f1f9102406822585eef9
Author: Goutam Adwant <[email protected]>
AuthorDate: Sat May 30 12:10:37 2026 -0700
Handle empty batches in Python worker counters (#38748)
* Handle empty batches in Python worker counters
* Apply suggestion from @gemini-code-assist[bot]
Co-authored-by: gemini-code-assist[bot]
<176961590+gemini-code-assist[bot]@users.noreply.github.com>
---------
Co-authored-by: gemini-code-assist[bot]
<176961590+gemini-code-assist[bot]@users.noreply.github.com>
---
sdks/python/apache_beam/runners/worker/opcounters.py | 2 ++
.../python/apache_beam/runners/worker/opcounters_test.py | 16 ++++++++++++++++
2 files changed, 18 insertions(+)
diff --git a/sdks/python/apache_beam/runners/worker/opcounters.py
b/sdks/python/apache_beam/runners/worker/opcounters.py
index 66a369ba3bb..6f5bb60762c 100644
--- a/sdks/python/apache_beam/runners/worker/opcounters.py
+++ b/sdks/python/apache_beam/runners/worker/opcounters.py
@@ -217,6 +217,8 @@ class OperationCounters(object):
batch_length = self.producer_batch_converter.get_length(
windowed_batch.values)
self.element_counter.update(batch_length)
+ if batch_length == 0:
+ return
mean_element_size = self.producer_batch_converter.estimate_byte_size(
windowed_batch.values) / batch_length
diff --git a/sdks/python/apache_beam/runners/worker/opcounters_test.py
b/sdks/python/apache_beam/runners/worker/opcounters_test.py
index 83b345f47c9..aef66a725f9 100644
--- a/sdks/python/apache_beam/runners/worker/opcounters_test.py
+++ b/sdks/python/apache_beam/runners/worker/opcounters_test.py
@@ -184,6 +184,22 @@ class OperationCountersTest(unittest.TestCase):
self.verify_counters(opcounts, 200, size_per_element)
+ def test_update_empty_batch(self):
+ opcounts = OperationCounters(
+ CounterFactory(),
+ 'some-name',
+ coders.FastPrimitivesCoder(),
+ 0,
+ producer_batch_converter=typehints.batch.BatchConverter.from_typehints(
+ element_type=typehints.Any,
+ batch_type=typehints.List[typehints.Any]))
+
+ self.verify_counters(opcounts, 0, math.nan)
+
+ opcounts.update_from_batch(GlobalWindows.windowed_batch([]))
+
+ self.verify_counters(opcounts, 0, math.nan)
+
def test_should_sample(self):
# Order of magnitude more buckets than highest constant in code under test.
buckets = [0] * 300
