apache/beam · error · ValueError

_not_found_err_message(self.feature_store_id…

Error message

_not_found_err_message(self.feature_store_id, self.entity_type_id, entity_id)

What it means

In the legacy handler's __call__, a NotFound from the ReadFeatureValues RPC (entity id absent from the EntityType) is converted to ValueError with _not_found_err_message(feature_store_id, entity_type_id, entity_id).

Solutions

Use exception_level=ExceptionLevel.WARN to tolerate missing entities.
Ingest the missing entity into the EntityType or fix stale keys upstream.
Verify feature_store_id/entity_type_id point at the intended resource.

Example fix

// before
VertexAIEntityTypeEnrichment('fs', 'et', 'user_id')
// after
VertexAIEntityTypeEnrichment('fs', 'et', 'user_id', exception_level=ExceptionLevel.WARN)

Defensive patterns

Strategy: try-catch

Validate before calling

handler = LegacyHandler('fs', 'et', 'user_id', exception_level=ExceptionLevel.WARN)

Try / catch

try:
    result = rows | Enrichment(handler)
except ValueError as e:
    logging.warning('Entity not found: %s', e)

Prevention

Use exception_level=WARN for partial-coverage key sets.
Keep entity ingestion pipelines ahead of enrichment consumers.

When it happens

Trigger: read_feature_values raised NotFound because entity_id does not exist in the entity_type_path while exception_level is RAISE.

Common situations: Entity never ingested, deleted entity, or rows keyed against the wrong EntityType/store after refactoring.

Understand the failure class

Background: "Not found" and "does not exist" errors: why "Task not found", "No such folder", and "Can't find" fire when a lookup comes back empty — this error's family across 14 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/enrichment_handlers/vertex_ai_feature_store.py:302

      entity_id = request._asdict()[self.row_key]
    except KeyError:
      raise KeyError(
          "Enrichment requests to Vertex AI Feature Store should "
          "contain a field: %s in the input `beam.Row` to join "
          "the input with fetched response. This is used as the "
          "`FeatureViewDataKey` to fetch feature values "
          "corresponding to this key." % self.row_key)

    try:
      selector = aiplatform.gapic.FeatureSelector(
          id_matcher=aiplatform.gapic.IdMatcher(ids=self.feature_ids))
      response = self.client.read_feature_values(
          request=aiplatform.gapic.ReadFeatureValuesRequest(
              entity_type=self.entity_type_path,
              entity_id=entity_id,
              feature_selector=selector))
    except NotFound:
      raise ValueError(
          _not_found_err_message(
              self.feature_store_id, self.entity_type_id, entity_id))

    response_dict = {}
    proto_to_dict = proto.Message.to_dict(response.entity_view)
    for key, msg in zip(response.header.feature_descriptors,
                        proto_to_dict['data']):
      if msg and 'value' in msg:
        response_dict[key.id] = list(msg['value'].values())[0]
        # skip fetching the metadata
      elif self.exception_level == ExceptionLevel.RAISE:
        raise ValueError(
            _not_found_err_message(
                self.feature_store_id, self.entity_type_id, entity_id))
      elif self.exception_level == ExceptionLevel.WARN:
        _LOGGER.warning(
            _not_found_err_message(
                self.feature_store_id, self.entity_type_id, entity_id))

View on GitHub (pinned to 12126d8942)