apache/beam · error · ValueError

count (%d) must be a positive integer.

Error message

count (%d) must be a positive integer.

What it means

AfterCount.__init__ validates that its count argument is a positive integer (numbers.Integral and >= 1) and raises ValueError otherwise. The trigger fires only when a pane holds at least `count` elements, so zero, negative, or non-integer counts are meaningless.

Source

Thrown at sdks/python/apache_beam/transforms/trigger.py:686

        context) if self.early else None
    late_proto = self.late.underlying.to_runner_api(
        context) if self.late else None
    return beam_runner_api_pb2.Trigger(
        after_end_of_window=beam_runner_api_pb2.Trigger.AfterEndOfWindow(
            early_firings=early_proto, late_firings=late_proto))

  def has_ontime_pane(self):
    return True


class AfterCount(TriggerFn):
  """Fire when there are at least count elements in this window pane."""

  COUNT_TAG = _CombiningValueStateTag('count', combiners.CountCombineFn())

  def __init__(self, count):
    if not isinstance(count, numbers.Integral) or count < 1:
      raise ValueError("count (%d) must be a positive integer." % count)
    self.count = count

  def __repr__(self):
    return 'AfterCount(%s)' % self.count

  def __eq__(self, other):
    return type(self) == type(other) and self.count == other.count

  def __hash__(self):
    return hash(self.count)

  def on_element(self, element, window, context):
    context.add_state(self.COUNT_TAG, 1)

  def on_merge(self, to_be_merged, merge_result, context):
    # states automatically merged
    pass

View on GitHub (pinned to 12126d8942)

Solutions

1. Pass an int >= 1, e.g. AfterCount(1).
2. Coerce config input: AfterCount(int(value)) after checking it is a whole number and > 0.
3. If you want time-based behavior with count 0 semantics, use AfterProcessingTime or a different trigger instead.

Example fix

// before
count = float(cfg['fire_count'])  # 2.0 -> ValueError
t = AfterCount(count)
// after
count = int(cfg['fire_count'])
assert count >= 1
t = AfterCount(count)

import numbers
if not isinstance(count, numbers.Integral) or count < 1:
    raise ValueError(f'count must be a positive int, got {count!r}')

def is_positive_int(v):
    return isinstance(v, numbers.Integral) and not isinstance(v, bool) and v >= 1

try:
    trigger = AfterCount(count)
except ValueError:
    trigger = AfterCount(1)  # safe minimum

When it happens

Trigger: AfterCount(0), AfterCount(-5), or AfterCount(2.5); any call where count is a float, string, or bool-derived expression that fails the isinstance(numbers.Integral) and >= 1 checks.

Common situations: Building trigger pipelines where the count comes from user config (YAML/CLI) and was parsed as a float or string; computing count dynamically and getting 0 on empty datasets; translating from another framework where 0 means 'disabled'.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

• combine_fn must be specified.
• Invalid tag.
• document_field cannot be empty
• Expected text content in {type(item).__name__} {item.id}, go
• Expected image content in {type(item).__name__} {item.id}, g

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/a3e97fa376ba7ee3. Report an issue: GitHub.