apache/beam · error · IOException

RabbitMqIO.Read uses message correlation ID, but received me

Error message

RabbitMqIO.Read uses message correlation ID, but received message has a null correlation ID

What it means

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().

Source

Thrown at sdks/java/io/rabbitmq/src/main/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqIO.java:562

    }

    @Override
    public boolean advance() throws IOException {
      try {
        Channel channel = connectionHandler.getChannel();
        // we consume message without autoAck (we want to do the ack ourselves)
        GetResponse delivery = channel.basicGet(queueName, false);
        if (delivery == null) {
          current = null;
          currentRecordId = null;
          currentTimestamp = null;
          checkpointMark.advanceWatermark(Instant.now());
          return false;
        }
        if (source.spec.useCorrelationId()) {
          String correlationId = delivery.getProps().getCorrelationId();
          if (correlationId == null) {
            throw new IOException(
                "RabbitMqIO.Read uses message correlation ID, but received "
                    + "message has a null correlation ID");
          }
          currentRecordId = correlationId.getBytes(StandardCharsets.UTF_8);
        }
        long deliveryTag = delivery.getEnvelope().getDeliveryTag();
        checkpointMark.sessionIds.add(deliveryTag);

        current = new RabbitMqMessage(source.spec.routingKey(), delivery);
        Date deliveryTimestamp = delivery.getProps().getTimestamp();
        currentTimestamp =
            (deliveryTimestamp != null) ? new Instant(deliveryTimestamp) : Instant.now();
        checkpointMark.advanceWatermark(currentTimestamp);
      } catch (IOException e) {
        throw e;
      } catch (Exception e) {
        throw new IOException(e);
      }

View on GitHub (pinned to 12126d8942)

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.

Example fix

// before
channel.basicPublish(exchange, rk, null, body); // no correlationId
// after
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().correlationId(UUID.randomUUID().toString()).build();
channel.basicPublish(exchange, rk, props, body);

Defensive patterns

Strategy: validation

Validate before calling

// Producer-side guard before publishing
AMQP.BasicProperties p = props;
if (p == null || p.getCorrelationId() == null) { throw new IllegalStateException("correlationId required by consumer"); }

Try / catch

// Wrap the read transform and alert on this specific failure
pipeline.apply(RabbitMqIO.read().withUseCorrelationId()); // ensure producer sets correlationId; catch IOException in DoFn-level logging

Prevention

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.

When it happens

Trigger: Reading with useCorrelationId enabled while the queue contains messages published without a correlationId property.

Common situations: 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.

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

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