Kafka实时导入Pinot遇UTF-8起始字节无效错误的解决方法

这个错误的核心原因是Kafka中的消息并非以UTF-8编码存储，Jackson的UTF8StreamJsonParser无法解析不符合UTF-8规则的字节流（比如异常中的0x96起始字节）。越南语字符“”的正确UTF-8编码是0xC3 0x82，而0x96属于Windows-1252或ISO-8859-1编码中的字符，说明生产者发送消息时使用了错误编码，或者Pinot消费时未指定对应编码。

方案1：修正Kafka生产者的编码 #

确保生产者发送消息时强制使用UTF-8编码：

• Java生产者：指定字符串序列化时使用UTF-8：

import java.nio.charset.StandardCharsets;
import org.apache.kafka.clients.producer.ProducerRecord;

String message = "{\"event\":{\"header\": \"v1\",\"body\":{\"gender\":\"M\", \"region\":\"Âmerica\"}}}";
producer.send(new ProducerRecord<>("your-topic", message.getBytes(StandardCharsets.UTF_8)));

• Python生产者：序列化时明确指定UTF-8：

from kafka import KafkaProducer
import json

producer = KafkaProducer(value_serializer=lambda v: json.dumps(v).encode('utf-8'))
producer.send('your-topic', {"event":{"header": "v1","body":{"gender":"M", "region":"Âmerica"}}})

方案2：自定义Pinot消息解码器（兼容非UTF-8编码） #

如果无法修改生产者编码，可以自定义Pinot的JSON解码器，指定消息实际使用的编码（比如Windows-1252或ISO-8859-1）：

1. 编写自定义解码器类，继承Pinot的JSONMessageDecoder：

import org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder;
import org.apache.pinot.spi.utils.JsonUtils;
import com.fasterxml.jackson.databind.JsonNode;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.Map;

public class CustomCharsetJSONDecoder extends JSONMessageDecoder {
    // 替换为消息实际使用的编码
    private static final Charset TARGET_CHARSET = Charset.forName("Windows-1252");

    @Override
    public JsonNode decode(byte[] payload, Map<String, String> decodingConfig) throws IOException {
        String jsonStr = new String(payload, TARGET_CHARSET);
        return JsonUtils.stringToJsonNode(jsonStr);
    }
}

2. 将编译后的Jar包放入Pinot的plugins目录。
3. 在实时表配置中指定自定义解码器：

streamConfigs:
  stream.kafka.decoder.class.name: com.yourcompany.CustomCharsetJSONDecoder

方案3：配置Pinot内置编码支持（适用于0.13.0+版本） #

如果使用Pinot 0.13.0及以上版本，可直接在表配置中指定消息编码：

streamConfigs:
  pinot.input.format.json.charset: Windows-1252 # 替换为实际编码

注：Pinot 0.12.1版本无此内置配置，需使用方案2。

1. 发送包含越南语特殊字符的测试消息到目标Kafka Topic。
2. 重启Pinot实时表消费者。
3. 检查Pinot日志是否消除UTF-8解析错误，同时查询表数据确认特殊字符显示正常。

内容的提问来源于stack exchange，提问作者qxk71551