如何在字节数组中反转 UTF-8 编码？

问：

我有以下问题：

1. 某个生产者将 Protobuf 消息作为二进制数据（字节数组）发送。

2. 这些二进制数据进入配置错误的 Kafka 集群，该集群将字节数组反序列化为字符串。

3. 然后，该群集将数据序列化为字符串，并将其发送给使用者。

4. 毫无戒心的消费者期望收到一个二进制字节数组，但得到的却是 UTF-8 编码的混乱。

我试图在 JUnit 测试中重现它。

假设我们有以下原型文件：

syntax = "proto3";

import "google/protobuf/wrappers.proto";
import "google/protobuf/timestamp.proto";

option java_package = "com.mycompany.proto";
option java_multiple_files = true;

package com.mycompany;

enum MessageType {
    NOT_SET = 0;
    TYPE_A = 1;
    TYPE_B = 2;
}

message MyMessagePart {
    string someValue = 1;
}

message MyMessage {
  // Numeric (integer) variable
  int32 myNumber = 1;

  // Text value
  string myText = 2;

  // Enum value
  MessageType mType = 3;

  // Message parts
  repeated MyMessagePart messagePart = 4;

  // Uint32 value
  google.protobuf.UInt32Value uint32Value = 5;

  // Timestamp
  google.protobuf.Timestamp timestamp = 6;
}

然后我写了下面的测试。

public class EncodingTest {
    @Test
    public void dealWithCorruptedBinaryData() throws InvalidProtocolBufferException {
        // 1. Create a Protobuf message
        final MyMessage msg = MyMessage.newBuilder()
                .setMyNumber(42)
                .setMyText("Hello")
                .setMType(MessageType.TYPE_A)
                .setUint32Value(UInt32Value.newBuilder()
                        .setValue(2067)
                        .build())
                .addMessagePart(MyMessagePart.newBuilder()
                        .setSomeValue("message part value")
                        .build())
                .build();

        // 2. Convert it to bytes
        final byte[] bytesSentByProducer = msg.toByteArray();

        // 3. Now bytesSentByProducer enter misconfigured Kafka
        // where they are deserialized using StringDeserializer
        final StringDeserializer deserializer = new StringDeserializer();
        final String dataReceivedInsideMisconfiguredKafka = deserializer.deserialize("inputTopic",
                bytesSentByProducer);

        // 4. Then, misconfigured Kafka serializes the data as String
        final StringSerializer serializer = new StringSerializer();
        final byte[] dataSentToConsumer = serializer.serialize("outputTopic", dataReceivedInsideMisconfiguredKafka);

        // Because dataSentToConsumer have been corrupted during deserialization
        // or serialization as string, conversion back to Protobuf does not work.

        final MyMessage receivedMessage = MyMessage.parseFrom(dataSentToConsumer);

    }
}

生产者创建一个 Protobuf 消息并将其编码为字节数组。msgbytesSentByProducer

配置错误的 Kafka 集群接收该字节数组，将其反序列化为字符串，将其序列化为字符串并将其发送给使用者。dataReceivedInsideMisconfiguredKafkadataSentToConsumer

由于 UTF-8 编码已损坏二进制数据，因此调用

final MyMessage receivedMessage = MyMessage.parseFrom(dataSentToConsumer);

导致异常：

com.google.protobuf.InvalidProtocolBufferException: While parsing a protocol message, the input ended unexpectedly in the middle of a field.  This could mean either that the input has been truncated or that an embedded message misreported its own length.

    at com.google.protobuf.InvalidProtocolBufferException.truncatedMessage(InvalidProtocolBufferException.java:107)
    at com.google.protobuf.CodedInputStream$ArrayDecoder.readRawByte(CodedInputStream.java:1245)
    at com.google.protobuf.CodedInputStream$ArrayDecoder.readRawVarint64SlowPath(CodedInputStream.java:1130)
    at com.google.protobuf.CodedInputStream$ArrayDecoder.readRawVarint32(CodedInputStream.java:1024)
    at com.google.protobuf.CodedInputStream$ArrayDecoder.readUInt32(CodedInputStream.java:954)
    at com.google.protobuf.UInt32Value.<init>(UInt32Value.java:58)
    at com.google.protobuf.UInt32Value.<init>(UInt32Value.java:14)

将字节数组转换回消息的过程与未损坏的字节数组 （） 一起使用。bytesSentByProducerMyMessage.parseFrom(bytesSentByProducer)

问题：

1. 是否可以转换为 ？dataSentToConsumerbytesSentByProducer

2. 如果是，如果我控制的唯一部分是消费者，我该如何解决这个问题？如何撤消在配置错误的 Kafka 集群中发生的 UTF-8 编码？

注意：显而易见的解决方案是正确配置 Kafka 集群。同一个消费者在另一个环境中工作正常，那里有一个普通的 Kafka 集群，它不会进行任何奇怪的转换。由于官僚主义的原因，这种明显且最简单的解决方案不可用。

我试过了什么

方法 1

private byte[] convertToOriginalBytes(final byte[] bytesAfter) throws CharacterCodingException {
  final Charset charset = StandardCharsets.UTF_8;
  final CharsetDecoder decoder = charset.newDecoder();
  final CharsetEncoder encoder = charset.newEncoder();
  final ByteBuffer byteBuffer = ByteBuffer.wrap(bytesAfter);
  final CharBuffer charBuffer = CharBuffer.allocate(bytesAfter.length);
  final CoderResult result = decoder.decode(byteBuffer, charBuffer, true);
  result.throwException();
  final ByteBuffer reversedByteBuffer = encoder.encode(charBuffer);

  final byte[] reversedBytes = new byte[reversedByteBuffer.remaining()];
  reversedByteBuffer.get(reversedBytes);
  return reversedBytes;
}

结果是一个例外。

java.nio.BufferUnderflowException
    at java.base/java.nio.charset.CoderResult.throwException(CoderResult.java:272)
    at com.mycompany.EncodingTest.convertToOriginalBytes(EncodingTest.java:67)
    at com.mycompany.EncodingTest.dealWithCorruptedBinaryData(EncodingTest.java:54)

方法 2

据我所知，UTF-8 有各种字节模式：

1. 0xxxxxxx对于单字节字符。
2. 110xxxxx 10xxxxxx用于双字节字符等。

我假设内部的某个地方和/或二进制数据被修改以符合这样的 UTF-8 规则。StringDeserializerStringSerializer

如果这种转换是可逆的，则可以操纵位来获取原始消息。