apache/beam · error · IllegalArgumentException

Encoded value of the consumer config property

Error message

Encoded value of the consumer config property %s was null

What it means

fromConfigRow deserializes a saved KafkaIO read transform config Row. Each consumer config value must be a non-null encoded byte array; a null value for a key that is not in KafkaIOUtils.DISALLOWED_CONSUMER_PROPERTIES indicates a corrupt or hand-crafted config Row and is rejected with IllegalArgumentException.

Solutions

1. Inspect the config Row's consumer_config map and populate a byte[]-encoded value for every key.
2. Re-serialize the transform from source with a fixed Beam version so values are encoded correctly.
3. If the key is unneeded, remove it from the consumer config map.
4. Verify the key is not a disallowed property that should have been filtered out.

Example fix

// before
consumerConfig.put("auto.offset.reset", null);
// after
consumerConfig.put("auto.offset.reset", toByteArray("earliest"));

Map<String, byte[]> consumerConfig = configRow.getMap("consumer_config");
if (consumerConfig != null) {
  consumerConfig.forEach((k, v) -> {
    if (v == null) throw new IllegalArgumentException("Null encoded consumer config value for key: " + k);
  });
}

boolean hasNonNullValues(Map<String, byte[]> m) { return m == null || m.values().stream().allMatch(java.util.Objects::nonNull); }

try {
  transform = fromConfigRow(configRow);
} catch (IllegalArgumentException e) {
  if (e.getMessage().contains("was null")) { /* rebuild consumer config with encoded values */ }
  throw e;
}

When it happens

Trigger: Restoring a KafkaIO read transform from a config Row whose consumer_config map contains a key with a null encoded value (getMap("consumer_config") entry mapped to null).

Common situations: Manually edited or partially-built translation payloads; config Rows produced by an older/buggy serializer; null values slipped in when constructing consumer config programmatically.

Related errors

• Encoded value for the offset consumer config key
• Expected the partition to be not null
• Expected the topic to be not null
• Error while parsing the element
• Expected watchTopicPartitionDuration to be available when…

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/bcc1a12cb89b48b5. 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:259

      String updateCompatibilityBeamVersion =
          options.as(StreamingOptions.class).getUpdateCompatibilityVersion();
      // We need to set a default 'updateCompatibilityBeamVersion' here since this PipelineOption
      // is not correctly passed in for pipelines that use Beam 2.55.0.
      // This is fixed for Beam 2.56.0 and later.
      updateCompatibilityBeamVersion =
          (updateCompatibilityBeamVersion != null) ? updateCompatibilityBeamVersion : "2.55.0";
      try {
        Read<?, ?> transform = KafkaIO.read();

        Map<String, byte[]> consumerConfig = configRow.getMap("consumer_config");
        if (consumerConfig != null) {
          Map<String, Object> updatedConsumerConfig = new HashMap<>();
          consumerConfig.forEach(
              (key, dataBytes) -> {
                // Adding all allowed properties.
                if (!KafkaIOUtils.DISALLOWED_CONSUMER_PROPERTIES.containsKey(key)) {
                  if (consumerConfig.get(key) == null) {
                    throw new IllegalArgumentException(
                        "Encoded value of the consumer config property " + key + " was null");
                  }
                  try {
                    updatedConsumerConfig.put(key, fromByteArray(consumerConfig.get(key)));
                  } catch (InvalidClassException e) {
                    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 =

View on GitHub (pinned to 12126d8942)