Comment puis-je produire des messages avec l'API Kafka 8.2 en Java?

J'essaie de travailler avec l'API kafka en java. J'utilise la dépendance maven suivante:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.8.2.0</version>
</dependency>

J'ai du mal à me connecter à un serveur kafka distant. J'ai changé le kafka du serveur'.l'attribut de port de fichier des propriétés doit être le port 8080. Je peux démarrer le zookeeper et le serveur kafka sans problème. Je peux également utiliser les applications console producer et consumer fournies avec le téléchargement kafka. (Version Scala 2.10)

J'utilise le code client suivant pour créer une télécommande KafkaProducer

Properties propsProducer = new Properties();

propsProducer.put("bootstrap.servers", "172.xx.xx.xxx:8080");
propsProducer.put("key.serializer", org.apache.kafka.common.serialization.ByteArraySerializer.class);
propsProducer.put("value.serializer", org.apache.kafka.common.serialization.ByteArraySerializer.class);
propsProducer.put("topic.metadata.refresh.interval.ms", "0");

KafkaProducer<byte[], byte[]> m_kafkaProducer = new KafkaProducer<byte[], byte[]>(propsProducer);

Une fois que j'ai créé le producteur, je peux exécuter la ligne suivante et obtenir des informations de sujet valides renvoyées, étant donné que strTopic est un nom de sujet existant.

List<PartitionInfo> partitionInfo = m_kafkaProducer.partitionsFor(strTopic);

Lorsque j'essaie d'envoyer un message, je fais ce qui suit:

ProducerRecord<byte[], byte[]> prMessage = new ProducerRecord<byte[],byte[]>(strTopic, strMessage.getBytes());

RecordMetadata futureData = m_kafkaProducer.send(prMessage).get();

L'appel à send() bloque indéfiniment et lorsque je termine manuellement le processus, je vois que l'ERREUR Ferme le socket en raison d'une erreur sur le serveur kafka(IOException, Réinitialisation de la connexion par le pair) erreur.

Aussi, ça ne vaut rien que le host.name, advertised.host.name, et annoncé.les propriétés du port sont toujours commentées sur le serveur.les propriétés de fichier. Oh, et si je change la ligne:

propsProducer.put("bootstrap.servers", "172.xx.xx.xxx:8080");

À

propsProducer.put("bootstrap.servers", "127.0.0.1:8080");

Et exécutez-le sur le même serveur que celui où le serveur kafka est installé, cela fonctionne mais j'essaie de travailler avec lui à distance.

Appréciez toute aide et si je peux clarifier du tout, faites-le moi savoir.