Comment vérifier si Kafka Serveur est en cours d'exécution?

Je veux m'assurer que kafka serveur est en cours d'exécution ou non avant le démarrage de la production et de la consommation des emplois. C'est dans l'environnement de windows et voici mon kafka du serveur de code dans eclipse...

Properties kafka = new Properties();
kafka.setProperty("broker.id", "1");
kafka.setProperty("port", "9092");
kafka.setProperty("log.dirs", "D://workspace//");
kafka.setProperty("zookeeper.connect", "localhost:2181");    
Option<String> option = Option.empty();
KafkaConfig config = new KafkaConfig(kafka);        
KafkaServer server = new KafkaServer(config, new CurrentTime(), option);
server.startup();

Dans ce cas if (server != null) n'est pas suffisant, car il est toujours vrai. Donc, il y a aucun moyen de savoir que mon kafka serveur est en cours d'exécution et prêt pour le producteur. Il est nécessaire pour moi de vérifier, car cela provoque la perte de quelques paquets de données.

Grâce.

Quel système d'exploitation utilisez-vous?
J'ai mis à jour ma question.

InformationsquelleAutor Khan | 2016-06-20

33

Tous Kafka courtiers doivent être attribués à un broker.id. Au démarrage d'un courtier va créer un éphémère nœud dans Zookeeper avec un chemin de /broker/ids/$id. Comme le nœud est éphémère, elle sera supprimée dès que le courtier se déconnecte, par exemple en fermant.

Vous pouvez afficher la liste de l'éphémère, courtier nœuds comme suit:

echo dump | nc localhost 2181 | grep brokers

La Gardienne de l'interface client expose un certain nombre de commandes; dump listes de toutes les sessions et éphémère des nœuds du cluster.

Note, le ci-dessus suppose:
- Vous êtes en cours d'exécution ZooKeeper sur le port par défaut (2181) sur localhost, et que localhost est le chef de file du cluster
- Votre zookeeper.connect Kafka config ne spécifie pas de chroot env pour votre Kafka cluster c'est à dire c'est juste host:port et pas host:port/path
InformationsquelleAutor Paul Carey

J'ai utilisé le AdminClient api.

Properties properties = new Properties();
properties.put("bootstrap.servers", "localhost:9092");
properties.put("connections.max.idle.ms", 10000);
properties.put("request.timeout.ms", 5000);
try (AdminClient client = KafkaAdminClient.create(properties))
{
    ListTopicsResult topics = client.listTopics();
    Set<String> names = topics.names().get();
    if (names.isEmpty())
    {
        //case: if no topic found.
    }
    return true;
}
catch (InterruptedException | ExecutionException e)
{
    //Kafka is not available
}

InformationsquelleAutor Mohammad Faisal

3

Paul réponse est très bon et c'est en fait la façon dont Kafka & Zk travailler ensemble à partir d'un courtier point de vue.

Je dirais qu'une autre option facile de vérifier si un Kafka serveur est en cours d'exécution est de créer une simple KafkaConsumer pointant vers le cluste et d'essayer un peu d'action, par exemple, listTopics(). Si kafka serveur n'est pas en cours d'exécution, vous obtiendrez un TimeoutException et puis vous pouvez utiliser un try-catch phrase.
```
  def validateKafkaConnection(kafkaParams : mutable.Map[String, Object]) : Unit = {
    val props = new Properties()
    props.put("bootstrap.servers", kafkaParams.get("bootstrap.servers").get.toString)
    props.put("group.id", kafkaParams.get("group.id").get.toString)
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    val simpleConsumer = new KafkaConsumer[String, String](props)
    simpleConsumer.listTopics()
  }
```
- J'ai commencé le zookeeper pas la Kafka, puis exécuté le kafka-topics.bat --list et j'ai obtenu tous les sujets. Il semble que seule la liste des sujets n'a pas pu vérifier que Kafka est en cours d'exécution.
- Cette commande permettra d'obtenir les informations de métadonnées à partir de la Gardienne (github.com/apache/kafka/blob/0.8.1/core/src/main/scala/kafka/...). La solution que j'ai fourni est d'obtenir l'information à partir de Kafka Cluster (si vous suivez le code source, vous obtiendrez ceci: github.com/apache/kafka/blob/0.10.0/clients/src/main/java/org/...). C'est parce que vous êtes de ne pas obtenir de toute erreur lors de Kafka cluster est en baisse.
InformationsquelleAutor dbustosp

La bonne option est d'utiliser AdminClient comme ci-dessous avant de commencer à produire ou à consommer les messages

private static final int ADMIN_CLIENT_TIMEOUT_MS = 5000;           
 try (AdminClient client = AdminClient.create(properties)) {
            client.listTopics(new ListTopicsOptions().timeoutMs(ADMIN_CLIENT_TIMEOUT_MS)).listings().get();
        } catch (ExecutionException ex) {
            LOG.error("Kafka is not available, timed out after {} ms", ADMIN_CLIENT_TIMEOUT_MS);
            return;
        }

Notez que AdminClient n'est disponible que depuis la version 0.11

InformationsquelleAutor Prashanth Reddy Mamidi

vous pouvez utiliser le code ci-dessous pour vérifier les courtiers disponibles si le serveur est en cours d'exécution.

import org.I0Itec.zkclient.ZkClient;
     public static boolean isBrokerRunning(){
        boolean flag = false;
        ZkClient zkClient = new ZkClient(endpoint.getZookeeperConnect(), 10000);//, kafka.utils.ZKStringSerializer$.MODULE$);
        if(zkClient!=null){
            int brokersCount = zkClient.countChildren(ZkUtils.BrokerIdsPath());
            if(brokersCount > 0){
                logger.info("Following Broker(s) {} is/are available on Zookeeper.",zkClient.getChildren(ZkUtils.BrokerIdsPath()));
                flag = true;    
            }
            else{
                logger.error("ERROR:No Broker is available on Zookeeper.");
            }
            zkClient.close();

        }
        return flag;
    }

Ici, Si j'ai plusieurs courtiers en cours d'exécution déjà des résultats faux serait renvoyé. Votre solution peut fonctionner avec précision pour un seul courtier de l'environnement.
Oui, pour un seul courtier, c'est parfait. BTW, vous n'avez pas mentionné pour plusieurs. permettez-moi de faire un essai alors. vous pouvez vérifier si tout un du courtier est de ne pas vérifier pour les autres, tant et aussi longtemps que tout courtier, vous pouvez ignorer les autres. D'autre part, vous devez vérifier pour la dernière exécution de l'agent en tant que première N courtiers peut-être dans très regrettable affaire.

InformationsquelleAutor usman

J'ai trouvé un événement OnError dans confluentes Kafka:

consumer.OnError += Consumer_OnError;

 private void Consumer_OnError(object sender, Error e)
    {
        Debug.Log("connection error: "+ e.Reason);
        ConsumerConnectionError(e);
    }

Et sa documentation dans le code:

    //
    //Summary:
    //    Raised on critical errors, e.g. connection failures or all brokers down. Note
    //    that the client will try to automatically recover from errors - these errors
    //    should be seen as informational rather than catastrophic
    //
    //Remarks:
    //    Executes on the same thread as every other Consumer event handler (except OnLog
    //    which may be called from an arbitrary thread).
    public event EventHandler<Error> OnError;

InformationsquelleAutor Muhammad Faizan Khan

0

Pour Linux, "ps aux | grep kafka" voir si kafka propriétés sont affichées dans les résultats. E. g. /chemin/vers/kafka/serveur.propriétés
- ils ont spécifié sous Windows non?
InformationsquelleAutor Raffy Arnaez

Vous devez vous connecter pour publier un commentaire.