{"record":{"id":"73abf708fcb9a382","repo":"apache/seatunnel","slug":"commonerrorcodedeprecated-illegal-argument-73abf7","errorCode":"CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT","errorMessage":"Partition key field not found: %s, rowType: %s","messagePattern":"Partition key field not found: (.+?), rowType: (.+?)","errorType":"error_code","errorClass":"PulsarConnectorException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java","lineNumber":329,"sourceCode":"                row -> {\n                    Object[] keyFields = new Object[keyFieldIndexArr.length];\n                    for (int i = 0; i < keyFieldIndexArr.length; i++) {\n                        keyFields[i] = row.getField(keyFieldIndexArr[i]);\n                    }\n                    return new SeaTunnelRow(keyFields);\n                };\n        return row -> keySerializationSchema.serialize(keyDataExtractor.apply(row));\n    }\n\n    private List<String> getPartitionKeyFields(\n            ReadonlyConfig pluginConfig, SeaTunnelRowType seaTunnelRowType) {\n        if (pluginConfig.getOptional(PulsarSinkOptions.PARTITION_KEY_FIELDS).isPresent()) {\n            List<String> partitionKeyFields =\n                    pluginConfig.get(PulsarSinkOptions.PARTITION_KEY_FIELDS);\n            List<String> rowTypeFieldNames = Arrays.asList(seaTunnelRowType.getFieldNames());\n            for (String partitionKeyField : partitionKeyFields) {\n                if (!rowTypeFieldNames.contains(partitionKeyField)) {\n                    throw new PulsarConnectorException(\n                            CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT,\n                            String.format(\n                                    \"Partition key field not found: %s, rowType: %s\",\n                                    partitionKeyField, rowTypeFieldNames));\n                }\n            }\n            return partitionKeyFields;\n        }\n        return Collections.emptyList();\n    }\n\n    private TransactionImpl createTransaction() {\n        try {\n            return (TransactionImpl)\n                    PulsarConfigUtil.getTransaction(pulsarClient, transactionTimeout);\n        } catch (Exception e) {\n            throw new PulsarConnectorException(\n                    PulsarConnectorErrorCode.CREATE_TRANSACTION_FAILED,","sourceCodeStart":311,"sourceCodeEnd":347,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java#L311-L347","documentation":"When partition-key-based routing is enabled via partitionKeyFields, each listed field must exist in the SeaTunnelRowType schema. If any name in the config does not match a row field, the writer throws this error at initialization (getPartitionKeyFields) with the offending field and full rowType listed.","triggerScenarios":"Configuring PulsarSinkOptions.PARTITION_KEY_FIELDS with a field name not present in seaTunnelRowType.getFieldNames(); getPartitionKeyFields runs during writer construction.","commonSituations":"Field renamed upstream (source schema change) while sink config kept the old name; case mismatch ('UserId' vs 'userId'); typo; schema derived dynamically so expected column missing.","solutions":["Correct partition_key_fields to exactly match field names in the upstream schema (case-sensitive).","Inspect the rowType printed in the message and align the config with it.","If the upstream schema can drift, adjust the transform to guarantee the partition key column exists or use a different routing mode."],"exampleFix":"// before\nsink {\n  Pulsar {\n    partition_key_fields = \"userId\"\n  }\n}\n// after (schema has 'user_id')\nsink {\n  Pulsar {\n    partition_key_fields = \"user_id\"\n  }\n}","handlingStrategy":"validation","validationCode":"List<String> rowFields = Arrays.asList(rowType.getFieldNames());\nfor (String f : partitionKeyFields) {\n    if (!rowFields.contains(f)) {\n        throw new IllegalArgumentException(\"partition_key_field not in schema: \" + f);\n    }\n}","typeGuard":"boolean allFieldsExist = (List<String> keys, SeaTunnelRowType t) -> Arrays.asList(t.getFieldNames()).containsAll(keys);","tryCatchPattern":"try {\n    writer = new PulsarSinkWriter(pluginConfig, rowType);\n} catch (PulsarConnectorException e) {\n    if (e.getMessage().startsWith(\"Partition key field not found\")) {\n        log.error(\"Fix partition_key_fields to match schema: {}\", rowType.getFieldNames());\n    }\n    throw e;\n}","preventionTips":["Cross-check partition_key_fields against the upstream schema (case-sensitive).","Update sink configs whenever source schemas change.","Avoid dynamic schema drift or add a transform guaranteeing key columns."],"tags":["pulsar","sink","schema-mismatch"],"backgroundTag":"invalid-argument-value","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}