Exemple d'erreur Apache Kafka: Échec de l'envoi du message après 3 essais

Je suis en cours d'exécution cette kafka producteur exemple mentionnés dans son site

Le code:

public class TestProducer {

    public static void main(String[] args) {
        long events = Long.parseLong(args[0]);
        Random rnd = new Random();
        Properties props = new Properties();
        props.put("metadata.broker.list", "host.broker-1:9093, host.broker-2:9093, host.broker-3:9095");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("partitioner.class", "test.app.SimplePartitioner");
        props.put("request.required.acks", "1");
        ProducerConfig config = new ProducerConfig(props);
        Producer<String, String> producer = new Producer<String, String>(config);
        for (long nEvents = 0; nEvents < events; nEvents++) { 
               long runtime = new Date().getTime();  
               String ip = "192.168.2." + rnd.nextInt(255); 
               String msg = runtime + ",www.example.com," + ip; 
               KeyedMessage<String, String> data = new KeyedMessage<String, String>("page_visits", ip, msg);
               producer.send(data);
        }
        producer.close();
    }
}

public class SimplePartitioner implements Partitioner{
    public SimplePartitioner (VerifiableProperties props) {

    }

    public int partition(Object key, int a_numPartitions) {
        int partition = 0;
        String stringKey = (String) key;
        int offset = stringKey.lastIndexOf('.');
        if (offset > 0) {
           partition = Integer.parseInt( stringKey.substring(offset+1)) % a_numPartitions;
        }
       return partition;
  }

}

Plus de détails:

Je suis en cours d'exécution de cette application sur un ordinateur hôte(appel est producteur) qui est à distance de l'hôte de courtier[1-3]

Je peux faire un ping et ssh le courtier d'accueil du producteur de l'hôte.
Fourni dans l'annonce.de l'hôte.nom du serveur.propriétés (ils sont nommés en tant que serveur[1-3].propriétés dans les courtiers respectivement

Les propriétés:

broker.id=1
port=9093
host.name=host.broker.internal.name
advertised.host.name=host-broker1
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/1/kafka-logs-1,/data/2/kafka-logs-2
num.partitions=1
num.recovery.threads.per.data.dir=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
log.cleaner.enable=false
zookeeper.connect=zk1:2181,zk2:2181,zk3:2181
zookeeper.connection.timeout.ms=6000

Aucune idée sur comment résoudre cette erreur?

source d'informationauteur krish7919

apache-kafka

9

J'ai eu ces erreurs lors de l'exécution d'un Kafka producteur:
```
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
```
Trouvé une solution:

Sur mon Mac box, après j'ai télécharger le scala-2.10 et kafka_2.10-0.8.1dans le kafka_2.10-0.8.1 répertoire, chaque chose est bien quand j'ai commencer zookeeper, kafka serveur, et de créer un test de sujet. Puis-je commencer un producteur pour le test sujet. mais il y a une erreur:
```
yhuangMac:kafka_2.10-0.8.1 yhuang$ ./bin/kafka-console-producer.sh –broker-list localhost:9092 –topic test
SLF4J: Failed to load class “org.slf4j.impl.StaticLoggerBinder”.
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
```
La raison en est que, dans le kafka répertoire libs, le kafka version fichier zip seulement inclus le fichier jar de slf4j-api, ils ont manqué d'un fichier jar: slf4j-nop.jar, nous devons donc aller à http://www.slf4j.orgtélécharger slf4j-1.7.7.zippuis décompressez-le, copiez le slf4j-api-1.7.7, slf4j-nop-1.7.7.jar dans kafka répertoire libs.

Redémarrer kafka producteur de nouveau, maintenant, aucune erreur n'est signalée.

Source: SOLUTION
3

Vous devez ajouter le SLF4j implémentation de journalisation. si vous utilisez maven comme l'outil de génération essayez d'ajouter ce qui suit à votre pom.xml et voir si cela fonctionne ..
```
    <dependency>
      <groupId>org.slf4j</groupId>
      <artifactId>slf4j-log4j12</artifactId>
      <version>1.7.5</version>
    </dependency>
```
2

C'est une solution à l'exception de la question initiale posée par Krish: "kafka.commun.FailedToSendMessageException: Impossible d'envoyer des messages après 3 essais."

La FAQ ici et ici dit que votre nom d'hôte doit être configuré correctement. Je n'ai pas connu cette condition. Mais j'ai trouvé un autre état, lors d'un kafka producteur donne ce message d'erreur: lorsque votre clé de partition dans le producteur est faux. C'est, si vous avez un sujet avec une partition, alors la partition de la clé dans le producteur peut être null(le message est envoyé à une partition aléatoire) ou 0(partitions de kafka sont numérotés à partir de 0). Si vous essayez d'utiliser une clé de partition de 1, cette exception est levée dans le producteur. Ou si vous avez 3 partitions dans le sujet, et que vous utilisez une clé de partition de 3(clé de 3 est invalide en raison valide les numéros de partition sont 0,1,2), cette exception est levée. Cette erreur est cohérente lorsque le numéro de la partition dans le producteur de la méthode send() ne correspond pas à la gamme de partitions dans le sujet.
J'ai utilisé kafka version 0.8.2. Le client API que j'ai utilisé était le paquet de kafka.javaapi.le producteur.Le producteur.
2

Cela peut se produire si le client ne peut pas toucher à la FOIS le nom d'hôte et l'adresse IP de l'kafka courtier.

Faire une entrée pour les clients \etc\hosts ou C:\Windows\System32\drivers\etc\hosts et qui a résolu ce problème pour moi.

J'ai obtenu l'erreur d'apache kafka:

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details

Ma Configuration:

OS: Ubuntu 14.04
sbt: sbt launcher version 0.13.5
scala: Scala code runner version 2.9.2

A été en mesure de le réparer avec ces commandes:

cd /home/el/apachekafka/kafka_2.10-0.8.1.1/libs
wget http://www.slf4j.org/dist/slf4j-1.7.7.tar.gz
tar -xvf slf4j-1.7.7.tar.gz
cd /home/el/apachekafka/kafka_2.10-0.8.1.1/libs/slf4j-1.7.7
cp slf4j-api-1.7.7.jar ..
cp slf4j-nop-1.7.7.jar ..

Puis exécutez de nouveau la commande et le producteur ne renvoie pas d'erreur.

1

ForHDP kafka utilisation courtier port: 6667

Autonome, pour kafka utilisation courtier port: 9092

L'erreur était à cause de la port pas de qui nous avons été en utilisant (HDP utilise 6667, mais nous avons été à l'aide de 9092)

bin/kafka-console-producer.sh --courtier-liste de broker-ip:9092 --rubrique test //ne fonctionne pas

bin/kafka-console-producer.sh --courtier-liste de broker-ip:6667 --sujet de test /de travail /

lien: Kafka console producteur d'Erreur dans hortonworks a HDP 2.3 Sandbox
0

Pas certain, mais une possibilité pourrait être que le sujet n'est pas créé sur Kafka.

Vérifier l'INTERFACE web pour kafka et assurez-vous que le sujet qui vous aide à savoir "page_visits" pour envoyer les données est créée.

Si non il est très facile de créer le sujet à l'aide de l'interface graphique.

Je suis tombé sur cette erreur dans hortonworks a HDP 2.2 où le port par défaut est définie à 6667.
Si votre kafka serveur est en cours d'exécution sur HDP sandbox la résolution est de mettre en
les métadonnées.courtier.liste 10.0.2.15:6667 Veuillez suivre ce code.

            Properties props = new Properties();
            props.put("metadata.broker.list", "10.0.2.15:6667");
            props.put("serializer.class", "kafka.serializer.StringEncoder");
            //props.put("producer.type","async");
            props.put("request.required.acks", "1");
            ProducerConfig config = new ProducerConfig(props);

            Producer<String, String> producer = new Producer<String, String>(config);
            try{        
            producer.send(new KeyedMessage<String, String>("zerg.hydra", jsonPayload));

            producer.close();
        }catch(Exception e){
            e.printStackTrace();
        }

Vous devez vous connecter pour publier un commentaire.