{"record":{"id":"bcc1a12cb89b48b5","repo":"apache/beam","slug":"encoded-value-of-the-consumer-config-property-s-was-null","errorCode":null,"errorMessage":"Encoded value of the consumer config property %s was null","messagePattern":"Encoded value of the consumer config property (.+?) was null","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":259,"sourceCode":"      String updateCompatibilityBeamVersion =\n          options.as(StreamingOptions.class).getUpdateCompatibilityVersion();\n      // We need to set a default 'updateCompatibilityBeamVersion' here since this PipelineOption\n      // is not correctly passed in for pipelines that use Beam 2.55.0.\n      // This is fixed for Beam 2.56.0 and later.\n      updateCompatibilityBeamVersion =\n          (updateCompatibilityBeamVersion != null) ? updateCompatibilityBeamVersion : \"2.55.0\";\n      try {\n        Read<?, ?> transform = KafkaIO.read();\n\n        Map<String, byte[]> consumerConfig = configRow.getMap(\"consumer_config\");\n        if (consumerConfig != null) {\n          Map<String, Object> updatedConsumerConfig = new HashMap<>();\n          consumerConfig.forEach(\n              (key, dataBytes) -> {\n                // Adding all allowed properties.\n                if (!KafkaIOUtils.DISALLOWED_CONSUMER_PROPERTIES.containsKey(key)) {\n                  if (consumerConfig.get(key) == null) {\n                    throw new IllegalArgumentException(\n                        \"Encoded value of the consumer config property \" + key + \" was null\");\n                  }\n                  try {\n                    updatedConsumerConfig.put(key, fromByteArray(consumerConfig.get(key)));\n                  } catch (InvalidClassException e) {\n                    throw new RuntimeException(e);\n                  }\n                }\n              });\n          transform = transform.withConsumerConfigUpdates(updatedConsumerConfig);\n        }\n        Collection<String> topics = configRow.getArray(\"topics\");\n        if (topics != null) {\n          transform = transform.withTopics(new ArrayList<>(topics));\n        }\n        Collection<Row> topicPartitionRows = configRow.getArray(\"topic_partitions\");\n        if (topicPartitionRows != null && !topicPartitionRows.isEmpty()) {\n          Collection<TopicPartition> topicPartitions =","sourceCodeStart":241,"sourceCodeEnd":277,"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#L241-L277","documentation":"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.","triggerScenarios":"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).","commonSituations":"Manually edited or partially-built translation payloads; config Rows produced by an older/buggy serializer; null values slipped in when constructing consumer config programmatically.","solutions":["Inspect the config Row's consumer_config map and populate a byte[]-encoded value for every key.","Re-serialize the transform from source with a fixed Beam version so values are encoded correctly.","If the key is unneeded, remove it from the consumer config map.","Verify the key is not a disallowed property that should have been filtered out."],"exampleFix":"// before\nconsumerConfig.put(\"auto.offset.reset\", null);\n// after\nconsumerConfig.put(\"auto.offset.reset\", toByteArray(\"earliest\"));\n","handlingStrategy":"validation","validationCode":"Map<String, byte[]> consumerConfig = configRow.getMap(\"consumer_config\");\nif (consumerConfig != null) {\n  consumerConfig.forEach((k, v) -> {\n    if (v == null) throw new IllegalArgumentException(\"Null encoded consumer config value for key: \" + k);\n  });\n}","typeGuard":"boolean hasNonNullValues(Map<String, byte[]> m) { return m == null || m.values().stream().allMatch(java.util.Objects::nonNull); }","tryCatchPattern":"try {\n  transform = fromConfigRow(configRow);\n} catch (IllegalArgumentException e) {\n  if (e.getMessage().contains(\"was null\")) { /* rebuild consumer config with encoded values */ }\n  throw e;\n}","preventionTips":["Never put null values into Kafka consumer config maps.","Encode values with toByteArray before serializing the transform.","Validate config Rows before submitting them to the upgrade path."],"tags":["kafka","deserialization","null-value","config"],"backgroundTag":"null-argument","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"}