improve some logs and fix cluster sync to actually wait

pkg/kafka/client/request_packet_handler.go

@@ -18,7 +18,7 @@ type RequestPacketHandler struct {
 func (rh *RequestPacketHandler) HandleRequest(client *Client, inPacket *netCodec.Packet) error {
 	packetReader := netCodec.NewPacketReader(inPacket)
 
-	log := rh.Log.WithValues("request_api_key", inPacket.ReqHeader.Key, "request_api_version", inPacket.ReqHeader.Version, "correlation_id", inPacket.ReqHeader.CorrelationId)
+	log := rh.Log.WithValues("from-address", client.conn.RemoteAddr().String(), "request_api_key", inPacket.ReqHeader.Key, "request_api_version", inPacket.ReqHeader.Version, "correlation_id", inPacket.ReqHeader.CorrelationId)
 
 	decoderMap, ok := codec.MessageDecoderMapping[inPacket.ReqHeader.Key]
 	if !ok {
@@ -58,7 +58,7 @@ func (rh *RequestPacketHandler) HandleRequest(client *Client, inPacket *netCodec
 	}
 
 	// TODO: figure out throttling stuff
-	log.Info("handling packet")
+	log.V(1).Info("handling packet")
 	respMessage, err := inHandler.Handle(client.Broker, log, reqMessage)
 	if err != nil {
 		return fmt.Errorf("error handling packet: %w", err)

pkg/kafka/logical_broker/broker.go

@@ -158,7 +158,7 @@ func (b *Broker) InitClusters() error {
 	if topic == nil {
 		_, err := cluster1.APICreateTopic("test1", 1, 1, map[string]*string{})
 		if err != nil {
-			return err
+			return fmt.Errorf("err creating topic test1: %w", err)
 		}
 	}
 
@@ -170,7 +170,7 @@ func (b *Broker) InitClusters() error {
 	if topic == nil {
 		_, err := cluster2.APICreateTopic("test2", 1, 1, map[string]*string{})
 		if err != nil {
-			return err
+			return fmt.Errorf("err creating topic test2: %w", err)
 		}
 	}
 
@@ -182,7 +182,7 @@ func (b *Broker) InitClusters() error {
 	if topic == nil {
 		_, err := cluster3.APICreateTopic("test3", 1, 3, map[string]*string{})
 		if err != nil {
-			return err
+			return fmt.Errorf("err creating topic test3: %w", err)
 		}
 	}
 

pkg/kafka/logical_broker/cluster.go

@@ -23,8 +23,9 @@ type Cluster struct {
 	Name string
 	log  logr.Logger
 
-	syncedOnce sync.Once
-	syncedChan chan struct{}
+	topicConfigSyncedOnce sync.Once
+	topicLeaderSyncedOnce sync.Once
+	syncedChan            chan struct{}
 
 	controller *Controller
 
@@ -76,6 +77,8 @@ func (c *Cluster) initFranzKafkaClient(addrs []string) error {
 }
 
 func (c *Cluster) WaitSynced() {
+	// Two loads, one for topic configs and one for leaders
+	<-c.syncedChan
 	<-c.syncedChan
 }
 

pkg/kafka/logical_broker/cluster_consomer_topic_leader.go

@@ -79,7 +79,7 @@ func (c *Cluster) ConsumeTopicLeaders() {
 				}
 			}
 
-			c.shouldBeSynced(highWaterMarks, lowWaterMarks)
+			c.shouldBeSynced(&c.topicLeaderSyncedOnce, highWaterMarks, lowWaterMarks)
 		}
 	}
 }

pkg/kafka/logical_broker/cluster_consumer_topic_config.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"os"
+	"sync"
 
 	"github.com/rmb938/krouter/pkg/kafka/logical_broker/topics"
 	"github.com/twmb/franz-go/pkg/kgo"
@@ -90,12 +91,12 @@ func (c *Cluster) ConsumeTopicConfigs() {
 				}
 			}
 
-			c.shouldBeSynced(highWaterMarks, lowWaterMarks)
+			c.shouldBeSynced(&c.topicConfigSyncedOnce, highWaterMarks, lowWaterMarks)
 		}
 	}
 }
 
-func (c *Cluster) shouldBeSynced(highWaterMarks, lowWaterMarks map[int32]int64) {
+func (c *Cluster) shouldBeSynced(once *sync.Once, highWaterMarks, lowWaterMarks map[int32]int64) {
 	synced := true
 	for partitionIndex, highMark := range highWaterMarks {
 		if lowMark, ok := lowWaterMarks[partitionIndex]; ok {
@@ -106,7 +107,7 @@ func (c *Cluster) shouldBeSynced(highWaterMarks, lowWaterMarks map[int32]int64)
 	}
 
 	if synced {
-		c.syncedOnce.Do(func() {
+		once.Do(func() {
 			c.syncedChan <- struct{}{}
 		})
 	}