{"record":{"id":"fa1ed182d42bdc07","repo":"apache/beam","slug":"received-message-missing-publishtime","errorCode":null,"errorMessage":"Received message missing publishTime","messagePattern":"Received message missing publishTime","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubGrpcClient.java","lineNumber":279,"sourceCode":"            .setMaxMessages(batchSize)\n            .build();\n    PullResponse response = subscriberStub().pull(request);\n    if (response.getReceivedMessagesCount() == 0) {\n      return ImmutableList.of();\n    }\n    List<IncomingMessage> incomingMessages = new ArrayList<>(response.getReceivedMessagesCount());\n    for (ReceivedMessage message : response.getReceivedMessagesList()) {\n      if (!message.hasMessage()) {\n        continue;\n      }\n      PubsubMessage pubsubMessage = message.getMessage();\n      Map<String, String> attributes = pubsubMessage.getAttributes();\n\n      // Timestamp.\n      long timestampMsSinceEpoch;\n      if (timestampAttribute == null || timestampAttribute.isEmpty()) {\n        if (!pubsubMessage.hasPublishTime()) {\n          throw new IllegalStateException(\"Received message missing publishTime\");\n        }\n        Timestamp timestampProto = pubsubMessage.getPublishTime();\n        timestampMsSinceEpoch =\n            timestampProto.getSeconds() * 1000 + timestampProto.getNanos() / 1000L / 1000L;\n      } else {\n        timestampMsSinceEpoch = extractTimestampAttribute(timestampAttribute, attributes);\n      }\n\n      // Ack id.\n      String ackId = message.getAckId();\n      if (ackId.isEmpty()) {\n        throw new IllegalStateException(\"Received message missing ackId\");\n      }\n\n      // Record id, if any.\n      String recordId = null;\n      if (idAttribute != null) {\n        recordId = attributes.get(idAttribute);","sourceCodeStart":261,"sourceCodeEnd":297,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubGrpcClient.java#L261-L297","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Use a real Cloud Pub/Sub endpoint; server publishes always set publishTime","Fix the emulator/test fixture to set publishTime on generated messages","Configure a timestampAttribute pointing to a producer-set attribute instead of relying on publishTime","Update the Pub/Sub client library/service so publishTime is populated"],"exampleFix":"// test fixture before\nPubsubMessage.newBuilder().setData(payload).build();\n// after\nPubsubMessage.newBuilder().setData(payload)\n    .setPublishTime(Timestamps.fromMillis(System.currentTimeMillis())).build();","handlingStrategy":"try-catch","validationCode":"if (timestampAttribute == null || timestampAttribute.isEmpty()) {\n  if (!pubsubMessage.hasPublishTime()) throw new IllegalStateException(\"Message lacks publishTime; cannot timestamp\");\n}","typeGuard":null,"tryCatchPattern":"try { messages = client.pull(path, null, 10, clock); } catch (IllegalStateException e) { log.error(\"Message missing publishTime: {}\", e.getMessage()); }","preventionTips":["Use the real Cloud Pub/Sub service; publishTime is always set there","Fix emulators/mocks to populate publishTime","Configure a timestampAttribute when testing with fixtures","Add fixture tests asserting publishTime presence"],"tags":["java","pubsub","grpc","timestamp","message"],"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-15T02:17:10.978Z"}