{"record":{"id":"d042217bc932fbcd","repo":"apache/beam","slug":"expected-the-topic-to-be-not-null","errorCode":null,"errorMessage":"Expected the topic to be not null","messagePattern":"Expected the topic to be not 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":283,"sourceCode":"                    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 =\n              topicPartitionRows.stream()\n                  .map(\n                      row -> {\n                        String topic = row.getString(\"topic\");\n                        if (topic == null) {\n                          throw new IllegalArgumentException(\"Expected the topic to be not null\");\n                        }\n                        Integer partition = row.getInt32(\"partition\");\n                        if (partition == null) {\n                          throw new IllegalArgumentException(\n                              \"Expected the partition to be not null\");\n                        }\n                        return new TopicPartition(topic, partition);\n                      })\n                  .collect(Collectors.toList());\n          transform = transform.withTopicPartitions(Lists.newArrayList(topicPartitions));\n        }\n        String topicPattern = configRow.getString(\"topic_pattern\");\n        if (topicPattern != null) {\n          transform = transform.withTopicPattern(topicPattern);\n        }\n\n        byte[] keyDeserializerProvider = configRow.getBytes(\"key_deserializer_provider\");\n        if (keyDeserializerProvider != null) {","sourceCodeStart":265,"sourceCodeEnd":301,"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#L265-L301","documentation":"fromConfigRow reconstructs TopicPartition objects from topic_partition Rows; each Row must carry a non-null \"topic\" string. A null topic means the config Row is malformed and deserialization cannot proceed, so IllegalArgumentException is thrown.","triggerScenarios":"Restoring a KafkaIO read transform whose config Row's topic_partition array contains a Row with a null \"topic\" field.","commonSituations":"Hand-constructed translation payloads; Rows built with missing fields by tooling; partially populated topic partitions from custom automation.","solutions":["Ensure every topic_partition Row sets the \"topic\" string field.","Regenerate the config Row from the original transform via toConfigRow/row().","Validate payload Rows (topic + partition non-null) before submitting the upgrade."],"exampleFix":"// before\nRow.create(SCHEMA); // topic never set\n// after\nRow.withSchema(SCHEMA).withFieldValues(ImmutableMap.of(\"topic\", \"my-topic\", \"partition\", 0)).build();\n","handlingStrategy":"validation","validationCode":"Collection<Row> tps = configRow.getArray(\"topic_partition\");\nif (tps != null) {\n  for (Row r : tps) {\n    if (r.getString(\"topic\") == null) throw new IllegalArgumentException(\"topic_partition Row missing topic\");\n  }\n}","typeGuard":"boolean hasTopic(Row r) { return r != null && r.getString(\"topic\") != null; }","tryCatchPattern":"try {\n  transform = fromConfigRow(configRow);\n} catch (IllegalArgumentException e) {\n  if (e.getMessage().contains(\"topic to be not null\")) { /* repair or regenerate the payload */ }\n  throw e;\n}","preventionTips":["Always populate topic and partition when constructing topic_partition Rows.","Generate payload Rows via toConfigRow rather than hand-building them.","Unit-test your upgrade payloads against fromConfigRow."],"tags":["kafka","deserialization","null-value","topic-partition"],"backgroundTag":"missing-required-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"}