apache/beam · error · ValueError

window_ms >= bucket_ms > 0 please

Error message

window_ms >= bucket_ms > 0 please

What it means

Beam's client-side rate tracker (used for Datastore write throttling) divides a moving time window into buckets; this only works if window_ms >= bucket_ms and bucket_ms > 0. __init__ raises ValueError when bucket_ms is zero/negative or exceeds the window, since no valid bucket layout can be computed.

Solutions

1. Swap or correct the arguments so window_ms >= bucket_ms > 0, e.g. (window_ms=60000, bucket_ms=1000).
2. Validate pipeline options before constructing the tracker; guard against 0/None defaults for bucket_ms.
3. Derive bucket_ms as window_ms divided by the desired number of buckets so the invariant holds by construction.

Example fix

// before
tracker = RateTracker(window_ms=1000, bucket_ms=2000)
// after
tracker = RateTracker(window_ms=60000, bucket_ms=1000)  # window >= bucket > 0

assert window_ms > 0 and 0 < bucket_ms <= window_ms, \
    'require window_ms >= bucket_ms > 0'

try:
    tracker = RateTracker(window_ms, bucket_ms)
except ValueError as e:
    tracker = RateTracker(window_ms=60000, bucket_ms=1000)
    logging.warning('Bad rate tracker params (%s), using defaults', e)

When it happens

Trigger: Constructing the rate tracker with (window_ms=1000, bucket_ms=2000) or bucket_ms=0 / negative values, e.g. misconfigured RateLimit or AdaptiveThrottler parameters passed from pipeline options.

Common situations: Hand-editing throttling constants for Datastore/Rio writes and swapping the two arguments, or computing bucket_ms from options that default to 0 when unset.

Understand the failure class

Background: "Invalid value" and "allowed values are" config errors: what your library rejected and how to fix it — this error's family across 41 libraries.

Related errors

• cache_root GCS bucket path is invalid.
• A schema was provided without a data format (or viceversa)…
• All PCollections must belong to the same pipeline.
• Batch size is too large! It should be smaller or equal than
• boolean cross product parameter required to explode more…

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

Appendix: source

Thrown at sdks/python/apache_beam/io/gcp/datastore/v1new/util.py:51

WRITE_BATCH_MAX_SIZE = 500
WRITE_BATCH_MAX_BYTES_SIZE = 9000000
WRITE_BATCH_MIN_SIZE = 5
WRITE_BATCH_TARGET_LATENCY_MS = 6000


class MovingSum(object):
  """Class that keeps track of a rolling window sum.

  For use in tracking recent performance of the connector.

  Intended to be similar to
  org.apache.beam.sdk.util.MovingFunction(..., Sum.ofLongs()), but for
  convenience we expose the count of entries as well so this doubles as a
  moving average tracker.
  """
  def __init__(self, window_ms, bucket_ms):
    if window_ms < bucket_ms or bucket_ms <= 0:
      raise ValueError("window_ms >= bucket_ms > 0 please")
    self._num_buckets = int(math.ceil(window_ms / bucket_ms))
    self._bucket_ms = bucket_ms
    self._Reset(now=0)  # initialize the moving window members

  def _Reset(self, now):
    self._current_index = 0  # pointer into self._buckets
    self._current_ms_since_epoch = math.floor(
        now / self._bucket_ms) * self._bucket_ms

    # _buckets is a list where each element is a list [sum, num_samples]
    # This is a circular buffer where
    # [_current_index] represents the time range
    #     [_current_ms_since_epoch, _current_ms_since_epoch+_bucket_ms)
    # [_current_index-1] represents immediatly prior time range
    #     [_current_ms_since_epoch-_bucket_ms, _current_ms_since_epoch)
    # etc, wrapping around from the start to the end of the array, so
    # [_current_index+1] is the element representing the oldest bucket.
    self._buckets = [[0, 0] for _ in range(0, self._num_buckets)]

View on GitHub (pinned to 12126d8942)