Erreur de lecture du champ 'topic_metadata': Erreur de lecture sur le tableau de taille 1139567, à seulement 45 octets disponibles

--Consommateur

Properties props = new Properties();
        String groupId = "consumer-tutorial-group";
        List<String> topics = Arrays.asList("consumer-tutorial");
        props.put("bootstrap.servers", "192.168.1.75:9092");
        props.put("group.id", groupId);
        props.put("enable.auto.commit", "true");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        try {
            consumer.subscribe(topics);
            while (true) {

                ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
                for (ConsumerRecord<String, String> record : records)
                    System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());


            }
        } catch (Exception e) {
            System.out.println(e.toString());
        } finally {
            consumer.close();
        }
    }

je suis en train d'écrire exécuter le code ci-dessus,c'est un simple code de la consommation qui tente de lire à partir d'un sujet, mais j'ai eu une étrange exception à la règle et je ne peut pas le manipuler.

org.apache.kafka.common.protocol.types.SchemaException: Error reading field 'topic_metadata': Error reading array of size 1139567, only 45 bytes available

je vous cite aussi mon producteur code

--Producteur

Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.1.7:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<String, String>(props);
        for(int i = 0; i < 100; i++)
            producer.send(new ProducerRecord<String, String>("consumer-tutorial", Integer.toString(i), Integer.toString(i)));

        producer.close();

Ici est kafka configs

--Start zookeeper

bin/zookeeper-server-start.sh config/zookeeper.propriétés

--Start Kafka Serveur

bin/kafka-server-start.sh config/serveur.propriétés

-- Créer un topic

bin/kafka-topics.sh --création --sujet de la consommation-tutoriel --réplication facteur 1 --partitions 3 --zookeeper 192.168.1.75:2181

--Kafka 0.10.0

<dependency>
           <groupId>org.apache.kafka</groupId>
           <artifactId>kafka-clients</artifactId>
           <version>0.10.0.0</version>
   </dependency>
   <dependency>
           <groupId>org.apache.kafka</groupId>
           <artifactId>kafka_2.11</artifactId>
           <version>0.10.0.0</version>
   </dependency>

OriginalL'auteur | 2016-06-02