{"record":{"id":"ba9fa134bbb8fd57","repo":"apache/beam","slug":"values-given-to-streaming-cache-should-be-either","errorCode":null,"errorMessage":"Values given to streaming cache should be either TestStreamFileHeader or TestStreamFileRecord.","messagePattern":"Values given to streaming cache should be either TestStreamFileHeader or TestStreamFileRecord\\.","errorType":"validation","errorClass":"TypeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py","lineNumber":353,"sourceCode":"    ]\n    headers = [next(r) for r in readers]\n    return StreamingCache.Reader(headers, readers).read()\n\n  def write(self, values, *labels):\n    \"\"\"Writes the given values to cache.\n    \"\"\"\n    directory = os.path.join(self._cache_dir, *labels[:-1])\n    filepath = os.path.join(directory, labels[-1])\n    if not os.path.exists(directory):\n      os.makedirs(directory)\n    with open(filepath, 'ab') as f:\n      for v in values:\n        if isinstance(v,\n                      (beam_interactive_api_pb2.TestStreamFileHeader,\n                       beam_interactive_api_pb2.TestStreamFileRecord)):\n          val = v.SerializeToString()\n        else:\n          raise TypeError(\n              'Values given to streaming cache should be either '\n              'TestStreamFileHeader or TestStreamFileRecord.')\n        f.write(self.load_pcoder(*labels).encode(val) + b'\\n')\n\n  def clear(self, *labels):\n    directory = os.path.join(self._cache_dir, *labels[:-1])\n    filepath = os.path.join(directory, labels[-1])\n    self._capture_keys.discard(labels[-1])\n    if os.path.exists(filepath):\n      os.remove(filepath)\n      return True\n    return False\n\n  def source(self, *labels):\n    \"\"\"Returns the StreamingCacheManager source.\n\n    This is beam.Impulse() because unbounded sources will be marked with this\n    and then the PipelineInstrument will replace these with a TestStream.","sourceCodeStart":335,"sourceCodeEnd":371,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py#L335-L371","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":"// before\ncache.write(*labels, values=[element])\n// after\nfrom apache_beam.portability.api import beam_interactive_api_pb2\nrecord = beam_interactive_api_pb2.TestStreamFileRecord(record=...)\ncache.write(*labels, values=[record])","handlingStrategy":"type-guard","validationCode":"from apache_beam.portability.api import beam_interactive_api_pb2\nassert all(isinstance(v, (beam_interactive_api_pb2.TestStreamFileHeader,\n                          beam_interactive_api_pb2.TestStreamFileRecord)) for v in values)","typeGuard":"def is_stream_cache_value(v):\n    from apache_beam.portability.api import beam_interactive_api_pb2\n    return isinstance(v, (beam_interactive_api_pb2.TestStreamFileHeader,\n                          beam_interactive_api_pb2.TestStreamFileRecord))","tryCatchPattern":"try:\n    cache.write(*labels, values=values)\nexcept TypeError as e:\n    if 'TestStreamFileHeader' in str(e):\n        values = [to_test_stream_record(v) for v in values]\n        cache.write(*labels, values=values)","preventionTips":["Only feed StreamingCache through the interactive TestStream/recording pipeline.","Serialize custom events into beam_interactive_api_pb2 messages before writing."],"tags":["apache-beam","interactive-beam","cache","type-error","protobuf"],"backgroundTag":"type-mismatch","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"}