apache/beam · error · ValueError

unknown Watch growth state tag: %r

Error message

unknown Watch growth state tag: %r

What it means

_PollingGrowthStateCoder.decode reads a tag byte to determine which subclass of _GrowthState was encoded and reconstructs it. An unknown tag means the payload does not correspond to any known state encoding — either corrupt data, a cross-version encoding mismatch, or a bug in encode/decode pairing.

Source

Thrown at sdks/python/apache_beam/io/watch.py:400

  def decode(self, encoded: bytes) -> _GrowthState:
    tag, payload = self._envelope_coder.decode(encoded)
    if tag == _StateTag.POLLING:
      termination_state, poll_watermark, items = self._polling_coder.decode(
          payload)
      return _PollingGrowthState(
          collections.OrderedDict(items), poll_watermark, termination_state)
    if tag == _StateTag.NON_POLLING:
      watermark, outputs = self._non_polling_coder.decode(payload)
      return _NonPollingGrowthState(PollResult(tuple(outputs), watermark))
    if tag == _StateTag.CURSOR_POLLING:
      termination_state, poll_watermark, items, cursor = (
          self._cursor_polling_coder.decode(payload))
      return _PollingGrowthState(
          collections.OrderedDict(items),
          poll_watermark,
          termination_state,
          cursor)
    raise ValueError('unknown Watch growth state tag: %r' % (tag, ))

  def is_deterministic(self) -> bool:
    return False


# ------------------------------------------------------------------------------
# Restriction tracker.
# ------------------------------------------------------------------------------


def _identity(value: Any) -> Any:
  return value


def _hash_output(key_coder: Coder, value: Any) -> bytes:
  return hashlib.blake2b(
      key_coder.encode(value), digest_size=_HASH_DIGEST_SIZE).digest()

View on GitHub (pinned to 12126d8942)

Solutions

1. Ensure all workers and the pipeline submission use the same Apache Beam version
2. If you added a custom _GrowthState subclass, register its tag in decode and write it in encode
3. Inspect the payload/tag byte for corruption; re-run the pipeline to regenerate state

Example fix

// before
class MyStateCoder(_PollingGrowthStateCoder):
  pass  # encode writes custom tag, decode never handles it
// after
class MyStateCoder(_PollingGrowthStateCoder):
  def encode(self, state):
    ...  # write matching tag
  def decode(self, payload):
    tag = payload[0]
    if tag == MY_TAG:
      ...  # handle it before falling through
    return super().decode(payload)

assert len(payload) > 0 and payload[0] in KNOWN_GROWTH_STATE_TAGS, 'bad growth-state tag'

try:
    state = coder.decode(payload)
except ValueError as e:
    log.error('growth state decode failed: %s', e)
    raise

When it happens

Trigger: Decoding Watch growth-state payloads produced by a different Beam version, decoding a manually corrupted/byte-modified payload, or a custom _GrowthState subclass whose coder writes a tag the decoder does not handle.

Common situations: Running pipelines across mixed Beam worker versions (runner-upgrade mid-job); hand-editing or reusing serialized coders; customizing Watch internals without updating both encode and decode paths.

Understand the failure class

Background: Invalid enum value errors: "Unknown type", "Invalid scope", "must be one of" — when a string is not on the library's allowed list — this error's family across 23 libraries.

Related errors

• Unknown type tag %x
• Expected 0 or 1, got %s
• No such coder: %s
• Unknown PaneInfo encoding 0x" + encoding.toString(16)
• Error deserializing via Coder

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