apache/beam · error · NotImplementedError

%s must override get_checkpoint_mark_coder() to return a Cod

Error message

%s must override get_checkpoint_mark_coder() to return a Coder for its CheckpointMark subclass.

What it means

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.

Source

Thrown at sdks/python/apache_beam/io/unbounded_source.py:260

    Contract:
      * When ``checkpoint_mark`` is ``None``, the returned reader's ``start()``
        produces the very first record of the source (or returns ``False`` if
        none yet).
      * When ``checkpoint_mark`` is not ``None``, the returned reader's
        ``start()`` produces the first record strictly after the position
        encoded by ``checkpoint_mark``. The reader must not re-deliver records
        already covered by the prior bundle.
    """
    raise NotImplementedError

  def get_checkpoint_mark_coder(self) -> Coder:
    """Returns the coder for this source's :class:`CheckpointMark` instances.

    The SDK may call this while encoding or decoding source restrictions.
    Implementations should be deterministic, side-effect free, and should not
    perform I/O.
    """
    raise NotImplementedError(
        '%s must override get_checkpoint_mark_coder() to return a Coder for '
        'its CheckpointMark subclass.' % type(self).__name__)

  def is_bounded(self) -> bool:
    # SourceBase.is_bounded raises; an unbounded source is, by definition, not.
    return False

  def default_output_coder(self) -> Coder:
    # Permissive default; override for a tighter coder.
    return coders.registry.get_coder(object)


# ------------------------------------------------------------------------------
# SDF wrapper internals: a private implementation detail of
# ReadFromUnboundedSource.
# ------------------------------------------------------------------------------

View on GitHub (pinned to 12126d8942)

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.

Example fix

// before
class MySource(UnboundedSource):
  def create_reader(self, options, checkpoint_mark):
    ...
  # get_checkpoint_mark_coder missing

// after
class MySource(UnboundedSource):
  def get_checkpoint_mark_coder(self):
    return MyCheckpointMarkCoder()  # deterministic, no I/O

Defensive patterns

Strategy: validation

Validate before calling

assert not getattr(type(source).get_checkpoint_mark_coder, '__isabstractmethod__', False), 'get_checkpoint_mark_coder not implemented'
# round-trip check
mark = source.create_reader(None, None).get_checkpoint_mark()
coder = source.get_checkpoint_mark_coder()
assert coder.decode(coder.encode(mark)) == mark

Type guard

def has_checkpoint_coder(cls) -> bool:
    fn = getattr(cls, 'get_checkpoint_mark_coder', None)
    return fn is not None and not getattr(fn, '__isabstractmethod__', True)

Try / catch

try:
  coder = source.get_checkpoint_mark_coder()
except NotImplementedError as e:
  logger.error('checkpoint coder missing: %s', e)
  raise

Prevention

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.

When it happens

Trigger: 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().

Common situations: 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.

Understand the failure class

Background: "NotImplementedError: Subclasses should override this method" / "must be implemented" — abstract method errors explained — this error's family across 40 libraries.

Related errors

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