{"record":{"id":"fd494c61a1574fd7","repo":"apache/seatunnel","slug":"common-illegal-argument-commonerrorcodedeprecated","errorCode":"COMMON_ILLEGAL_ARGUMENT (CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT)","errorMessage":"Field name { %s } is not found!","messagePattern":"Field name (.+?) is not found!","errorType":"error_code","errorClass":"KafkaConnectorException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/serialize/DefaultSeaTunnelRowSerializer.java","lineNumber":258,"sourceCode":"                        || MessageFormat.NATIVE.equals(format))\n                && topic == null) {\n            int topicFieldIndex =\n                    rowType.indexOf(CompatibleDebeziumJsonDeserializationSchema.FIELD_TOPIC);\n            return row -> row.getField(topicFieldIndex).toString();\n        }\n\n        String regex = \"\\\\$\\\\{(.*?)\\\\}\";\n        Pattern pattern = Pattern.compile(regex, Pattern.DOTALL);\n        Matcher matcher = pattern.matcher(topic);\n        boolean isExtractTopic = matcher.find();\n        if (!isExtractTopic) {\n            return row -> topic;\n        }\n\n        String topicField = matcher.group(1);\n        List<String> fieldNames = Arrays.asList(rowType.getFieldNames());\n        if (!fieldNames.contains(topicField)) {\n            throw new KafkaConnectorException(\n                    CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT,\n                    String.format(\"Field name { %s } is not found!\", topic));\n        }\n        int topicFieldIndex = rowType.indexOf(topicField);\n        return row -> {\n            Object topicFieldValue = row.getField(topicFieldIndex);\n            if (topicFieldValue == null) {\n                throw new KafkaConnectorException(\n                        CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT, \"The column value is empty!\");\n            }\n            return topicFieldValue.toString();\n        };\n    }\n\n    private static Function<SeaTunnelRow, byte[]> keyExtractor(\n            List<String> keyFields,\n            SeaTunnelRowType rowType,\n            MessageFormat format,","sourceCodeStart":240,"sourceCodeEnd":276,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/serialize/DefaultSeaTunnelRowSerializer.java#L240-L276","documentation":"Kafka sink's DefaultSeaTunnelRowSerializer.topicExtractor supports a '{field_name}' topic pattern. When the referenced field does not exist in the SeaTunnelRowType of the incoming rows, it throws KafkaConnectorException with code ILLEGAL_ARGUMENT ('Field name { %s } is not found!'). Note the message interpolates the whole topic string, not the extracted field name.","triggerScenarios":"Configuring sink topic as a pattern like 'logs-{partition}' where 'partition' is not a field of the SeaTunnel row schema (rowType.getFieldNames() does not contain it).","commonSituations":"Typo in the field name inside the topic pattern; renamed upstream columns without updating the Kafka sink topic config; using a pattern field that exists only in a different table's schema in multi-table jobs.","solutions":["Add the referenced field to the upstream data so it exists in the row schema.","Correct the field name in the topic config to match an actual field in the row's SeaTunnelRowType (exact, case-sensitive match).","Use a static topic string if you do not need per-row topic routing."],"exampleFix":"# before\ntopic = \"events-{topic_name}\"   # no field 'topic_name' in schema\n# after\ntopic = \"events-{topic}\"       # 'topic' is an actual row field","handlingStrategy":"validation","validationCode":"// validate at job-config time\nString topicPattern = \"events-{partition}\";\njava.util.regex.Matcher m = java.util.regex.Pattern.compile(\"\\\\{(\\\\w+)\\\\}\").matcher(topicPattern);\nif (m.find()) {\n    String field = m.group(1);\n    if (!java.util.Arrays.asList(rowType.getFieldNames()).contains(field)) {\n        throw new IllegalArgumentException(\"topic field not in schema: \" + field);\n    }\n}","typeGuard":null,"tryCatchPattern":"try {\n    serializer = DefaultSeaTunnelRowSerializer.create(...);\n} catch (KafkaConnectorException e) {\n    if (e.getSeaTunnelErrorCode().getCode().equals(CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT.getCode())) {\n        // fix topic pattern or add missing field, then rebuild serializer\n    }\n    throw e;\n}","preventionTips":["Verify every {field} in the topic pattern exists in the source schema (case-sensitive).","Update topic patterns whenever upstream column names change.","Prefer static topics when per-row routing is not required."],"tags":["kafka","config","schema"],"backgroundTag":"missing-required-config-field","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}