apache/beam · error · IllegalArgumentException

Expected the producer config to have…

Error message

Expected the producer config to have 'ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG' set. Found: %s

What it means

KafkaIOTranslation.toConfigRow serializes a KafkaIO write transform for upgrade. The bootstrap servers are stored under the producer config but are read from ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG; if the underlying WriteRecords producer config lacks that key, the translator cannot fill the bootstrap_servers field and throws IllegalArgumentException showing the full config.

Solutions

1. Build the write transform with .withBootstrapServers("host:port") so the value lands in the producer config.
2. If relying on kafka.producer.bootstrapServers system property, also set it in the producer config map explicitly before translation.
3. Inspect the 'Found: {...}' output to see which keys are present and add the missing bootstrap servers entry.

Example fix

// before
KafkaIO.<byte[], byte[]>write().withProducerConfig(producerConfigWithoutBootstrap)
// after
KafkaIO.<byte[], byte[]>write()
    .withBootstrapServers("broker1:9092")
    .withProducerConfig(producerConfig)

Map<String, Object> producerConfig = writeRecordsTransform.getProducerConfig();
if (producerConfig == null || !producerConfig.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) {
  throw new IllegalArgumentException("Producer config must include bootstrap.servers before translation");
}

boolean hasBootstrapServers(org.apache.kafka.clients.producer.KafkaProducer.ProducerConfig unused, Map<String, Object> cfg) {
  return cfg != null && cfg.containsKey(org.apache.kafka.clients.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG);
}

try {
  row = translation.toConfigRow(transform, ...);
} catch (IllegalArgumentException e) {
  if (e.getMessage().contains("BOOTSTRAP_SERVERS_CONFIG")) { /* rebuild transform with withBootstrapServers */ }
  throw e;
}

When it happens

Trigger: Calling toConfigRow/row() on a KafkaIO write transform whose getProducerConfig() has no ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG entry — e.g. the transform was built without withBootstrapServers and relies on the kafka.producer.bootstrapServers property or cluster default.

Common situations: Pipelines configured only via system properties (kafka.producer.bootstrapServers) rather than withBootstrapServers(); transforms assembled programmatically with incomplete producer configs; version changes where bootstrap servers moved between config maps.

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

• One of topics and topicPartitions must be configurated.
• At least one of target_batch_overhead or…
• : closing producer after unrecoverable error. The work…
• : consumer thread is interrupted
• consumerPollingTimeout should be > 0.

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

        throws IOException {
      {
        // Setting an empty payload since Kafka transform payload is not actually used by runners
        // currently.
        // This can be implemented if runners started actually using the Kafka transform payload.
        return FunctionSpec.newBuilder().setUrn(getUrn()).setPayload(ByteString.empty()).build();
      }
    }

    @Override
    public Row toConfigRow(Write<?, ?> transform) {
      Map<String, Object> fieldValues = new HashMap<>();

      WriteRecords<?, ?> writeRecordsTransform = transform.getWriteRecordsTransform();

      if (!writeRecordsTransform
          .getProducerConfig()
          .containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) {
        throw new IllegalArgumentException(
            "Expected the producer config to have 'ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG' set. Found: "
                + writeRecordsTransform.getProducerConfig());
      }
      fieldValues.put(
          "bootstrap_servers",
          writeRecordsTransform.getProducerConfig().get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG));
      if (writeRecordsTransform.getTopic() != null) {
        fieldValues.put("topic", writeRecordsTransform.getTopic());
      }
      if (writeRecordsTransform.getKeySerializer() != null) {
        fieldValues.put("key_serializer", toByteArray(writeRecordsTransform.getKeySerializer()));
      }
      if (writeRecordsTransform.getValueSerializer() != null) {
        fieldValues.put(
            "value_serializer", toByteArray(writeRecordsTransform.getValueSerializer()));
      }
      if (writeRecordsTransform.getProducerFactoryFn() != null) {
        fieldValues.put(

View on GitHub (pinned to 12126d8942)