Kafka 消息密钥 - byte[] 和 String 同时答案

【问题标题】：Kafka Message Key - byte[] and String SimultaneouslyKafka 消息密钥 - byte[] 和 String 同时
【发布时间】：2016-11-02 08:56:30
【问题描述】：

我对 Kafka 有一个非常令人困惑的问题 - 特别是试图获取消息的密钥。

关键似乎认为它既是字符串又是字节[]

以下代码产生以下异常：

    Map<String, Integer> topicCount = new HashMap<>();
    topicCount.put(myConsumer.getTopic(), 1);

    Map<String, List<KafkaStream<byte[], byte[]>>> consumerStreams = myConsumer.getConsumer().createMessageStreams(topicCount);
    List<KafkaStream<byte[], byte[]>> streams = consumerStreams.get(myConsumer.getTopic());
    System.out.println("Listening to topic " + myConsumer.getTopic());
    for (final KafkaStream stream : streams) {
        ConsumerIterator<String, byte[]> it = stream.iterator();
        while (it.hasNext()) {

            System.out.println("Message received from topic");

            MessageAndMetadata<String, byte[]> o = it.next();

            Object messageKey = o.key();
            System.out.println("messageKey is type: " + messageKey.getClass().getName());
            System.out.println("messageKey is type: " + messageKey.getClass().getCanonicalName());
            System.out.println("o keyDecoder: " + o.keyDecoder());

            System.out.println("Key from message: " + o.key());  //This throws exception - [B cannot be cast to java.lang.String
            //System.out.println("Key as String: " + new String(o.key(), StandardCharsets.UTF_8));    //uncomment this compile Exception - no suitable constructor found for String(java.lang.String,java.nio.charset.Charset)

            byte[] bytesIn = o.message();       //getting the bytes is fine

            System.out.println("MessageAndMetadata: " + o);

            ///other code cut
        }
    }

例外：

Listening to topic MyKafkaTopic
Message received from topic
messageKey is type: [B
messageKey is type: byte[]
o decoder: kafka.serializer.DefaultDecoder@2e0d0acd
[WARNING] 
java.lang.reflect.InvocationTargetException
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.codehaus.mojo.exec.ExecJavaMojo$1.run(ExecJavaMojo.java:293)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.ClassCastException: [B cannot be cast to java.lang.String
    at com.foo.bar.KafkaCFS.process(KafkaCFS.java:153)
    at com.foo.bar.KafkaCFS.run(KafkaCFS.java:63)
    at com.foo.bar.App.main(App.java:90)
    ... 6 more

马文：

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka_2.10</artifactId>
  <version>0.9.0.1</version>
</dependency>

如果我取消注释 System.out 行，那么我什至无法编译：

[ERROR] COMPILATION ERROR :
[INFO] -------------------------------------------------------------
[ERROR] /C:/Dev/main/java/com/foo/bar/KafkaCFS.java:[152,56] no suitable constructor found for String(java.lang.String,java.nio.charset.Charset)
    constructor java.lang.String.String(byte[],int) is not applicable
    (argument mismatch; java.lang.String cannot be converted to byte[])

编译器如何认为 Key 是一个字符串（这是我所期望的）但运行时它是一个字节数组？

如何将 Key 作为字符串获取？

谢谢，

卡。

【问题讨论】：

标签： string casting byte apache-kafka

【解决方案1】：

这不匹配！您将流声明为KafkaStream<byte[], byte[]>，然后您期望ConsumerIterator<String, byte[]> it = stream.iterator(); 它应该是ConsumerIterator<byte[], byte[]> it = stream.iterator(); 以匹配泛型。然后你可以得到o.key()并通过new String(o.key());创建一个字符串

【讨论】：

【解决方案2】：

最好设置 KafkaStream 泛型参数类型为 (byte[], byte[])。尝试像这样更改代码：

ConsumerIterator<byte[], byte[]> it = stream.iterator();
while (it.hasNext()) {
    String key = new String(it.next().key());
    ...
}

【讨论】：