apache/beam · error · IllegalStateException

to(queueUrl) or to(dynamicDestination) is required

Error message

to(queueUrl) or to(dynamicDestination) is required

What it means

SqsIO.WriteBatches' setup validates the write destination: if neither a static queueUrl nor a dynamicDestination is configured it throws IllegalStateException 'to(queueUrl) or to(dynamicDestination) is required'. The Sink cannot determine which SQS queue to send batches to.

Source

Thrown at sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/sqs/SqsIO.java:613

        this.entryMapper = entryMapper;
        this.handler =
            AsyncBatchWriteHandler.byId(
                spec.concurrentRequests(),
                spec.batchSize(),
                spec.clientConfiguration().retry(),
                Stats.NONE,
                (queue, records) -> sendMessageBatch(sqs, queue, records),
                error -> error.code(),
                record -> record.id(),
                error -> error.id());
        this.scheduler =
            spec.strictTimeouts() ? Executors.newSingleThreadScheduledExecutor() : null;
        if (spec.queueUrl() != null) {
          this.batches = new Single();
        } else if (spec.dynamicDestination() != null) {
          this.batches = new Dynamic(spec.dynamicDestination());
        } else {
          throw new IllegalStateException("to(queueUrl) or to(dynamicDestination) is required");
        }
      }

      private static CompletableFuture<List<BatchResultErrorEntry>> sendMessageBatch(
          SqsAsyncClient sqs, String queue, List<SendMessageBatchRequestEntry> records) {
        SendMessageBatchRequest request =
            SendMessageBatchRequest.builder().queueUrl(queue).entries(records).build();
        return sqs.sendMessageBatch(request).thenApply(resp -> resp.failed());
      }

      public void startBundle() {
        handler.reset();
        if (scheduler != null && spec.strictTimeouts()) {
          long timeout = spec.batchTimeout().getMillis();
          long period = timeout / CHECKS_PER_TIMEOUT_PERIOD;
          expirationCheck =
              scheduler.scheduleWithFixedDelay(
                  () -> batches.submitExpired(false), timeout, period, MILLISECONDS);

View on GitHub (pinned to 12126d8942)

Solutions

1. Call .to(String queueUrl) with the SQS queue URL on the writeBatches builder
2. Or call .to(DynamicDestinations) to resolve the queue per record
3. Validate configuration before submitting the pipeline

Example fix

// before
sqsData.apply(SqsIO.writeBatches().withSqsClient(client));
// after
sqsData.apply(SqsIO.writeBatches()
    .withSqsClient(client)
    .to("https://sqs.us-east-1.amazonaws.com/123456789012/my-queue"));

SqsIO.Write<String> w = SqsIO.writeBatches().withSqsClient(client); if (w == null) throw new IllegalArgumentException("must call to(queueUrl) or to(dynamicDestination)");

try { data.apply(SqsIO.writeBatches().withSqsClient(client)); } catch (IllegalStateException e) { /* add required .to(...) and resubmit */ }

When it happens

Trigger: Applying SqsIO.writeBatches() to a pipeline without calling .to(queueUrl) or .to(DynamicDestination); calling to() with a null/empty value so spec.queueUrl() and spec.dynamicDestination() are both null.

Common situations: Copy-pasted write transforms where the .to() call was dropped; building the sink conditionally and forgetting both branches; refactors moving from static to dynamic destinations without setting the other.

Understand the failure class

Background: "Must pass :limit option" / "Missing required option" — required option errors explained — this error's family across 41 libraries.

Related errors

• Unrecognized value for stable unique names:
• No filesystem found for scheme
• Exploded field %s must be an iterable type, got %s.
• boolean cross product parameter required to explode more tha
• ${config}

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