apache/beam · error · ValueError

Expected apache_beam.utils.timestamp.Timestamp, or google.pr

Error message

Expected apache_beam.utils.timestamp.Timestamp, or google.protobuf.timestamp_pb2.Timestamp. Got %s

What it means

make_state_event converts a job-state timestamp into a proto Timestamp for a beam_job_api_pb2.JobStateEvent. The service raises ValueError when the timestamp is neither an apache_beam.utils.timestamp.Timestamp nor a google.protobuf.timestamp_pb2.Timestamp, because no other type can be written to the proto field.

Source

Thrown at sdks/python/apache_beam/runners/portability/abstract_job_service.py:57

from apache_beam.portability.api import beam_job_api_pb2
from apache_beam.portability.api import beam_job_api_pb2_grpc
from apache_beam.portability.api import beam_runner_api_pb2
from apache_beam.portability.api import endpoints_pb2
from apache_beam.runners.portability import artifact_service
from apache_beam.utils.timestamp import Timestamp

_LOGGER = logging.getLogger(__name__)

StateEvent = tuple[int, Union[timestamp_pb2.Timestamp, Timestamp]]


def make_state_event(state, timestamp):
  if isinstance(timestamp, Timestamp):
    proto_timestamp = timestamp.to_proto()
  elif isinstance(timestamp, timestamp_pb2.Timestamp):
    proto_timestamp = timestamp
  else:
    raise ValueError(
        "Expected apache_beam.utils.timestamp.Timestamp, "
        "or google.protobuf.timestamp_pb2.Timestamp. "
        "Got %s" % type(timestamp))

  return beam_job_api_pb2.JobStateEvent(state=state, timestamp=proto_timestamp)


class AbstractJobServiceServicer(beam_job_api_pb2_grpc.JobServiceServicer):
  """Manages one or more pipelines, possibly concurrently.
  Experimental: No backward compatibility guaranteed.
  Servicer for the Beam Job API.
  """
  def __init__(self):
    self._jobs: dict[str, AbstractBeamJob] = {}

  def create_beam_job(
      self,
      preparation_id,  # stype: str

View on GitHub (pinned to 12126d8942)

Solutions

Coerce the value before it reaches the service: apache_beam.utils.timestamp.Timestamp.of(value) for epoch/seconds/datetime inputs.
For datetime objects, convert explicitly with timestamp_pb2.Timestamp().FromDatetime(dt).
Fix the custom job class so get_state_stream()/get_message_stream() yield Timestamp or timestamp_pb2.Timestamp.

Example fix

// before
yield state, datetime.datetime.utcnow()
// after
from apache_beam.utils.timestamp import Timestamp
yield state, Timestamp.now()

Defensive patterns

Strategy: validation

Validate before calling

from apache_beam.utils.timestamp import Timestamp
from google.protobuf import timestamp_pb2

def valid_job_timestamp(ts):
    return isinstance(ts, (Timestamp, timestamp_pb2.Timestamp))

assert valid_job_timestamp(my_ts), "convert with Timestamp.of(ts) first"

Type guard

def is_beam_timestamp(ts) -> bool:
    return isinstance(ts, (Timestamp, timestamp_pb2.Timestamp))

Try / catch

try:
    event = make_state_event(state, ts)
except (ValueError, TypeError):
    event = make_state_event(state, Timestamp.of(ts))  # coerce datetime/epoch

Prevention

Always yield Timestamp or timestamp_pb2.Timestamp from custom job get_state_stream/get_message_stream.
Coerce datetimes/epoch numbers at the boundary with Timestamp.of().
Assert the timestamp type in your job class before yielding.

When it happens

Trigger: Calling make_state_event (directly, or indirectly via GetState/GetStateStream/GetMessageStream backed by a custom AbstractBeamJob whose get_state_stream()/get_message_stream() yield timestamps) with any other type, e.g. datetime.datetime, int/float epoch seconds, or a string.

Common situations: Implementing a custom Beam job class that yields datetime objects or epoch numbers instead of the two supported Timestamp types; assuming datetimes are auto-converted.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

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