{"record":{"id":"7d3409be5e1fbf42","repo":"apache/beam","slug":"received-message-missing-ackid","errorCode":null,"errorMessage":"Received message missing ackId","messagePattern":"Received message missing ackId","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":291,"sourceCode":"      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);\n      }\n      if (recordId == null || recordId.isEmpty()) {\n        // Fall back to the Pubsub provided message id.\n        recordId = pubsubMessage.getMessageId();\n      }\n\n      incomingMessages.add(\n          IncomingMessage.of(\n              pubsubMessage, timestampMsSinceEpoch, requestTimeMsSinceEpoch, ackId, recordId));\n    }\n    return incomingMessages;\n  }","sourceCodeStart":273,"sourceCodeEnd":309,"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#L273-L309","documentation":"Each message returned by a Pub/Sub pull carries an ackId used to acknowledge/nack it. PubsubGrpcClient.pull throws IllegalStateException when the ReceivedMessage's ackId is empty, since without it the message can never be acknowledged and would be redelivered forever.","triggerScenarios":"pull() processes a ReceivedMessage whose getAckId() returns an empty string while reading from a subscription via PubsubIO.read through the gRPC client.","commonSituations":"Mocked or synthetic subscription responses in tests missing ackId; broken proxy/emulator implementations; corrupted streaming-pull responses from a non-standard backend.","solutions":["Fix the test mock/emulator to populate ackId on every ReceivedMessage","Use the official Google Pub/Sub service or current client libraries","Filter out invalid received messages before they reach Beam (proxy-side validation)","Retry the pull; a transiently corrupted response is typically replaced by a valid one"],"exampleFix":"// mock before\nReceivedMessage.newBuilder().setMessage(msg).build();\n// after\nReceivedMessage.newBuilder().setMessage(msg).setAckId(\"ack-123\").build();","handlingStrategy":"try-catch","validationCode":"if (receivedMessage.getAckId() == null || receivedMessage.getAckId().isEmpty()) {\n  throw new IllegalStateException(\"ReceivedMessage has empty ackId\");\n}","typeGuard":null,"tryCatchPattern":"try { messages = client.pull(subscription, deadline, batchSize, clock); } catch (IllegalStateException e) { log.error(\"Missing ackId: {}\", e.getMessage()); }","preventionTips":["Use official Pub/Sub client libraries and service endpoints","Populate ackId in any mocked pull responses","Never strip ackId in proxies between Beam and Pub/Sub","Log raw responses when integrating custom backends"],"tags":["java","pubsub","grpc","ackid","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"}