{"record":{"id":"5d532e22ce8d0adc","repo":"apache/beam","slug":"window-ms-bucket-ms-0-please-util","errorCode":null,"errorMessage":"window_ms >= bucket_ms > 0 please","messagePattern":"window_ms >= bucket_ms > 0 please","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/gcp/datastore/v1new/util.py","lineNumber":51,"sourceCode":"WRITE_BATCH_MAX_SIZE = 500\nWRITE_BATCH_MAX_BYTES_SIZE = 9000000\nWRITE_BATCH_MIN_SIZE = 5\nWRITE_BATCH_TARGET_LATENCY_MS = 6000\n\n\nclass MovingSum(object):\n  \"\"\"Class that keeps track of a rolling window sum.\n\n  For use in tracking recent performance of the connector.\n\n  Intended to be similar to\n  org.apache.beam.sdk.util.MovingFunction(..., Sum.ofLongs()), but for\n  convenience we expose the count of entries as well so this doubles as a\n  moving average tracker.\n  \"\"\"\n  def __init__(self, window_ms, bucket_ms):\n    if window_ms < bucket_ms or bucket_ms <= 0:\n      raise ValueError(\"window_ms >= bucket_ms > 0 please\")\n    self._num_buckets = int(math.ceil(window_ms / bucket_ms))\n    self._bucket_ms = bucket_ms\n    self._Reset(now=0)  # initialize the moving window members\n\n  def _Reset(self, now):\n    self._current_index = 0  # pointer into self._buckets\n    self._current_ms_since_epoch = math.floor(\n        now / self._bucket_ms) * self._bucket_ms\n\n    # _buckets is a list where each element is a list [sum, num_samples]\n    # This is a circular buffer where\n    # [_current_index] represents the time range\n    #     [_current_ms_since_epoch, _current_ms_since_epoch+_bucket_ms)\n    # [_current_index-1] represents immediatly prior time range\n    #     [_current_ms_since_epoch-_bucket_ms, _current_ms_since_epoch)\n    # etc, wrapping around from the start to the end of the array, so\n    # [_current_index+1] is the element representing the oldest bucket.\n    self._buckets = [[0, 0] for _ in range(0, self._num_buckets)]","sourceCodeStart":33,"sourceCodeEnd":69,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/gcp/datastore/v1new/util.py#L33-L69","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Swap or correct the arguments so window_ms >= bucket_ms > 0, e.g. (window_ms=60000, bucket_ms=1000).","Validate pipeline options before constructing the tracker; guard against 0/None defaults for bucket_ms.","Derive bucket_ms as window_ms divided by the desired number of buckets so the invariant holds by construction."],"exampleFix":"// before\ntracker = RateTracker(window_ms=1000, bucket_ms=2000)\n// after\ntracker = RateTracker(window_ms=60000, bucket_ms=1000)  # window >= bucket > 0","handlingStrategy":"validation","validationCode":"assert window_ms > 0 and 0 < bucket_ms <= window_ms, \\\n    'require window_ms >= bucket_ms > 0'","typeGuard":null,"tryCatchPattern":"try:\n    tracker = RateTracker(window_ms, bucket_ms)\nexcept ValueError as e:\n    tracker = RateTracker(window_ms=60000, bucket_ms=1000)\n    logging.warning('Bad rate tracker params (%s), using defaults', e)","preventionTips":["Double-check argument order (window first, bucket second)","Derive bucket_ms from window_ms / num_buckets","Validate throttling options loaded from pipeline flags"],"tags":["value-error","configuration","apache-beam"],"backgroundTag":"invalid-config-value","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}