Le producteur de Kafka ne peut pas envoyer de données au serveur

Question

Voici mon code. Je suis en mesure de créer le sujet mais je ne suis pas en mesure d'envoyer des données à l'intérieur du sujet en raison d'une raison quelconque. Je reçois ces erreurs après une longue période. J'utilise version kafka 2.11-0.8.2.1

org.apache.kafka.clients.producer.KafkaProducer$FutureFailure@5474c6c 
org.apache.kafka.clients.producer.KafkaProducer$FutureFailure@4b6995df 

Ceci est le fichier server.log pour kafka

[2016-12-27 21:05:54,873] ERROR Closing socket for /127.0.0.1 because of error (kafka.network.Processor) 
java.io.IOException: An established connection was aborted by the software in your host machine 
at sun.nio.ch.SocketDispatcher.read0(Native Method) 
at sun.nio.ch.SocketDispatcher.read(Unknown Source) 
at sun.nio.ch.IOUtil.readIntoNativeBuffer(Unknown Source) 
at sun.nio.ch.IOUtil.read(Unknown Source) 
at sun.nio.ch.SocketChannelImpl.read(Unknown Source) 
at kafka.utils.Utils$.read(Utils.scala:380) 
at kafka.network.BoundedByteBufferReceive.readFrom(BoundedByteBufferReceive.scala:54) 
at kafka.network.Processor.read(SocketServer.scala:444) 
at kafka.network.Processor.run(SocketServer.scala:340) 
at java.lang.Thread.run(Unknown Source) 
[2016-12-27 21:07:54,727] INFO Closing socket connection to /127.0.0.1. (kafka.network.Processor) 
[2016-12-27 21:16:08,559] INFO Closing socket connection to /127.0.0.1. (kafka.network.Processor) 

Voici mon code java pour envoyer des nombres entiers au système kafka:

Properties props = new Properties(); 
    props.put("bootstrap.servers", "localhost: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"); 
    props.put("timeout.ms", "50"); 

    Producer<String, String> producer = new KafkaProducer<>(props); 
     for(int i = 0; i < 2; i++) 
      System.out.println(producer.send(new ProducerRecord<String, String>("testtopic", Integer.toString(i), 
        Integer.toString(i))).toString()); 

producer.close(); 

Voici pom.xml

<dependencies> 
    <dependency> 
    <groupId>org.apache.kafka</groupId> 
    <artifactId>kafka-clients</artifactId> 
    <version>0.10.1.0</version> 
</dependency> 
<dependency> 
    <groupId>org.apache.kafka</groupId> 
    <artifactId>kafka_2.11</artifactId> 
    <version>0.8.2.1</version> 
</dependency> 
<dependency> 
    <groupId>org.slf4j</groupId> 
    <artifactId>slf4j-simple</artifactId> 
    <version>1.6.4</version> 
</dependency> 
<dependency> 
    <groupId>log4j</groupId> 
    <artifactId>log4j</artifactId> 
    <version>1.2.16</version> 
    <exclusions> 
    <exclusion> 
     <groupId>javax.jms</groupId> 
     <artifactId>jms</artifactId> 
    </exclusion> 
    </exclusions> 
</dependency> 
</dependencies> 

Answer 1

J'ai dégradé la version de Kafka à kafka_2.10-0.9.0.0 et les propriétés suivantes fonctionnent avec elle.

Properties props = new Properties(); 
    props.put("metadata.broker.list", "localhost: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"); 
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); 
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); 
    props.put("serializer.class", "kafka.serializer.StringEncoder"); 
    ProducerConfig producerConfig = new ProducerConfig(props); 
    kafka.javaapi.producer.Producer<String, String> producer 
    = new kafka.javaapi.producer.Producer<String, String>(producerConfig); 

Mon fichier pom.xml est la suivante:

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> 
<modelVersion>4.0.0</modelVersion> 
<groupId>TwitterKafkaPostgre</groupId> 
<artifactId>TwitterKafkaPostgre</artifactId> 
<version>0.0.1-SNAPSHOT</version> 
<dependencies> 
    <dependency> 
    <groupId>com.twitter</groupId> 
    <artifactId>hbc-core</artifactId> <!-- or hbc-twitter4j --> 
    <version>2.2.0</version> <!-- or whatever the latest version is --> 
    </dependency> 
    <dependency> 
    <groupId>org.apache.kafka</groupId> 
    <artifactId>kafka-clients</artifactId> 
    <version>0.9.0.0</version> 
</dependency> 
<dependency> 
    <groupId>org.apache.kafka</groupId> 
    <artifactId>kafka_2.11</artifactId> 
    <version>0.9.0.0</version> 
</dependency> 
<dependency> 
    <groupId>log4j</groupId> 
    <artifactId>log4j</artifactId> 
    <version>1.2.16</version> 
    <exclusions> 
    <exclusion> 
     <groupId>javax.jms</groupId> 
     <artifactId>jms</artifactId> 
    </exclusion> 
    </exclusions> 
</dependency> 
<dependency> 
    <groupId>org.slf4j</groupId> 
    <artifactId>slf4j-simple</artifactId> 
    <version>1.6.4</version> 
</dependency> 
<dependency> 
    <groupId>com.google.guava</groupId> 
    <artifactId>guava</artifactId> 
    <version>18.0</version> 
</dependency> 

Answer 2

Rien ne se distingue sauf

props.put("timeout.ms", "50"); 

délai de demande devrait être plus que l'intervalle d'interrogation par défaut qui, par défaut est à 5 minutes de Kafka. Donc, je suppose que si vous laissez la valeur par défaut (qui est juste au-dessus de 5 minutes), cela devrait fonctionner.