apache/beam · error · ValueError

An unsupported source was specified

Error message

An unsupported source was specified: '%s'. Please specify one of the following sources: %s

What it means

ManagedReadTransform looked up the given source identifier in its table of known managed read sources (Kafka, BigQuery, Postgres, MySQL, SQL Server, Delta Lake, etc.) and found no entry; the accepted source keys are listed in the message.

Solutions

1. Use one of the supported source keys, e.g. 'kafka', 'bigquery', 'postgres', 'mysql', 'sql_server', 'delta'
2. Check spelling/case of the source key against the list in the error message

When it happens

Trigger: Thrown at sdks/python/apache_beam/transforms/managed.py:119 when the library encounters an invalid state.

Common situations: See trigger scenarios.

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/managed.py:119

      KAFKA: ManagedTransforms.Urns.KAFKA_READ.urn,
      BIGQUERY: ManagedTransforms.Urns.BIGQUERY_READ.urn,
      POSTGRES: ManagedTransforms.Urns.POSTGRES_READ.urn,
      MYSQL: ManagedTransforms.Urns.MYSQL_READ.urn,
      SQL_SERVER: ManagedTransforms.Urns.SQL_SERVER_READ.urn,
      DELTA: ManagedTransforms.Urns.DELTA_LAKE_READ.urn,
  }

  def __init__(
      self,
      source: str,
      config: Optional[dict[str, Any]] = None,
      config_url: Optional[str] = None,
      expansion_service=None):
    super().__init__()
    self._source = source
    identifier = self._READ_TRANSFORMS.get(source.lower())
    if not identifier:
      raise ValueError(
          f"An unsupported source was specified: '{source}'. Please specify "
          f"one of the following sources: {list(self._READ_TRANSFORMS.keys())}")

    # Store parameters for deferred expansion service creation
    self._identifier = identifier
    self._provided_expansion_service = expansion_service
    self._underlying_identifier = identifier
    self._yaml_config = yaml.dump(config)
    self._config_url = config_url

  def expand(self, input):
    # Create expansion service with access to pipeline options
    expansion_service = _resolve_expansion_service(
        self._source,
        self._identifier,
        self._provided_expansion_service,
        pipeline_options=input.pipeline._options)

View on GitHub (pinned to 12126d8942)