apache/beam · error · IllegalArgumentException

Expected the partition to be not null

Error message

Expected the partition to be not null

What it means

Same reconstruction path as the topic check: each topic_partition Row must have a non-null "partition" int32. A null partition makes it impossible to build a TopicPartition, so IllegalArgumentException is thrown.

Solutions

1. Set the "partition" int32 field on every topic_partition Row.
2. Regenerate the config Row from the source transform with row()/toConfigRow.
3. Validate all topic partition Rows before upgrade submission.

Example fix

// before
row.getString("partition"); // null, wrong field
// after
row.getInt32("partition"); // ensure int32 partition is populated

for (Row r : configRow.getArray("topic_partition")) {
  if (r.getInt32("partition") == null) throw new IllegalArgumentException("topic_partition Row missing partition");
}

boolean hasPartition(Row r) { return r != null && r.getInt32("partition") != null; }

try {
  transform = fromConfigRow(configRow);
} catch (IllegalArgumentException e) {
  if (e.getMessage().contains("partition 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 Rows have a null "partition" field.

Common situations: Manually built or tool-generated config Rows missing the partition field; schema drift where partition was dropped during serialization.

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 topic 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/4ff129f3bb87805d. 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:287

          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) {

          byte[] keyCoder = configRow.getBytes("key_coder");
          if (keyCoder != null) {
            transform =

View on GitHub (pinned to 12126d8942)