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<String> 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.
This question is related to
java
apache-kafka
kafka-producer-api
Firstly you need to create AdminClient bean:
@Bean
public AdminClient adminClient(){
Map<String, Object> configs = new HashMap<>();
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<String, ConsumerGroupDescription> 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.
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;
}
For Linux, "ps aux | grep kafka" see if kafka properties are shown in the results. E.g. /path/to/kafka/server.properties
Paul's answer is very good and it is actually how Kafka & 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()
}
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 > 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;
}
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:
2181
) on localhost
, and that localhost
is the leader for the clusterzookeeper.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
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<Error> OnError;
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<String> names = topics.names().get();
if (names.isEmpty())
{
// case: if no topic found.
}
return true;
}
catch (InterruptedException | ExecutionException e)
{
// Kafka is not available
}
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 <your-ip-address>:<kafka-port> -t test-topic
once you run the above command and if kafkacat is able to make the connection then it means that kafka is up and running
Source: Stackoverflow.com