«ClassCastException: kafka.message.Message не может быть приведен к java.lang.String» на Кафка Producer

Question

Я пытаюсь написать простой производитель, который записывает сообщения Кафки Я скачал kafka0.7

Запуск сервера Кафка, производителя и потребителя

> bin/kafka-server-start.sh config/server.properties 
> bin/kafka-console-consumer.sh --zookeeper zooserver:2181 --topic test --from-beginning 
> bin/kafka-console-producer.sh --zookeeper zooserver:2181 --topic test 

и все работает!

Тогда я создал простой потребитель, и он также работает.

SimpleConsumer consumer = new SimpleConsumer("172.21.110.134", 9092, 10000, 1024000); 
… 
ByteBufferMessageSet messages = consumer.fetch(fetchRequest); 
System.out.println("consumed: " + Utils.toString(msg.message().payload(), "UTF-8")); 

Но следующий производитель всегда выдает ошибку

ProducerConfig config = new ProducerConfig(props); 
… 
Producer<String, Message> producer = new Producer<String, Message>(config); 
producer.send(new ProducerData<String, Message>("test", new Message("Hello World".getBytes()))); 

Getting следующее исключение

Exception in thread "main" java.lang.ClassCastException: kafka.message.Message cannot be cast to java.lang.String 
      at kafka.serializer.StringEncoder.toMessage(Encoder.scala:30) 
      at kafka.producer.ProducerPool$$anonfun$send$1$$anonfun$3$$anonfun$apply$1.apply(ProducerPool.scala:107) 

Я использую в Java коде зависимостей Maven

<dependency> 
     <groupId>com.twitter</groupId> 
     <artifactId>kafka_2.9.2</artifactId> 
     <version>0.7.0</version> 
    </dependency> 

    <dependency> 
     <groupId>org.scala-lang</groupId> 
     <artifactId>scala-library</artifactId> 
     <version>2.9.2</version> 
    </dependency> 

п помощь в аренде, что я делаю неправильно? Почему я не могу писать через простой Java-код?

Answer 1

Проблема заключается в том, что вы настроили неправильно сериалайзер, это должно быть

props.put("serializer.class", "kafka.serializer.DefaultEncoder"); 

Не «StringEncoder»