apache/beam · error · TypeError

Values given to streaming cache should be either…

Error message

Values given to streaming cache should be either TestStreamFileHeader or TestStreamFileRecord.

What it means

StreamingCache.write only accepts protobuf messages of type TestStreamFileHeader or TestStreamFileRecord, since cache files are sequences of serialized TestStream events. Any other value type raises TypeError.

Solutions

Wrap events in beam_interactive_api_pb2.TestStreamFileHeader / TestStreamFileRecord before writing.
Convert raw TestStream elements using the interactive TestStream service/protobuf builders so each value is one of the two accepted types.
Use the StreamingCache's public recording path (background caching job) instead of calling write() directly.

Example fix

// before
cache.write(*labels, values=[element])
// after
from apache_beam.portability.api import beam_interactive_api_pb2
record = beam_interactive_api_pb2.TestStreamFileRecord(record=...)
cache.write(*labels, values=[record])

Defensive patterns

Strategy: type-guard

Validate before calling

from apache_beam.portability.api import beam_interactive_api_pb2
assert all(isinstance(v, (beam_interactive_api_pb2.TestStreamFileHeader,
                          beam_interactive_api_pb2.TestStreamFileRecord)) for v in values)

Type guard

def is_stream_cache_value(v):
    from apache_beam.portability.api import beam_interactive_api_pb2
    return isinstance(v, (beam_interactive_api_pb2.TestStreamFileHeader,
                          beam_interactive_api_pb2.TestStreamFileRecord))

Try / catch

try:
    cache.write(*labels, values=values)
except TypeError as e:
    if 'TestStreamFileHeader' in str(e):
        values = [to_test_stream_record(v) for v in values]
        cache.write(*labels, values=values)

Prevention

Only feed StreamingCache through the interactive TestStream/recording pipeline.
Serialize custom events into beam_interactive_api_pb2 messages before writing.

When it happens

Trigger: Calling write(labels, values) with raw python objects, strings, or different protobuf types instead of beam_interactive_api_pb2.TestStreamFileHeader/TestStreamFileRecord instances; custom cache writers feeding unserialized events.

Common situations: Hand-rolling interactive cache population for tests; writing TestStream events without converting to the interactive API pb2 messages; mixing batch-style cache writers with streaming cache files.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py:353

    ]
    headers = [next(r) for r in readers]
    return StreamingCache.Reader(headers, readers).read()

  def write(self, values, *labels):
    """Writes the given values to cache.
    """
    directory = os.path.join(self._cache_dir, *labels[:-1])
    filepath = os.path.join(directory, labels[-1])
    if not os.path.exists(directory):
      os.makedirs(directory)
    with open(filepath, 'ab') as f:
      for v in values:
        if isinstance(v,
                      (beam_interactive_api_pb2.TestStreamFileHeader,
                       beam_interactive_api_pb2.TestStreamFileRecord)):
          val = v.SerializeToString()
        else:
          raise TypeError(
              'Values given to streaming cache should be either '
              'TestStreamFileHeader or TestStreamFileRecord.')
        f.write(self.load_pcoder(*labels).encode(val) + b'\n')

  def clear(self, *labels):
    directory = os.path.join(self._cache_dir, *labels[:-1])
    filepath = os.path.join(directory, labels[-1])
    self._capture_keys.discard(labels[-1])
    if os.path.exists(filepath):
      os.remove(filepath)
      return True
    return False

  def source(self, *labels):
    """Returns the StreamingCacheManager source.

    This is beam.Impulse() because unbounded sources will be marked with this
    and then the PipelineInstrument will replace these with a TestStream.

View on GitHub (pinned to 12126d8942)