apache/beam · error · IllegalStateException

Received message missing publishTime

Error message

Received message missing publishTime

What it means

When reading via PubsubGrpcClient.pull without a custom timestampAttribute, the client relies on the server-assigned publishTime of each received message. If the PubsubMessage proto lacks publishTime, an IllegalStateException is thrown because Beam cannot assign an element timestamp.

Source

Thrown at sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubGrpcClient.java:279

            .setMaxMessages(batchSize)
            .build();
    PullResponse response = subscriberStub().pull(request);
    if (response.getReceivedMessagesCount() == 0) {
      return ImmutableList.of();
    }
    List<IncomingMessage> incomingMessages = new ArrayList<>(response.getReceivedMessagesCount());
    for (ReceivedMessage message : response.getReceivedMessagesList()) {
      if (!message.hasMessage()) {
        continue;
      }
      PubsubMessage pubsubMessage = message.getMessage();
      Map<String, String> attributes = pubsubMessage.getAttributes();

      // Timestamp.
      long timestampMsSinceEpoch;
      if (timestampAttribute == null || timestampAttribute.isEmpty()) {
        if (!pubsubMessage.hasPublishTime()) {
          throw new IllegalStateException("Received message missing publishTime");
        }
        Timestamp timestampProto = pubsubMessage.getPublishTime();
        timestampMsSinceEpoch =
            timestampProto.getSeconds() * 1000 + timestampProto.getNanos() / 1000L / 1000L;
      } else {
        timestampMsSinceEpoch = extractTimestampAttribute(timestampAttribute, attributes);
      }

      // Ack id.
      String ackId = message.getAckId();
      if (ackId.isEmpty()) {
        throw new IllegalStateException("Received message missing ackId");
      }

      // Record id, if any.
      String recordId = null;
      if (idAttribute != null) {
        recordId = attributes.get(idAttribute);

View on GitHub (pinned to 12126d8942)

Solutions

1. Use a real Cloud Pub/Sub endpoint; server publishes always set publishTime
2. Fix the emulator/test fixture to set publishTime on generated messages
3. Configure a timestampAttribute pointing to a producer-set attribute instead of relying on publishTime
4. Update the Pub/Sub client library/service so publishTime is populated

Example fix

// test fixture before
PubsubMessage.newBuilder().setData(payload).build();
// after
PubsubMessage.newBuilder().setData(payload)
    .setPublishTime(Timestamps.fromMillis(System.currentTimeMillis())).build();

if (timestampAttribute == null || timestampAttribute.isEmpty()) {
  if (!pubsubMessage.hasPublishTime()) throw new IllegalStateException("Message lacks publishTime; cannot timestamp");
}

try { messages = client.pull(path, null, 10, clock); } catch (IllegalStateException e) { log.error("Message missing publishTime: {}", e.getMessage()); }

When it happens

Trigger: PubsubIO.read (default timestamp extraction, no timestampAttribute configured) pulls a message from a subscription whose pulled ReceivedMessage contains a PubsubMessage with no publish_time field set.

Common situations: Custom/fake Pub/Sub emulators or test doubles that omit publishTime; misrouted messages from non-standard publishers; older service endpoints or proxies stripping the field.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

• Received message missing ackId
• Empty timestamp.
• Dropping output field '{}' before writing to PubSub because
• Extract file timestamp failed: got file timestamp == 0.
• micros_instant logical type encountered a Java Instant with

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