kafka: CheckPartitions() method for health checks

kafka/kafka.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/abh/certman"
 	"github.com/segmentio/kafka-go"
+	"go.ntppool.org/common/logger"
 )
 
 const (
@@ -203,3 +204,17 @@ func (k *Kafka) NewWriter(topic string) (*kafka.Writer, error) {
 
 	return w, nil
 }
+
+func (k *Kafka) CheckPartitions() error {
+	partitions, err := k.conn.ReadPartitions()
+	if err != nil {
+		return err
+	}
+
+	// should this result in an error?
+	if len(partitions) == 0 {
+		log := logger.Setup()
+		log.Info("kafka connection has no partitions")
+	}
+	return nil
+}