apache/beam · error

Configured topic ARN

Error message

Configured topic ARN '{}' does not exist.

What it means

SnsIO Write's expand() validates that the configured topic ARN actually exists by calling getTopicAttributes. When AWS responds with NotFoundException or InvalidParameterException, it logs this warning and returns false, which leads the sink to fail with an explicit 'topic does not exist' error instead of an opaque AWS error at write time.

Solutions

1. Verify the ARN format: arn:aws:sns:<region>:<account-id>:<topic-name> and correct any typos
2. Ensure the SNS client region (via AwsOptions) matches the topic's region
3. Recreate the topic if it was deleted, or point the sink at the existing topic
4. Confirm the caller has sns:GetTopicAttributes permission (though permission errors surface differently)

Example fix

// before
SnsIO.<String>write().to("arn:aws:sns:us-east-1:123456789012:topc")
// after (corrected name + matching region)
SnsIO.<String>write().to("arn:aws:sns:us-east-1:123456789012:my-topic")
  .withSnsClientProvider(Region.US_EAST_1);

// validate the ARN before building the pipeline
boolean exists = false;
try (SnsClient c = SnsClient.create()) {
  c.getTopicAttributes(b -> b.topicArn(topicArn));
  exists = true;
} catch (NotFoundException | InvalidParameterException e) { /* does not exist */ }
if (!exists) throw new IllegalArgumentException("Topic not found: " + topicArn);

static boolean isValidTopicArn(String arn) {
  return arn != null && arn.matches("^arn:aws:sns:[a-z0-9-]+:\\d{12}:.+ $");
}

try {
  snsClient.getTopicAttributes(b -> b.topicArn(arn));
} catch (NotFoundException e) {
  throw new IllegalArgumentException("Topic does not exist (check region/account): " + arn, e);
}

When it happens

Trigger: Writing with SnsIO.write().to(topicArn) where the ARN is misspelled, points to another region, references a deleted topic, or is formatted with wrong account/region parts so SNS returns NotFoundException.

Common situations: Typos in ARN; cross-region topic (client built for wrong region); topic deleted between pipeline assembly and run; using a topic name instead of a full ARN; wrong AWS account.

Understand the failure class

Background: 'Could not be found', 'does not exist', 'not found in database': the resource-not-found family when an ID, slug, key, or URI lookup comes back empty — this error's family across 20 libraries.

Related errors

• All parts but the last must be larger than
• AWS credential provider type
• AWS dependencies are not installed, and no alternative…
• Basepath %r must be S3 path.
• can not be serialized to Json

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

Appendix: source

Thrown at sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/sns/SnsIO.java:186

      checkArgument(getPublishRequestBuilder() != null, "withPublishRequestBuilder() is required");

      AwsOptions awsOptions = input.getPipeline().getOptions().as(AwsOptions.class);
      checkArgument(getClientConfiguration() != null, "withClientConfiguration() is required");
      ClientBuilderFactory.validate(awsOptions, getClientConfiguration());
      if (getTopicArn() != null) {
        checkArgument(checkTopicExists(awsOptions), "Topic arn %s does not exist", getTopicArn());
      }

      return input.apply(ParDo.of(new SnsWriterFn<>(this)));
    }

    private boolean checkTopicExists(AwsOptions options) {
      try (SnsClient client = buildClient(options)) {
        client.getTopicAttributes(b -> b.topicArn(getTopicArn()));
        return true;
      } catch (NotFoundException | InvalidParameterException e) {
        LoggerFactory.getLogger(Write.class)
            .warn("Configured topic ARN '{}' does not exist.", getTopicArn(), e);
        return false;
      }
    }

    private SnsClient buildClient(AwsOptions options) {
      return ClientBuilderFactory.buildClient(
          options.as(AwsOptions.class), SnsClient.builder(), getClientConfiguration());
    }

    static class SnsWriterFn<T> extends DoFn<T, PublishResponse> {
      private static final Logger LOG = LoggerFactory.getLogger(SnsWriterFn.class);
      private static final Counter SNS_WRITE_FAILURES =
          Metrics.counter(SnsWriterFn.class, "SNS_Write_Failures");

      private final Write<T> spec;
      private transient SnsClient producer;

      SnsWriterFn(Write<T> spec) {

View on GitHub (pinned to 12126d8942)