{"record":{"id":"bade79d35d90b2a6","repo":"apache/beam","slug":"one-of-topics-and-topicpartitions-must-be-configurated","errorCode":null,"errorMessage":"One of topics and topicPartitions must be configurated.","messagePattern":"One of topics and topicPartitions must be configurated\\.","errorType":"validation","errorClass":"InvalidTableException","httpStatus":null,"severity":"error","filePath":"sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/BeamKafkaTable.java","lineNumber":156,"sourceCode":"      kafkaRead =\n          KafkaIO.<byte[], byte[]>read()\n              .withBootstrapServers(bootstrapServers)\n              .withTopics(topics)\n              .withConsumerConfigUpdates(configUpdates)\n              .withKeyDeserializerAndCoder(ByteArrayDeserializer.class, ByteArrayCoder.of())\n              .withValueDeserializerAndCoder(ByteArrayDeserializer.class, ByteArrayCoder.of())\n              .withTimestampPolicyFactory(timestampPolicyFactory);\n    } else if (topicPartitions != null) {\n      kafkaRead =\n          KafkaIO.<byte[], byte[]>read()\n              .withBootstrapServers(bootstrapServers)\n              .withTopicPartitions(topicPartitions)\n              .withConsumerConfigUpdates(configUpdates)\n              .withKeyDeserializerAndCoder(ByteArrayDeserializer.class, ByteArrayCoder.of())\n              .withValueDeserializerAndCoder(ByteArrayDeserializer.class, ByteArrayCoder.of())\n              .withTimestampPolicyFactory(timestampPolicyFactory);\n    } else {\n      throw new InvalidTableException(\"One of topics and topicPartitions must be configurated.\");\n    }\n    return kafkaRead;\n  }\n\n  @Override\n  public POutput buildIOWriter(PCollection<Row> input) {\n    checkArgument(\n        topics != null && topics.size() == 1, \"Only one topic can be acceptable as output.\");\n\n    return input\n        .apply(\"out_reformat\", getPTransformForOutput())\n        .setCoder(ProducerRecordCoder.of(ByteArrayCoder.of(), ByteArrayCoder.of()))\n        .apply(\"persistent\", createKafkaWrite());\n  }\n\n  private KafkaIO.WriteRecords<byte[], byte[]> createKafkaWrite() {\n    return KafkaIO.<byte[], byte[]>writeRecords()\n        .withBootstrapServers(bootstrapServers)","sourceCodeStart":138,"sourceCodeEnd":174,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/BeamKafkaTable.java#L138-L174","documentation":"BeamKafkaTable.createKafkaRead requires either a list of 'topics' or a 'topicPartitions' specification to know what to consume. When both are absent, it throws InvalidTableException during buildIOReader, i.e. when the SQL query over the Kafka table is expanded.","triggerScenarios":"Declaring a Kafka table in Beam SQL with neither 'topics' nor 'topicPartitions' in the table location/properties, then running a SELECT against it.","commonSituations":"Empty or malformed location string in CREATE EXTERNAL TABLE ... TYPE 'kafka'; properties JSON dropped during templating; typo like 'topic' (singular) instead of 'topics'.","solutions":["Add a 'topics' value (comma-separated) to the table's location/properties.","Alternatively provide 'topicPartitions' specifying explicit topic:partition assignments.","Check spelling and parsing of the CREATE EXTERNAL TABLE LOCATION/proPERTIES string."],"exampleFix":"// before\nCREATE EXTERNAL TABLE orders (...) TYPE 'kafka' LOCATION '';\n// after\nCREATE EXTERNAL TABLE orders (...) TYPE 'kafka' LOCATION 'orders-topic';","handlingStrategy":"validation","validationCode":"boolean hasTopics = topics != null && !topics.isEmpty();\nboolean hasPartitions = topicPartitions != null && !topicPartitions.isEmpty();\nif (!hasTopics && !hasPartitions) throw new InvalidTableException(\"Configure topics or topicPartitions\");","typeGuard":"boolean kafkaTableConfigured(Map<String,String> props) {\n  return (props.get(\"topics\") != null && !props.get(\"topics\").isEmpty())\n      || (props.get(\"topicPartitions\") != null && !props.get(\"topicPartitions\").isEmpty());\n}","tryCatchPattern":"try { env.executeDdl(selectStmt); } catch (InvalidTableException e) { /* add topics/topicPartitions to table definition */ }","preventionTips":["Always specify the topic in the CREATE EXTERNAL TABLE LOCATION.","Validate the Kafka table properties JSON before submitting queries.","Watch for singular/plural typos ('topic' vs 'topics')."],"tags":["beam-sql","kafka","missing-config"],"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-20T03:17:13.778Z"}