apache/beam · error · NotImplementedError

NotImplementedError(request.artifact.type_urn)

Error message

NotImplementedError(request.artifact.type_urn)

What it means

In the artifact staging service, GetArtifact opens a read handle based on the artifact's type_urn; only FILE and EMBEDDED urns are handled, anything else raises NotImplementedError. The requested artifact type is not supported by this implementation.

Solutions

Stage the artifact as a FILE or EMBEDDED type supported by this service.
Align SDK and job-service versions so both agree on artifact type urns.
Subclass ArtifactStagingService and override GetArtifact if a custom artifact type is genuinely required.

Example fix

// before
artifact.type_urn = 'beam:artifact:type:custom:v1'
// after
artifact.type_urn = common_urns.artifact_types.FILE.urn
artifact.type_payload = beam_runner_api_pb2.FileArtifactPayload(
    path='/tmp/artifact.txt').SerializeToString()

Defensive patterns

Strategy: validation

Validate before calling

from apache_beam.portability.common_urns import artifact_types

assert artifact.type_urn in (
    artifact_types.FILE.urn, artifact_types.EMBEDDED.urn), \
    f"unsupported artifact type: {artifact.type_urn}"

Type guard

def is_supported_artifact(artifact) -> bool:
    return artifact.type_urn in (
        common_urns.artifact_types.FILE.urn,
        common_urns.artifact_types.EMBEDDED.urn)

Try / catch

try:
    for chunk in stub.GetArtifact(req):
        consume(chunk)
except NotImplementedError:
    restage_artifact_as_file(artifact)  # convert custom type to FILE

Prevention

Stage artifacts only with FILE or EMBEDDED type urns.
Keep SDK and job-service versions aligned to avoid urn skew.
Log artifact.type_urn when staging so mismatches are diagnosable.

When it happens

Trigger: Calling the GetArtifact RPC with an artifact whose type_urn is neither FILEArtifactPayload nor EmbeddedFilePayload — e.g. a URL-based or custom artifact type produced by another service.

Common situations: SDK/job-service version skew with divergent artifact type urns; artifacts staged by a third-party or custom staging service; portable pipeline cross-version mismatches.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/portability/artifact_service.py:83

        replacements=request.artifacts)

  def GetArtifact(self, request, context=None):
    if request.artifact.type_urn == common_urns.artifact_types.FILE.urn:
      payload = proto_utils.parse_Bytes(
          request.artifact.type_payload,
          beam_runner_api_pb2.ArtifactFilePayload)
      read_handle = self._file_reader(payload.path)
    elif request.artifact.type_urn == common_urns.artifact_types.URL.urn:
      payload = proto_utils.parse_Bytes(
          request.artifact.type_payload, beam_runner_api_pb2.ArtifactUrlPayload)
      read_handle = urlopen(payload.url)
    elif request.artifact.type_urn == common_urns.artifact_types.EMBEDDED.urn:
      payload = proto_utils.parse_Bytes(
          request.artifact.type_payload,
          beam_runner_api_pb2.EmbeddedFilePayload)
      read_handle = BytesIO(payload.data)
    else:
      raise NotImplementedError(request.artifact.type_urn)

    with read_handle as fin:
      while True:
        chunk = fin.read(self._chunk_size)
        if not chunk:
          break
        yield beam_artifact_api_pb2.GetArtifactResponse(data=chunk)


class ArtifactStagingService(
    beam_artifact_api_pb2_grpc.ArtifactStagingServiceServicer):
  def __init__(
      self,
      file_writer: Callable[[str, Optional[str]], tuple[BinaryIO, str]],
  ):
    self._lock = threading.Lock()
    self._jobs_to_stage: dict[
        str,

View on GitHub (pinned to 12126d8942)