apache/beam · error · IllegalArgumentException

Expected the topic to be not null

Error message

Expected the topic to be not null

What it means

fromConfigRow reconstructs TopicPartition objects from topic_partition Rows; each Row must carry a non-null "topic" string. A null topic means the config Row is malformed and deserialization cannot proceed, so IllegalArgumentException is thrown.

Solutions

1. Ensure every topic_partition Row sets the "topic" string field.
2. Regenerate the config Row from the original transform via toConfigRow/row().
3. Validate payload Rows (topic + partition non-null) before submitting the upgrade.

Example fix

// before
Row.create(SCHEMA); // topic never set
// after
Row.withSchema(SCHEMA).withFieldValues(ImmutableMap.of("topic", "my-topic", "partition", 0)).build();

Collection<Row> tps = configRow.getArray("topic_partition");
if (tps != null) {
  for (Row r : tps) {
    if (r.getString("topic") == null) throw new IllegalArgumentException("topic_partition Row missing topic");
  }
}

boolean hasTopic(Row r) { return r != null && r.getString("topic") != null; }

try {
  transform = fromConfigRow(configRow);
} catch (IllegalArgumentException e) {
  if (e.getMessage().contains("topic to be not null")) { /* repair or regenerate the payload */ }
  throw e;
}

When it happens

Trigger: Restoring a KafkaIO read transform whose config Row's topic_partition array contains a Row with a null "topic" field.

Common situations: Hand-constructed translation payloads; Rows built with missing fields by tooling; partially populated topic partitions from custom automation.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

• Expected the partition to be not null
• Encoded value for the offset consumer config key
• Encoded value of the consumer config property
• Error while parsing the element
• AWS credential provider type

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

Appendix: source

Thrown at sdks/java/io/kafka/upgrade/src/main/java/org/apache/beam/sdk/io/kafka/upgrade/KafkaIOTranslation.java:283

                    throw new RuntimeException(e);
                  }
                }
              });
          transform = transform.withConsumerConfigUpdates(updatedConsumerConfig);
        }
        Collection<String> topics = configRow.getArray("topics");
        if (topics != null) {
          transform = transform.withTopics(new ArrayList<>(topics));
        }
        Collection<Row> topicPartitionRows = configRow.getArray("topic_partitions");
        if (topicPartitionRows != null && !topicPartitionRows.isEmpty()) {
          Collection<TopicPartition> topicPartitions =
              topicPartitionRows.stream()
                  .map(
                      row -> {
                        String topic = row.getString("topic");
                        if (topic == null) {
                          throw new IllegalArgumentException("Expected the topic to be not null");
                        }
                        Integer partition = row.getInt32("partition");
                        if (partition == null) {
                          throw new IllegalArgumentException(
                              "Expected the partition to be not null");
                        }
                        return new TopicPartition(topic, partition);
                      })
                  .collect(Collectors.toList());
          transform = transform.withTopicPartitions(Lists.newArrayList(topicPartitions));
        }
        String topicPattern = configRow.getString("topic_pattern");
        if (topicPattern != null) {
          transform = transform.withTopicPattern(topicPattern);
        }

        byte[] keyDeserializerProvider = configRow.getBytes("key_deserializer_provider");
        if (keyDeserializerProvider != null) {

View on GitHub (pinned to 12126d8942)