{"record":{"id":"1f6ed3dee040095d","repo":"apache/beam","slug":"expected-the-producer-config-to-have-consumerconfig","errorCode":null,"errorMessage":"Expected the producer config to have 'ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG' set. Found: %s","messagePattern":"Expected the producer config to have 'ConsumerConfig\\.BOOTSTRAP_SERVERS_CONFIG' set\\. Found: (.+?)","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/kafka/upgrade/src/main/java/org/apache/beam/sdk/io/kafka/upgrade/KafkaIOTranslation.java","lineNumber":524,"sourceCode":"        throws IOException {\n      {\n        // Setting an empty payload since Kafka transform payload is not actually used by runners\n        // currently.\n        // This can be implemented if runners started actually using the Kafka transform payload.\n        return FunctionSpec.newBuilder().setUrn(getUrn()).setPayload(ByteString.empty()).build();\n      }\n    }\n\n    @Override\n    public Row toConfigRow(Write<?, ?> transform) {\n      Map<String, Object> fieldValues = new HashMap<>();\n\n      WriteRecords<?, ?> writeRecordsTransform = transform.getWriteRecordsTransform();\n\n      if (!writeRecordsTransform\n          .getProducerConfig()\n          .containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) {\n        throw new IllegalArgumentException(\n            \"Expected the producer config to have 'ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG' set. Found: \"\n                + writeRecordsTransform.getProducerConfig());\n      }\n      fieldValues.put(\n          \"bootstrap_servers\",\n          writeRecordsTransform.getProducerConfig().get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG));\n      if (writeRecordsTransform.getTopic() != null) {\n        fieldValues.put(\"topic\", writeRecordsTransform.getTopic());\n      }\n      if (writeRecordsTransform.getKeySerializer() != null) {\n        fieldValues.put(\"key_serializer\", toByteArray(writeRecordsTransform.getKeySerializer()));\n      }\n      if (writeRecordsTransform.getValueSerializer() != null) {\n        fieldValues.put(\n            \"value_serializer\", toByteArray(writeRecordsTransform.getValueSerializer()));\n      }\n      if (writeRecordsTransform.getProducerFactoryFn() != null) {\n        fieldValues.put(","sourceCodeStart":506,"sourceCodeEnd":542,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/upgrade/src/main/java/org/apache/beam/sdk/io/kafka/upgrade/KafkaIOTranslation.java#L506-L542","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Build the write transform with .withBootstrapServers(\"host:port\") so the value lands in the producer config.","If relying on kafka.producer.bootstrapServers system property, also set it in the producer config map explicitly before translation.","Inspect the 'Found: {...}' output to see which keys are present and add the missing bootstrap servers entry."],"exampleFix":"// before\nKafkaIO.<byte[], byte[]>write().withProducerConfig(producerConfigWithoutBootstrap)\n// after\nKafkaIO.<byte[], byte[]>write()\n    .withBootstrapServers(\"broker1:9092\")\n    .withProducerConfig(producerConfig)\n","handlingStrategy":"validation","validationCode":"Map<String, Object> producerConfig = writeRecordsTransform.getProducerConfig();\nif (producerConfig == null || !producerConfig.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) {\n  throw new IllegalArgumentException(\"Producer config must include bootstrap.servers before translation\");\n}","typeGuard":"boolean hasBootstrapServers(org.apache.kafka.clients.producer.KafkaProducer.ProducerConfig unused, Map<String, Object> cfg) {\n  return cfg != null && cfg.containsKey(org.apache.kafka.clients.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG);\n}","tryCatchPattern":"try {\n  row = translation.toConfigRow(transform, ...);\n} catch (IllegalArgumentException e) {\n  if (e.getMessage().contains(\"BOOTSTRAP_SERVERS_CONFIG\")) { /* rebuild transform with withBootstrapServers */ }\n  throw e;\n}","preventionTips":["Always call withBootstrapServers() when building KafkaIO write transforms.","Do not rely solely on the kafka.producer.bootstrapServers system property.","Assert producerConfig.containsKey(\"bootstrap.servers\") in pipeline-construction unit tests."],"tags":["kafka","producer-config","bootstrap-servers","missing-config"],"backgroundTag":"missing-required-config-field","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}