apache/beam · error · InvalidTableException

One of topics and topicPartitions must be configurated.

Error message

One of topics and topicPartitions must be configurated.

What it means

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.

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.

Example fix

// before
CREATE EXTERNAL TABLE orders (...) TYPE 'kafka' LOCATION '';
// after
CREATE EXTERNAL TABLE orders (...) TYPE 'kafka' LOCATION 'orders-topic';

Defensive patterns

Strategy: validation

Validate before calling

boolean hasTopics = topics != null && !topics.isEmpty();
boolean hasPartitions = topicPartitions != null && !topicPartitions.isEmpty();
if (!hasTopics && !hasPartitions) throw new InvalidTableException("Configure topics or topicPartitions");

Type guard

boolean kafkaTableConfigured(Map<String,String> props) {
  return (props.get("topics") != null && !props.get("topics").isEmpty())
      || (props.get("topicPartitions") != null && !props.get("topicPartitions").isEmpty());
}

Try / catch

try { env.executeDdl(selectStmt); } catch (InvalidTableException e) { /* add topics/topicPartitions to table definition */ }

Prevention

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').

When it happens

Trigger: Declaring a Kafka table in Beam SQL with neither 'topics' nor 'topicPartitions' in the table location/properties, then running a SELECT against it.

Common situations: Empty or malformed location string in CREATE EXTERNAL TABLE ... TYPE 'kafka'; properties JSON dropped during templating; typo like 'topic' (singular) instead of 'topics'.

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

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

Appendix: source

Thrown at sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/BeamKafkaTable.java:156

      kafkaRead =
          KafkaIO.<byte[], byte[]>read()
              .withBootstrapServers(bootstrapServers)
              .withTopics(topics)
              .withConsumerConfigUpdates(configUpdates)
              .withKeyDeserializerAndCoder(ByteArrayDeserializer.class, ByteArrayCoder.of())
              .withValueDeserializerAndCoder(ByteArrayDeserializer.class, ByteArrayCoder.of())
              .withTimestampPolicyFactory(timestampPolicyFactory);
    } else if (topicPartitions != null) {
      kafkaRead =
          KafkaIO.<byte[], byte[]>read()
              .withBootstrapServers(bootstrapServers)
              .withTopicPartitions(topicPartitions)
              .withConsumerConfigUpdates(configUpdates)
              .withKeyDeserializerAndCoder(ByteArrayDeserializer.class, ByteArrayCoder.of())
              .withValueDeserializerAndCoder(ByteArrayDeserializer.class, ByteArrayCoder.of())
              .withTimestampPolicyFactory(timestampPolicyFactory);
    } else {
      throw new InvalidTableException("One of topics and topicPartitions must be configurated.");
    }
    return kafkaRead;
  }

  @Override
  public POutput buildIOWriter(PCollection<Row> input) {
    checkArgument(
        topics != null && topics.size() == 1, "Only one topic can be acceptable as output.");

    return input
        .apply("out_reformat", getPTransformForOutput())
        .setCoder(ProducerRecordCoder.of(ByteArrayCoder.of(), ByteArrayCoder.of()))
        .apply("persistent", createKafkaWrite());
  }

  private KafkaIO.WriteRecords<byte[], byte[]> createKafkaWrite() {
    return KafkaIO.<byte[], byte[]>writeRecords()
        .withBootstrapServers(bootstrapServers)

View on GitHub (pinned to 12126d8942)