What causes "Unknown annotation type %r (type %s) for %s" in apache/beam?

Convert the annotation value to a str or bytes before attaching (e.g. json.dumps(payload))

apache/beam · error · TypeError

Unknown annotation type %r (type %s) for %s

Error message

Unknown annotation type %r (type %s) for %s

What it means

TypeError raised in encode_annotations' annotation_to_bytes helper when an annotation value is not bytes, an ascii-encodable str, or a protobuf Message. Pipeline annotations must be serializable to bytes for the runner-api proto.

Source

Thrown at sdks/python/apache_beam/pipeline.py:1635

    if self.resource_hints:
      for part in self.parts:
        part._merge_outer_resource_hints()


def encode_annotations(annotations: Optional[dict[str, Any]]):
  """Encodes non-byte annotation values as bytes."""
  if not annotations:
    return {}

  def annotation_to_bytes(key, a: Any) -> bytes:
    if isinstance(a, bytes):
      return a
    elif isinstance(a, str):
      return a.encode('ascii')
    elif isinstance(a, message.Message):
      return a.SerializeToString()
    else:
      raise TypeError(
          'Unknown annotation type %r (type %s) for %s' % (a, type(a), key))

  return {key: annotation_to_bytes(key, a) for (key, a) in annotations.items()}


_global_annotations_stack_data = threading.local()


def _global_annotations_stack():
  try:
    return _global_annotations_stack_data.stack
  except AttributeError:
    _global_annotations_stack_data.stack = [{}]
    return _global_annotations_stack_data.stack


@contextlib.contextmanager
def transform_annotations(**annotations):

View on GitHub (pinned to 12126d8942)

Solutions

Convert the annotation value to a str or bytes before attaching (e.g. json.dumps(payload))
Use a protobuf message if you need structured annotations
Serialize custom objects to bytes yourself (with an agreed encoding on the consumer side)

Example fix

// before
pipeline.annotations['meta'] = {'owner': 'team'}
// after
import json
pipeline.annotations['meta'] = json.dumps({'owner': 'team'})

Defensive patterns

Strategy: validation

Validate before calling

from google.protobuf.message import Message
for k, v in annotations.items():
    assert isinstance(v, (bytes, str, Message)), f'Bad annotation {k}: {type(v)}'

Type guard

from google.protobuf.message import Message

def is_valid_annotation(v):
    return isinstance(v, (bytes, str, Message))

Try / catch

try:
    encoded = pipeline.annotations
except TypeError as e:
    logging.error('Unsupported annotation type: %s', e)

Prevention

Store annotations only as str, bytes, or protobuf Messages
Use json.dumps for structured metadata
Document the annotation schema for custom runner consumers

When it happens

Trigger: Setting pipeline/transform annotations (via annotations param or with_annotations) to dict/int/list objects instead of str, bytes, or protobuf Message.

Common situations: Attaching structured metadata (JSON dicts, numbers) to pipeline annotations for custom runners; passing numpy values.

Related errors

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