{"record":{"id":"fac295d58680e393","repo":"apache/seatunnel","slug":"illegal-argument-fac295","errorCode":"ILLEGAL_ARGUMENT","errorMessage":"Field '%s' cannot be in both partition_key_fields and kafka_headers_fields","messagePattern":"Field '(.+?)' cannot be in both partition_key_fields and kafka_headers_fields","errorType":"error_code","errorClass":"KafkaConnectorException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkWriter.java","lineNumber":213,"sourceCode":"\n        String delimiter = DEFAULT_FIELD_DELIMITER;\n\n        if (pluginConfig.get(FIELD_DELIMITER) != null) {\n            delimiter = pluginConfig.get(FIELD_DELIMITER);\n        }\n        if (pluginConfig.get(PARTITION_KEY_FIELDS) != null && pluginConfig.get(PARTITION) != null) {\n            throw new KafkaConnectorException(\n                    KafkaConnectorErrorCode.GET_TRANSACTIONMANAGER_FAILED,\n                    \"Cannot select both `partiton` and `partition_key_fields`. You can configure only one of them\");\n        }\n\n        // Validate that partition_key_fields and kafka_headers_fields don't overlap\n        List<String> partitionKeyFields = getPartitionKeyFields(pluginConfig, seaTunnelRowType);\n        List<String> headerFields = getHeaderFields(pluginConfig, seaTunnelRowType);\n        if (!partitionKeyFields.isEmpty() && !headerFields.isEmpty()) {\n            for (String headerField : headerFields) {\n                if (partitionKeyFields.contains(headerField)) {\n                    throw new KafkaConnectorException(\n                            CommonErrorCode.ILLEGAL_ARGUMENT,\n                            String.format(\n                                    \"Field '%s' cannot be in both partition_key_fields and kafka_headers_fields\",\n                                    headerField));\n                }\n            }\n        }\n\n        if (pluginConfig.get(PARTITION_KEY_FIELDS) != null) {\n            return DefaultSeaTunnelRowSerializer.create(\n                    topic,\n                    partitionKeyFields,\n                    headerFields,\n                    seaTunnelRowType,\n                    messageFormat,\n                    delimiter,\n                    pluginConfig);\n        }","sourceCodeStart":195,"sourceCodeEnd":231,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkWriter.java#L195-L231","documentation":"getSerializer() validates that no field appears in both partition_key_fields and kafka_headers_fields; a field used as the partition key cannot also be written as a message header. Throws ILLEGAL_ARGUMENT naming the offending field.","triggerScenarios":"A column name is listed in both partition_key_fields and kafka_headers_fields lists; detected at writer initialization while building the serializer.","commonSituations":"User keys the partition by the same column they also mirror into Kafka headers (e.g. tenant_id), assuming overlap is allowed.","solutions":["Remove the overlapping field from one of the two lists","If the value is needed in both places, use a duplicate/copy column for the header and keep the original as the partition key"],"exampleFix":"// before\npartition_key_fields = [\"tenant_id\"]\nkafka_headers_fields = [\"tenant_id\", \"trace_id\"]\n// after\npartition_key_fields = [\"tenant_id\"]\nkafka_headers_fields = [\"trace_id\"]","handlingStrategy":"validation","validationCode":"Set<String> overlap = new HashSet<>(partitionKeyFields);\noverlap.retainAll(headerFields);\nif (!overlap.isEmpty()) {\n  throw new IllegalArgumentException(\"Fields in both partition_key_fields and kafka_headers_fields: \" + overlap);\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Keep partition-key and header field lists disjoint","Duplicate a column instead of reusing it when it must appear in both roles"],"tags":["kafka","config-validation","partition-key","headers"],"backgroundTag":"conflicting-config-options","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"}