{"record":{"id":"8d3f66c10b51c9de","repo":"apache/beam","slug":"fieldname-field-is-invalid-at-the-top-level-for-kafka-in-the","errorCode":null,"errorMessage":"'${fieldName}' field is invalid at the top level for Kafka in the nested schema.","messagePattern":"'(.+?)' field is invalid at the top level for Kafka in the nested schema\\.","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/Schemas.java","lineNumber":100,"sourceCode":"        case HEADERS_FIELD:\n          checkFieldHasType(field, HEADERS_FIELD_TYPE);\n          break;\n        case EVENT_TIMESTAMP_FIELD:\n          checkFieldHasType(field, FieldType.DATETIME);\n          break;\n        case MESSAGE_KEY_FIELD:\n          checkFieldHasType(field, FieldType.BYTES);\n          break;\n        case PAYLOAD_FIELD:\n          checkArgument(\n              fieldHasType(field, FieldType.BYTES)\n                  || field.getType().getTypeName().equals(TypeName.ROW),\n              String.format(\n                  \"'%s' field must either have a 'BYTES NOT NULL' or 'ROW' schema.\",\n                  field.getName()));\n          break;\n        default:\n          throw new IllegalArgumentException(\n              String.format(\n                  \"'%s' field is invalid at the top level for Kafka in the nested schema.\",\n                  field.getName()));\n      }\n    }\n  }\n}\n","sourceCodeStart":82,"sourceCodeEnd":108,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/Schemas.java#L82-L108","documentation":"Thrown by Schemas.validateNestedSchema when a field of a Kafka table's nested schema has a top-level name that is not one of the recognized Kafka schema fields (e.g. 'key', 'value', 'timestamp', 'headers'). The switch over field names has a default branch that rejects unknown fields with IllegalArgumentException.","triggerScenarios":"Declaring a Kafka external table whose schema contains a top-level field not in the supported set — e.g. adding a field 'payload' or misspelling 'value' as 'values' in a nested Kafka schema definition.","commonSituations":"Misspelling the reserved field names; attempting to add custom top-level columns that must instead be nested inside a ROW-typed 'value' field; copying a schema from a different provider whose field naming differs.","solutions":["Use only the supported top-level Kafka fields (key, value, timestamp, headers, event_timestamp etc.) in the nested schema.","Move custom columns into a ROW-typed 'value' field instead of declaring them at the top level.","Fix spelling/case of the field names and mark key/value as BYTES NOT NULL or ROW as required by validateNestedSchema."],"exampleFix":"// before\nCREATE EXTERNAL TABLE t (payload BYTES NOT NULL, `timestamp` TIMESTAMP) TYPE 'kafka'\n// after\nCREATE EXTERNAL TABLE t (value BYTES NOT NULL, `timestamp` TIMESTAMP) TYPE 'kafka'","handlingStrategy":"validation","validationCode":"Set<String> topLevel = Set.of(\"key\", \"value\", \"headers\", \"timestamp\");\nfor (Field f : schema.getFields()) {\n  if (!topLevel.contains(f.getName())) throw new IllegalArgumentException(\"Invalid top-level field: \" + f.getName());\n}","typeGuard":null,"tryCatchPattern":"try { validateNestedSchema(schema); } catch (IllegalArgumentException e) { /* fix schema or fall back to key/value BYTES schema */ }","preventionTips":["Nest custom columns under a ROW-typed 'value' field","Validate the schema string before executing the DDL","Reference a working example schema"],"tags":["java","apache-beam","kafka","sql","schema"],"backgroundTag":"schema-validation-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}