How to check whether Kafka Server is running

Question

I want to ensure whether kafka server is running or not before starting production and consumption jobs  It is in windows environment and here s my kafka server s code in 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 lt String gt  option   Option empty    KafkaConfig config   new KafkaConfig kafka           KafkaServer server   new KafkaServer config  new CurrentTime    option   server startup      In this case if  server    null  is not enough because it is always true  So is there any way to know that my kafka server is running and ready for producer  It is necessary for me to check this because it causes loss of some starting data packets

User · Answer

Paul s answer is very good and it is actually how Kafka  amp  Zk work together from a broker point of view    I would say that another easy option to check if a Kafka server is running is to create a simple KafkaConsumer pointing to the cluste and try some action  for example  listTopics    If kafka server is not running  you will get a TimeoutException and then you can use a try-catch sentence     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

User · Answer

Firstly you need to create AdminClient bean     Bean  public AdminClient adminClient       Map lt String  Object gt  configs   new HashMap lt  gt        configs put AdminClientConfig BOOTSTRAP SERVERS CONFIG     StringUtils arrayToCommaDelimitedString new Object    your bootstrap server address        return AdminClient create configs        Then  you can use this script   while  true       Map lt String  ConsumerGroupDescription gt  groupDescriptionMap            adminClient describeConsumerGroups Collections singletonList groupId             all             get 10  TimeUnit SECONDS       ConsumerGroupDescription consumerGroupDescription   groupDescriptionMap get groupId       log debug  Kafka consumer group      state                       groupId                  consumerGroupDescription state         if  consumerGroupDescription state   equals ConsumerGroupState STABLE             boolean isReady   true          for  MemberDescription member   consumerGroupDescription members                  if  member assignment      null    member assignment   topicPartitions   isEmpty                  isReady   false                                   if  isReady                break                                  log debug  Kafka consumer group      is not ready  Waiting      groupId           TimeUnit SECONDS sleep 1       This script will check the state of the consumer group every second till the state will be STABLE  Because all consumers assigned to topic partitions  you can conclude that server is running and ready

User · Answer

The good option is to use AdminClient as below before starting to produce or consume the 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

User · Answer

you can use below code to check for brokers available if server is running       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  gt  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

User · Answer

I found an event OnError  in confluent Kafka    consumer OnError    Consumer OnError    private void Consumer OnError object sender  Error e                Debug Log  connection error     e Reason           ConsumerConnectionError e           And its documentation in 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 lt Error gt  OnError

User · Answer

All Kafka brokers must be assigned a broker id  On startup a broker will create an ephemeral node in Zookeeper with a path of  broker ids  id  As the node is ephemeral it will be removed as soon as the broker disconnects  e g  by shutting down   You can view the list of the ephemeral broker nodes like so   echo dump   nc localhost 2181   grep brokers   The ZooKeeper client interface exposes a number of commands  dump lists all the sessions and ephemeral nodes for the cluster   Note  the above assumes    You re running ZooKeeper on the default port  2181  on localhost  and that localhost is the leader for the cluster Your zookeeper connect Kafka config doesn t specify a chroot env for your Kafka cluster i e  it s just host port and not host port path

User · Answer

You can install Kafkacat tool on your machine For example on Ubuntu You can install it using apt-get install kafkacat  once kafkacat is installed then you can use following command to connect it kafkacat -b  lt your-ip-address gt   lt kafka-port gt  -t test-topic   Replace  lt your-ip-address gt  with your machine ip  lt kafka-port gt  can be replaced by the port on which kafka is running  Normally it is 9092  once you run the above command and if kafkacat is able to make the connection then it means that kafka is up and running

User · Answer

For Linux   ps aux   grep kafka  see if kafka properties are shown in the results  E g   path to kafka server properties

User · Answer

I used the 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 lt String gt  names   topics names   get        if  names isEmpty                     case  if no topic found            return true    catch  InterruptedException   ExecutionException e           Kafka is not available

[java] How to check whether Kafka Server is running?

Examples related to java

Examples related to apache-kafka

Examples related to kafka-producer-api