Is there a way to check server health or connectivity?

[AngryPirate]

I would like to know this since once I start writing messages to disk once they have 'timed out' on the producer channel, which usually happens when the kafka server is down or unreachable, to check when I can re-queue these messages into the producer.

Basically, a way to periodically check to see if I can connect to the kafka server with my current ConfigMap.

I tried using the AdminClient to create a topic as a check of health, but this seems to exit the program after one failure or just hang up the calling go routine.
A quick, ping test would be useful, using your current ConfigMap values or perhaps there is a better way that already exists?

Thanks for any help.

• Scott

UPDATE
This is what I did... I got an AdminClient, fetched the Broker metadata and listed the brokerIDs. If no metadata or brokerIDs returned, then the server was still down.

func getAdminClient() (*kafka.AdminClient, error) {	var funcName = "getAdminClient()"	if adminClient != nil {	logger.Debugf("%v::AdminClient already present", funcName)	return adminClient, nil	}	if adminConfigMap == nil {	adminConfigMap = getAdminConfigMap() // Get the kafka.ConfigMap	}	logger.Debugf("%v::Getting AdminClient", funcName)	ac, err := kafka.NewAdminClient(adminConfigMap)	if err == nil {	logger.Debugf("%v::Got AdminClient", funcName)	adminClient = ac	} else {	logger.Debugf("%v::unable to get AdminClient from the server. It must still be down. Error: %v", funcName, err)	}	return adminClient, err } func getBrokerList(topic *string) (brokers []int32, err error) {	var funcName = "getBrokerList()"	ac, err := getAdminClient()	if err == nil {	// unc (a *AdminClient) GetMetadata(topic *string, allTopics bool, timeoutMs int) (*Metadata, error) {	logger.Tracef("%v::Getting metadata", funcName)	md, err := ac.GetMetadata(topic, true, 5000) // 5 seconds	if err != nil {	//ac.Close() // only close if you aren't going to reuse the AdminClient for the duration of your program	return nil, err	} else {	logger.Tracef("%v::Got metadata: %v", funcName, md)	}	brokers = make([]int32, len(md.Brokers))	for i, mdBroker := range md.Brokers {	brokers[i] = mdBroker.ID	}	} else {	//ac.Close() // only close if you aren't going to reuse the AdminClient for the duration of your program	return nil, err	}	//ac.Close(). // only close if you aren't going to reuse the AdminClient for the duration of your program	return brokers, nil } func main() { topicName := "myTopic" brokerIDs, err := getBrokerList(&topicName) isServerHealthy := false if err != nil {	logger.Debugf("%v::unable to get metadata or brokerID list. The kafka server must still be down. Error: %v", funcName, err)	} else {	isServerHealthy = true	logger.Debugf("%v::Broker IDs: %v", funcName, brokerIDs)	} } 

my ConfigMap properties were all of the connection properties in my ConfigMap that I use to publish events, with a couple additional properties that are:

adminConfigMap.Set("request.timeout.ms=1000") adminConfigMap.Set("connections.max.idle.ms=2000") 

Note: adminConfigMap is my kafka.ConfigMap object I use for doing AdminClient stuff.