{"record":{"id":"ccaea9cbdf430738","repo":"apache/beam","slug":"rabbitmqio-read-uses-message-correlation-id-but-received","errorCode":null,"errorMessage":"RabbitMqIO.Read uses message correlation ID, but received message has a null correlation ID","messagePattern":"RabbitMqIO\\.Read uses message correlation ID, but received message has a null correlation ID","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/rabbitmq/src/main/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqIO.java","lineNumber":562,"sourceCode":"    }\n\n    @Override\n    public boolean advance() throws IOException {\n      try {\n        Channel channel = connectionHandler.getChannel();\n        // we consume message without autoAck (we want to do the ack ourselves)\n        GetResponse delivery = channel.basicGet(queueName, false);\n        if (delivery == null) {\n          current = null;\n          currentRecordId = null;\n          currentTimestamp = null;\n          checkpointMark.advanceWatermark(Instant.now());\n          return false;\n        }\n        if (source.spec.useCorrelationId()) {\n          String correlationId = delivery.getProps().getCorrelationId();\n          if (correlationId == null) {\n            throw new IOException(\n                \"RabbitMqIO.Read uses message correlation ID, but received \"\n                    + \"message has a null correlation ID\");\n          }\n          currentRecordId = correlationId.getBytes(StandardCharsets.UTF_8);\n        }\n        long deliveryTag = delivery.getEnvelope().getDeliveryTag();\n        checkpointMark.sessionIds.add(deliveryTag);\n\n        current = new RabbitMqMessage(source.spec.routingKey(), delivery);\n        Date deliveryTimestamp = delivery.getProps().getTimestamp();\n        currentTimestamp =\n            (deliveryTimestamp != null) ? new Instant(deliveryTimestamp) : Instant.now();\n        checkpointMark.advanceWatermark(currentTimestamp);\n      } catch (IOException e) {\n        throw e;\n      } catch (Exception e) {\n        throw new IOException(e);\n      }","sourceCodeStart":544,"sourceCodeEnd":580,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/rabbitmq/src/main/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqIO.java#L544-L580","documentation":"When RabbitMqIO.Read is configured with withUseCorrelationId(true), the source uses the message's correlation ID as the record ID for deduplication. If a consumed message's properties carry a null correlation ID, the source cannot deduplicate and throws this IOException in advance().","triggerScenarios":"Reading with useCorrelationId enabled while the queue contains messages published without a correlationId property.","commonSituations":"Producers publishing plain messages (no correlation ID) into a queue consumed by a deduplicating reader; mixed producers on the same queue; a producer upgrade that dropped the correlation ID.","solutions":["Fix the producer to always set basicPublish properties with a correlationId.","Purge or drain messages lacking correlation IDs from the queue.","Disable withUseCorrelationId() (or remove it) if per-message deduplication is not required.","Shim in a proxy publisher that assigns a correlation ID when absent."],"exampleFix":"// before\nchannel.basicPublish(exchange, rk, null, body); // no correlationId\n// after\nAMQP.BasicProperties props = new AMQP.BasicProperties.Builder().correlationId(UUID.randomUUID().toString()).build();\nchannel.basicPublish(exchange, rk, props, body);","handlingStrategy":"validation","validationCode":"// Producer-side guard before publishing\nAMQP.BasicProperties p = props;\nif (p == null || p.getCorrelationId() == null) { throw new IllegalStateException(\"correlationId required by consumer\"); }","typeGuard":null,"tryCatchPattern":"// Wrap the read transform and alert on this specific failure\npipeline.apply(RabbitMqIO.read().withUseCorrelationId()); // ensure producer sets correlationId; catch IOException in DoFn-level logging","preventionTips":["Enforce correlationId at all producers via shared publishing utility.","Audit queues for legacy messages without correlation IDs before enabling withUseCorrelationId.","Only enable withUseCorrelationId when deduplication is strictly needed."],"tags":["rabbitmq","amqp","deduplication","apache-beam"],"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"}