{"record":{"id":"db679d5c7a8f1bd3","repo":"apache/beam","slug":"no-publish-time-present-in-message-s","errorCode":null,"errorMessage":"No publish time present in message: %s","messagePattern":"No publish time present in message: (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/direct/transform_evaluator.py","lineNumber":734,"sourceCode":"      parsed_message = PubsubMessage._from_message(message)\n      if (timestamp_attribute and\n          timestamp_attribute in parsed_message.attributes):\n        rfc3339_or_milli = parsed_message.attributes[timestamp_attribute]\n        try:\n          timestamp = Timestamp(micros=int(rfc3339_or_milli) * 1000)\n        except ValueError:\n          try:\n            timestamp = Timestamp.from_rfc3339(rfc3339_or_milli)\n          except ValueError as e:\n            raise ValueError('Bad timestamp value: %s' % e)\n        if timestamp.precision() > Timestamp.MICROS_PRECISION:\n          # Element timestamps are limited to microsecond resolution, so\n          # ignore sub-microsecond digits, as the Dataflow service does.\n          timestamp = timestamp.to_precision(\n              Timestamp.MICROS_PRECISION, allow_lossy_conversion=True)\n      else:\n        if message.publish_time is None:\n          raise ValueError('No publish time present in message: %s' % message)\n        try:\n          timestamp = Timestamp.from_utc_datetime(message.publish_time)\n        except ValueError as e:\n          raise ValueError('Bad timestamp value for message %s: %s', message, e)\n\n      return timestamp, parsed_message\n\n    # Because of the AutoAck, we are not able to reread messages if this\n    # evaluator fails with an exception before emitting a bundle. However,\n    # the DirectRunner currently doesn't retry work items anyway, so the\n    # pipeline would enter an inconsistent state on any error.\n    sub_client = self._get_subscriber_client(self._applied_ptransform)\n    response = sub_client.pull(\n        subscription=self._sub_name, max_messages=10, timeout=30)\n    results = [_get_element(rm.message) for rm in response.received_messages]\n    ack_ids = [rm.ack_id for rm in response.received_messages]\n    if ack_ids:\n      sub_client.acknowledge(subscription=self._sub_name, ack_ids=ack_ids)","sourceCodeStart":716,"sourceCodeEnd":752,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/direct/transform_evaluator.py#L716-L752","documentation":"When no timestamp_attribute is configured, the DirectRunner derives element timestamps from the Pub/Sub message's server-set publish_time. If the message carries no publish time, the evaluator raises ValueError('No publish time present in message').","triggerScenarios":"Reading Pub/Sub messages (via the client-based path) whose PubsubMessage.publish_time is None — typically messages published through custom/partial clients or deserialized payloads lacking the field — without a timestamp_attribute fallback.","commonSituations":"Using google-cloud-pubsub clients or mocks that don't set publish_time; test harnesses constructing PubsubMessage manually; protocol-buffer messages built before publish, before the server stamps the time.","solutions":["Set timestamp_attribute in ReadFromPubSub to a publisher-provided attribute carrying the event time.","Ensure messages come from the real Pub/Sub service, which always stamps publish_time.","In tests/mocks, populate PubsubMessage(publish_time=...) before feeding the evaluator.","Upgrade apache-beam if your client library stopped populating publish_time."],"exampleFix":"// before\nmsg = PubsubMessage(data, attributes)  # publish_time missing\n// after\nmsg = PubsubMessage(data, attributes)\nmsg.publish_time = datetime.now(timezone.utc)","handlingStrategy":"fallback","validationCode":"if msg.publish_time is None and 'event_time' not in msg.attributes:\n    raise ValueError('message lacks publish_time and no timestamp_attribute fallback')","typeGuard":"def has_usable_timestamp(msg, ts_attr=None) -> bool:\n    return (ts_attr is not None and ts_attr in msg.attributes) or msg.publish_time is not None","tryCatchPattern":"try:\n    read_pubsub()\nexcept ValueError as e:\n    if 'No publish time present' in str(e):\n        switch_to_timestamp_attribute_source()","preventionTips":["Always set timestamp_attribute so timestamps come from your own data.","Test with real Pub/Sub service messages, not hand-built mocks.","Populate publish_time in test fixtures."],"tags":["python","apache-beam","pubsub","timestamp"],"backgroundTag":"missing-required-argument","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T21:17:11.552Z"}