Modify some functions

Signed-off-by: obaydullahmhs <[email protected]>
kubedb · Dec 21, 2023 · 06c468e · 06c468e
1 parent dd488a2
commit 06c468e
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 13 deletions.
diff --git a/kafka/client.go b/kafka/client.go
@@ -48,7 +48,6 @@ type MessageMetadata struct {
 }
 
 func (c *Client) IsDBConnected() (bool, error) {
-	// TODO: try using refreshcontroller
 	controller, err := c.RefreshController()
 	if err != nil || controller == nil {
 		klog.Error(err, "Failed to Get kafka controller")
@@ -61,9 +60,9 @@ func (c *Client) IsDBConnected() (bool, error) {
 		return false, err
 	}
 	if connected {
-		klog.Info(fmt.Sprintf("Successfully connected broker: %s", controller.Addr()))
+		klog.V(5).Info(fmt.Sprintf("Successfully connected broker: %s", controller.Addr()))
 	} else {
-		klog.Info(fmt.Sprintf("Failed to connect broker: %s", controller.Addr()))
+		klog.Error(fmt.Sprintf("Failed to connect broker: %s", controller.Addr()))
 	}
 
 	return connected, nil
@@ -121,7 +120,7 @@ func (a *AdminClient) EnsureKafkaTopic(topic string, topicConfig map[string]*str
 	}
 	return nil
 }
-func (c *Client) DeleteTopic(topics ...string) {
+func (c *Client) DeleteTopics(topics ...string) {
 	broker, err := c.RefreshController()
 	if err != nil {
 		klog.Error(err, "Failed to refresh controller for kafka-health topic")
@@ -164,7 +163,7 @@ func (p *ProducerClient) SendMessageWithProducer(partition int32, topic, key, me
 	return &msgMetadata, nil
 }
 
-func (c ConsumerClient) ConsumeMessages(partition int32, topic string, offset int64, signal *chan bool, message *chan MessageMetadata) error {
+func (c *ConsumerClient) ConsumeMessages(partition int32, topic string, offset int64, signal *chan bool, message *chan MessageMetadata) error {
 	var err error
 	var partitionConsumer kafkago.PartitionConsumer
 	partitionConsumer, err = c.ConsumePartition(topic, partition, offset)
@@ -179,7 +178,7 @@ func (c ConsumerClient) ConsumeMessages(partition int32, topic string, offset in
 		case <-*signal:
 			return nil
 		case err := <-partitionConsumer.Errors():
-			klog.Info(fmt.Sprintf("could not process message, err: %s", err.Error()))
+			klog.Error(fmt.Sprintf("could not process message, err: %s", err.Error()))
 			return err
 		case msg := <-partitionConsumer.Messages():
 			msgMetadata := MessageMetadata{

diff --git a/kafka/kubedb_client_builder.go b/kafka/kubedb_client_builder.go
@@ -66,7 +66,7 @@ func (o *KubeDBClientBuilder) GetConfig() (*kafkago.Config, error) {
 	clientConfig := kafkago.NewConfig()
 	if !o.db.Spec.DisableSecurity {
 		if o.db.Spec.AuthSecret == nil {
-			klog.Info("Authsecret not set")
+			klog.Info("Auth-secret not set")
 			return nil, errors.New("auth-secret is not set")
 		}
 
@@ -77,10 +77,10 @@ func (o *KubeDBClientBuilder) GetConfig() (*kafkago.Config, error) {
 		}, authSecret)
 		if err != nil {
 			if kerr.IsNotFound(err) {
-				klog.Error(err, "Authsecret not found")
+				klog.Error(err, "Auth-secret not found")
 				return nil, errors.New("auth-secret is not found")
 			}
-			klog.Error(err, "Failed to get authsecret")
+			klog.Error(err, "Failed to get auth-secret")
 			return nil, err
 		}
 
@@ -136,7 +136,6 @@ func (o *KubeDBClientBuilder) GetConfig() (*kafkago.Config, error) {
 }
 
 func (o *KubeDBClientBuilder) GetKafkaClient() (*Client, error) {
-
 	clientConfig, err := o.GetConfig()
 	if err != nil {
 		return nil, err
@@ -155,7 +154,6 @@ func (o *KubeDBClientBuilder) GetKafkaClient() (*Client, error) {
 }
 
 func (o *KubeDBClientBuilder) GetKafkaProducerClient() (*ProducerClient, error) {
-
 	clientConfig, err := o.GetConfig()
 	if err != nil {
 		return nil, err
@@ -174,7 +172,6 @@ func (o *KubeDBClientBuilder) GetKafkaProducerClient() (*ProducerClient, error)
 }
 
 func (o *KubeDBClientBuilder) GetKafkaAdminClient() (*AdminClient, error) {
-
 	clientConfig, err := o.GetConfig()
 	if err != nil {
 		return nil, err
@@ -193,7 +190,6 @@ func (o *KubeDBClientBuilder) GetKafkaAdminClient() (*AdminClient, error) {
 }
 
 func (o *KubeDBClientBuilder) GetKafkaConsumerClient() (*ConsumerClient, error) {
-
 	clientConfig, err := o.GetConfig()
 	if err != nil {
 		return nil, err