apache/beam · error · IllegalStateException

need to set the topic of a PubsubIO.Write transform if not u

Error message

need to set the topic of a PubsubIO.Write transform if not using dynamic topic destinations.

What it means

PubsubIO.Write requires a topic to publish to. If no topic was configured via from(topic) and dynamic destinations are not enabled, the transform has nowhere to send records, so expand() throws IllegalStateException. Dynamic destinations (via to(topicFunction)) is the only alternative to a fixed topic.

Source

Thrown at sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java:1739

     * details on how to configure an Error Handler. Schema errors are not handled by Error
     * Handlers, and will be handled using the default behavior of the runner.
     */
    public Write<T> withErrorHandler(ErrorHandler<BadRecord, ?> badRecordErrorHandler) {
      return toBuilder()
          .setBadRecordErrorHandler(badRecordErrorHandler)
          .setBadRecordRouter(BadRecordRouter.RECORDING_ROUTER)
          .build();
    }

    /** Enable validation of the PubSub Write. */
    public Write<T> withValidation() {
      return toBuilder().setValidate(true).build();
    }

    @Override
    public PDone expand(PCollection<T> input) {
      if (getTopicProvider() == null && !getDynamicDestinations()) {
        throw new IllegalStateException(
            "need to set the topic of a PubsubIO.Write transform if not using "
                + "dynamic topic destinations.");
      }

      ValueProvider<PubsubTopic> topicProvider = getTopicProvider();
      SerializableFunction<ValueInSingleWindow<T>, PubsubIO.PubsubTopic> topicFunction =
          getTopicFunction();
      if (topicFunction == null && topicProvider != null) {
        topicFunction = v -> topicProvider.get();
      }
      int maxMessageSize = PUBSUB_MESSAGE_MAX_TOTAL_SIZE;
      if (input.isBounded() == PCollection.IsBounded.BOUNDED) {
        maxMessageSize =
            Math.min(
                maxMessageSize,
                MoreObjects.firstNonNull(
                    getMaxBatchBytesSize(), MAX_PUBLISH_BATCH_BYTE_SIZE_DEFAULT));
      }

View on GitHub (pinned to 12126d8942)

Solutions

1. Add .to("projects/PROJECT/topics/TOPIC") to the PubsubIO.Write builder.
2. Or use .to(SerializableFunction<ValueInSingleWindow<T>, PubsubTopic>) for dynamic destinations.
3. Review the builder chain to ensure a topic-setting call is present on all code paths.

Example fix

// before
PubsubIO.writeMessages().apply(input);
// after
PubsubIO.writeMessages().to("projects/my-proj/topics/myTopic").apply(input);

if (write.getTopicProvider() == null && !write.getDynamicDestinations()) {
  throw new IllegalStateException("PubsubIO.Write requires .to(...) or dynamic destinations");
}

When it happens

Trigger: Building a PubsubIO.writeMessages() transform without calling .to(...) or .from(topic), without withDynamicDestinations(), then applying it to a PCollection.

Common situations: Forgetting the .to() builder call, refactoring away the topic configuration, or conditionally configuring the writer in code paths where neither branch sets a topic.

Understand the failure class

Background: "is required", "must be set", "missing required field": configuration validation errors across open-source libraries — this error's family across 36 libraries.

Related errors

• PubSubIO cannot be configured with both a dead letter topic
• getClass() + " needs to override getOutputCoder()."
• Only %s objects with the same window supplier are compatible
• ApproximateUnique.PerKey needs an estimation error between 1
• ApproximateUnique.PerKey requires its input to use KvCoder

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