{"record":{"id":"b4dfdbb315b9f059","repo":"apache/beam","slug":"encoded-value-for-the-offset-consumer-config-key-s-was-null","errorCode":null,"errorMessage":"Encoded value for the offset consumer config key %s was null.","messagePattern":"Encoded value for the offset consumer config key (.+?) 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":428,"sourceCode":"          }\n          transform =\n              transform.withDynamicRead(\n                  org.joda.time.Duration.millis(watchTopicPartitionDuration.toMillis()));\n        }\n\n        byte[] timestampPolicyFactory = configRow.getBytes(\"timestamp_policy_factory\");\n        if (timestampPolicyFactory != null) {\n          transform =\n              transform.withTimestampPolicyFactory(\n                  (TimestampPolicyFactory) fromByteArray(timestampPolicyFactory));\n        }\n        Map<String, byte[]> offsetConsumerConfig = configRow.getMap(\"offset_consumer_config\");\n        if (offsetConsumerConfig != null) {\n          Map<String, Object> updatedOffsetConsumerConfig = new HashMap<>();\n          offsetConsumerConfig.forEach(\n              (key, dataBytes) -> {\n                if (offsetConsumerConfig.get(key) == null) {\n                  throw new IllegalArgumentException(\n                      \"Encoded value for the offset consumer config key \" + key + \" was null.\");\n                }\n                try {\n                  updatedOffsetConsumerConfig.put(\n                      key, fromByteArray(offsetConsumerConfig.get(key)));\n                } catch (InvalidClassException e) {\n                  throw new RuntimeException(e);\n                }\n              });\n          transform = transform.withOffsetConsumerConfigOverrides(updatedOffsetConsumerConfig);\n        }\n\n        byte[] checkStopReadinfFn = configRow.getBytes(\"check_stop_reading_fn\");\n        if (checkStopReadinfFn != null) {\n          transform =\n              transform.withCheckStopReadingFn(\n                  (SerializableFunction<TopicPartition, Boolean>)\n                      fromByteArray(checkStopReadinfFn));","sourceCodeStart":410,"sourceCodeEnd":446,"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#L410-L446","documentation":"fromConfigRow decodes the offset_consumer_config map, which overrides consumer settings used for offset/fetch operations. Each entry's encoded value must be a non-null byte[]; a null value is a malformed payload and is rejected with IllegalArgumentException.","triggerScenarios":"Restoring a transform whose config Row's offset_consumer_config map contains a key mapped to null.","commonSituations":"Hand-built or buggy serialization of offset consumer overrides; values stripped by tooling; old payloads where nulls were allowed.","solutions":["Provide encoded byte[] values for every offset_consumer_config key.","Remove unneeded keys from the map.","Regenerate the config Row from the source transform using withOffsetConsumerConfigOverrides with valid values."],"exampleFix":"// before\noffsetConsumerConfig.put(\"max.poll.records\", null);\n// after\noffsetConsumerConfig.put(\"max.poll.records\", toByteArray(\"500\"));\n","handlingStrategy":"validation","validationCode":"Map<String, byte[]> occ = configRow.getMap(\"offset_consumer_config\");\nif (occ != null) {\n  occ.forEach((k, v) -> { if (v == null) throw new IllegalArgumentException(\"Null encoded offset consumer config value for: \" + k); });\n}","typeGuard":"boolean offsetConfigValid(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(\"offset consumer config key\")) { /* fix or drop the offending key */ }\n  throw e;\n}","preventionTips":["Encode all offset consumer override values with toByteArray before serialization.","Never insert null values into offset consumer config maps.","Regenerate payloads with withOffsetConsumerConfigOverrides when in doubt."],"tags":["kafka","deserialization","null-value","offset-consumer-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"}