{"record":{"id":"bb1700a6b17d1d24","repo":"apache/beam","slug":"need-to-set-the-topic-of-a-pubsubio-write-transform-if-not","errorCode":null,"errorMessage":"need to set the topic of a PubsubIO.Write transform if not using dynamic topic destinations.","messagePattern":"need to set the topic of a PubsubIO\\.Write transform if not using dynamic topic destinations\\.","errorType":"validation","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java","lineNumber":1739,"sourceCode":"     * details on how to configure an Error Handler. Schema errors are not handled by Error\n     * Handlers, and will be handled using the default behavior of the runner.\n     */\n    public Write<T> withErrorHandler(ErrorHandler<BadRecord, ?> badRecordErrorHandler) {\n      return toBuilder()\n          .setBadRecordErrorHandler(badRecordErrorHandler)\n          .setBadRecordRouter(BadRecordRouter.RECORDING_ROUTER)\n          .build();\n    }\n\n    /** Enable validation of the PubSub Write. */\n    public Write<T> withValidation() {\n      return toBuilder().setValidate(true).build();\n    }\n\n    @Override\n    public PDone expand(PCollection<T> input) {\n      if (getTopicProvider() == null && !getDynamicDestinations()) {\n        throw new IllegalStateException(\n            \"need to set the topic of a PubsubIO.Write transform if not using \"\n                + \"dynamic topic destinations.\");\n      }\n\n      ValueProvider<PubsubTopic> topicProvider = getTopicProvider();\n      SerializableFunction<ValueInSingleWindow<T>, PubsubIO.PubsubTopic> topicFunction =\n          getTopicFunction();\n      if (topicFunction == null && topicProvider != null) {\n        topicFunction = v -> topicProvider.get();\n      }\n      int maxMessageSize = PUBSUB_MESSAGE_MAX_TOTAL_SIZE;\n      if (input.isBounded() == PCollection.IsBounded.BOUNDED) {\n        maxMessageSize =\n            Math.min(\n                maxMessageSize,\n                MoreObjects.firstNonNull(\n                    getMaxBatchBytesSize(), MAX_PUBLISH_BATCH_BYTE_SIZE_DEFAULT));\n      }","sourceCodeStart":1721,"sourceCodeEnd":1757,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java#L1721-L1757","documentation":"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.","triggerScenarios":"Building a PubsubIO.writeMessages() transform without calling .to(...) or .from(topic), without withDynamicDestinations(), then applying it to a PCollection.","commonSituations":"Forgetting the .to() builder call, refactoring away the topic configuration, or conditionally configuring the writer in code paths where neither branch sets a topic.","solutions":["Add .to(\"projects/PROJECT/topics/TOPIC\") to the PubsubIO.Write builder.","Or use .to(SerializableFunction<ValueInSingleWindow<T>, PubsubTopic>) for dynamic destinations.","Review the builder chain to ensure a topic-setting call is present on all code paths."],"exampleFix":"// before\nPubsubIO.writeMessages().apply(input);\n// after\nPubsubIO.writeMessages().to(\"projects/my-proj/topics/myTopic\").apply(input);","handlingStrategy":"validation","validationCode":"if (write.getTopicProvider() == null && !write.getDynamicDestinations()) {\n  throw new IllegalStateException(\"PubsubIO.Write requires .to(...) or dynamic destinations\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always call .to(...) immediately after PubsubIO.writeMessages().","Wrap PubsubIO writer construction in a factory method that enforces a topic argument.","Add a pipeline-construction unit test that expands every write transform."],"tags":["java","pubsub","missing-configuration","pipeline-construction"],"backgroundTag":"missing-required-config-field","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}