Kafka中KTable调试输出为Bytes格式 如何转换为String格式

你可以根据使用场景选择以下方案：

• 方案1：定义KTable时直接指定序列化规则
  在创建KTable时通过Consumed参数指定Value的序列化器为String类型，后续读取到的KTable数据直接为String格式：

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.kstream.Consumed;

// 直接读取为String类型的KTable
KTable<String, String> kTable = builder.table(
    "你的topic名称",
    Consumed.with(Serdes.String(), Serdes.String())
);

• 方案2：已有Bytes类型KTable转换
  如果已经拿到Bytes类型的KTable，可通过mapValues方法批量转换Value格式：

import java.nio.charset.StandardCharsets;

KTable<String, String> convertedKTable = bytesKTable.mapValues(
    bytesValue -> new String(bytesValue.get(), StandardCharsets.UTF_8)
);

• 方案3：调试场景临时转换
  如果只是调试时需要查看Bytes内容，可直接在打印时做临时转换：

bytesKTable.toStream().foreach((key, value) -> {
    String strValue = new String(value.get(), StandardCharsets.UTF_8);
    System.out.printf("key: %s, value: %s%n", key, strValue);
});

*注意：转换时必须指定字符编码，避免不同环境默认编码不同导致乱码问题。

你提供的代码片段截图如下：

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