{"record":{"id":"317a29403fdefd2b","repo":"apache/beam","slug":"s-must-override-get-checkpoint-mark-coder-to-return-a-coder","errorCode":null,"errorMessage":"%s must override get_checkpoint_mark_coder() to return a Coder for its CheckpointMark subclass.","messagePattern":"(.+?) must override get_checkpoint_mark_coder\\(\\) to return a Coder for its CheckpointMark subclass\\.","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/unbounded_source.py","lineNumber":260,"sourceCode":"    Contract:\n      * When ``checkpoint_mark`` is ``None``, the returned reader's ``start()``\n        produces the very first record of the source (or returns ``False`` if\n        none yet).\n      * When ``checkpoint_mark`` is not ``None``, the returned reader's\n        ``start()`` produces the first record strictly after the position\n        encoded by ``checkpoint_mark``. The reader must not re-deliver records\n        already covered by the prior bundle.\n    \"\"\"\n    raise NotImplementedError\n\n  def get_checkpoint_mark_coder(self) -> Coder:\n    \"\"\"Returns the coder for this source's :class:`CheckpointMark` instances.\n\n    The SDK may call this while encoding or decoding source restrictions.\n    Implementations should be deterministic, side-effect free, and should not\n    perform I/O.\n    \"\"\"\n    raise NotImplementedError(\n        '%s must override get_checkpoint_mark_coder() to return a Coder for '\n        'its CheckpointMark subclass.' % type(self).__name__)\n\n  def is_bounded(self) -> bool:\n    # SourceBase.is_bounded raises; an unbounded source is, by definition, not.\n    return False\n\n  def default_output_coder(self) -> Coder:\n    # Permissive default; override for a tighter coder.\n    return coders.registry.get_coder(object)\n\n\n# ------------------------------------------------------------------------------\n# SDF wrapper internals: a private implementation detail of\n# ReadFromUnboundedSource.\n# ------------------------------------------------------------------------------\n\n","sourceCodeStart":242,"sourceCodeEnd":278,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/unbounded_source.py#L242-L278","documentation":"UnboundedSource.get_checkpoint_mark_coder() in apache_beam/io/unbounded_source.py raises NotImplementedError with a descriptive message ('%s must override get_checkpoint_mark_coder() ...') because, unlike the other interface stubs, the SDK requires an explicit Coder for CheckpointMark instances and cannot supply a default. It is invoked (via _checkpoint_coder) while encoding or decoding source restrictions/checkpoints.","triggerScenarios":"Encoding a checkpoint mark or source restriction (e.g. during checkpointing, bundle finalization, or pipeline serialization) on an UnboundedSource subclass that has not overridden get_checkpoint_mark_coder().","commonSituations":"Custom streaming sources where the author implemented checkpoint marks but assumed a default coder exists; running a source that never needed checkpointing locally on a runner that serializes checkpoints; upgrades to code paths calling _checkpoint_coder.","solutions":["Override get_checkpoint_mark_coder() in your source to return a deterministic Coder for your CheckpointMark subclass (e.g. a custom Coder or coders.registry.get_coder).","Ensure the returned coder performs no I/O and is side-effect free, per the interface contract.","Test round-trip encode/decode of your CheckpointMark with the returned coder before running on a checkpointing runner."],"exampleFix":"// before\nclass MySource(UnboundedSource):\n  def create_reader(self, options, checkpoint_mark):\n    ...\n  # get_checkpoint_mark_coder missing\n\n// after\nclass MySource(UnboundedSource):\n  def get_checkpoint_mark_coder(self):\n    return MyCheckpointMarkCoder()  # deterministic, no I/O\n","handlingStrategy":"validation","validationCode":"assert not getattr(type(source).get_checkpoint_mark_coder, '__isabstractmethod__', False), 'get_checkpoint_mark_coder not implemented'\n# round-trip check\nmark = source.create_reader(None, None).get_checkpoint_mark()\ncoder = source.get_checkpoint_mark_coder()\nassert coder.decode(coder.encode(mark)) == mark","typeGuard":"def has_checkpoint_coder(cls) -> bool:\n    fn = getattr(cls, 'get_checkpoint_mark_coder', None)\n    return fn is not None and not getattr(fn, '__isabstractmethod__', True)","tryCatchPattern":"try:\n  coder = source.get_checkpoint_mark_coder()\nexcept NotImplementedError as e:\n  logger.error('checkpoint coder missing: %s', e)\n  raise","preventionTips":["Implement get_checkpoint_mark_coder() whenever you define a CheckpointMark subclass.","Keep the coder deterministic and free of I/O, per the interface contract.","Add a round-trip encode/decode unit test for the checkpoint mark and its coder."],"tags":["apache-beam","python","abstract-method","streaming-source","checkpointing","coder"],"backgroundTag":"abstract-method-not-implemented","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}