{"record":{"id":"fd74ab17fef19c7a","repo":"apache/seatunnel","slug":"operation-not-supported-fd74ab","errorCode":"OPERATION_NOT_SUPPORTED","errorMessage":"kafka_headers_fields is not supported with NATIVE format. Please use JSON, TEXT, or other formats.","messagePattern":"kafka_headers_fields is not supported with NATIVE format\\. Please use JSON, TEXT, or other formats\\.","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":188,"sourceCode":"        }\n\n        kafkaProperties.put(\n                ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, pluginConfig.get(BOOTSTRAP_SERVERS));\n        kafkaProperties.put(\n                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());\n        kafkaProperties.put(\n                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());\n        return kafkaProperties;\n    }\n\n    private SeaTunnelRowSerializer<byte[], byte[]> getSerializer(\n            ReadonlyConfig pluginConfig, SeaTunnelRowType seaTunnelRowType) {\n        MessageFormat messageFormat = pluginConfig.get(FORMAT);\n        String topic = pluginConfig.get(TOPIC);\n        if (MessageFormat.NATIVE.equals(messageFormat)) {\n            // Validate that kafka_headers_fields is not configured for NATIVE format\n            if (pluginConfig.get(KAFKA_HEADERS_FIELDS) != null) {\n                throw new KafkaConnectorException(\n                        CommonErrorCode.OPERATION_NOT_SUPPORTED,\n                        \"kafka_headers_fields is not supported with NATIVE format. Please use JSON, TEXT, or other formats.\");\n            }\n            checkNativeSeaTunnelType(seaTunnelRowType);\n            return DefaultSeaTunnelRowSerializer.create(topic, messageFormat, seaTunnelRowType);\n        }\n\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","sourceCodeStart":170,"sourceCodeEnd":206,"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#L170-L206","documentation":"KafkaSinkWriter.getSerializer() rejects the kafka_headers_fields option when the message FORMAT is NATIVE, because native Kafka message-format serialization cannot attach custom headers derived from row fields. The library throws OPERATION_NOT_SUPPORTED to fail fast instead of silently dropping the headers.","triggerScenarios":"Configuring sink with format = NATIVE and a non-null kafka_headers_fields list; validation happens inside getSerializer() when the sink writer is created.","commonSituations":"User wants both SeaTunnel native serialization and custom Kafka headers; copy-pasting a config that worked with JSON/TEXT format and switching format to NATIVE.","solutions":["Remove kafka_headers_fields from the sink config when using NATIVE format","Switch format to JSON or TEXT if custom Kafka headers are required","Put headers support behind a format that supports it and keep NATIVE only for pure payload serialization"],"exampleFix":"// before\nKafka {\n  format = NATIVE\n  kafka_headers_fields = [\"hdr\"]\n}\n// after\nKafka {\n  format = JSON\n  kafka_headers_fields = [\"hdr\"]\n}","handlingStrategy":"validation","validationCode":"// before submitting\nboolean nativeFormat = \"NATIVE\".equalsIgnoreCase(config.getString(\"format\"));\nif (nativeFormat && config.getString(\"kafka_headers_fields\") != null) {\n  throw new IllegalArgumentException(\"kafka_headers_fields requires a non-NATIVE format\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Never combine format=NATIVE with kafka_headers_fields","Document per-format supported options in your config templates","Validate sink configs in CI before deployment"],"tags":["kafka","config-validation","native-format","headers"],"backgroundTag":"unsupported-operation","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"}