feat: qol improvements

api/api.go

@@ -38,6 +38,7 @@ type api struct {
 	app              *fiber.App
 	config           *dcp.Dcp
 	registerer       *metric.Registerer
+	membershipInfo   *membership.Model
 	bus              EventBus.Bus
 }
 
@@ -94,12 +95,19 @@ func (s *api) info(c *fiber.Ctx) error {
 	if err := c.BodyParser(&req); err != nil {
 		return c.Status(fiber.StatusBadRequest).SendString("invalid request body")
 	}
-	logger.Log.Debug("new info arrived for member: %v/%v", req.MemberNumber, req.TotalMembers)
 
-	s.bus.Publish(helpers.MembershipChangedBusEventName, &membership.Model{
+	newInfo := &membership.Model{
 		MemberNumber: req.MemberNumber,
 		TotalMembers: req.TotalMembers,
-	})
+	}
+
+	if newInfo.IsChanged(s.membershipInfo) {
+		s.membershipInfo = newInfo
+
+		logger.Log.Debug("new info arrived for member: %v/%v", newInfo.MemberNumber, newInfo.TotalMembers)
+
+		s.bus.Publish(helpers.MembershipChangedBusEventName, newInfo)
+	}
 
 	return c.SendString("OK")
 }

config/dcp.go

@@ -50,8 +50,8 @@ type DCPGroup struct {
 }
 
 type DCPListener struct {
-	BufferSize uint       `yaml:"bufferSize"`
 	SkipUntil  *time.Time `yaml:"skipUntil"`
+	BufferSize uint       `yaml:"bufferSize"`
 }
 
 type ExternalDcpConfig struct {
@@ -61,9 +61,9 @@ type ExternalDcpConfig struct {
 type ExternalDcp struct {
 	BufferSize           any               `yaml:"bufferSize"`
 	ConnectionBufferSize any               `yaml:"connectionBufferSize"`
+	Listener             DCPListener       `yaml:"listener"`
 	Group                DCPGroup          `yaml:"group"`
 	ConnectionTimeout    time.Duration     `yaml:"connectionTimeout"`
-	Listener             DCPListener       `yaml:"listener"`
 	Config               ExternalDcpConfig `yaml:"config"`
 }
 
@@ -118,16 +118,16 @@ type Logging struct {
 
 type Dcp struct {
 	ConnectionBufferSize any                `yaml:"connectionBufferSize"`
+	Metric               Metric             `yaml:"metric"`
 	BucketName           string             `yaml:"bucketName"`
-	ScopeName            string             `yaml:"scopeName"`
-	Password             string             `yaml:"password"`
 	RootCAPath           string             `yaml:"rootCAPath"`
 	Username             string             `yaml:"username"`
 	Logging              Logging            `yaml:"logging"`
+	ScopeName            string             `yaml:"scopeName"`
+	Password             string             `yaml:"password"`
 	Metadata             Metadata           `yaml:"metadata"`
 	CollectionNames      []string           `yaml:"collectionNames"`
 	Hosts                []string           `yaml:"hosts"`
-	Metric               Metric             `yaml:"metric"`
 	Checkpoint           Checkpoint         `yaml:"checkpoint"`
 	LeaderElection       LeaderElection     `yaml:"leaderElection"`
 	Dcp                  ExternalDcp        `yaml:"dcp"`

couchbase/membership.go

@@ -266,10 +266,16 @@ func (h *cbMembership) rebalance(instances []Instance) {
 		logger.Log.Error("error while rebalance, self = %v, err: %v", string(h.id), err)
 		panic(err)
 	} else {
-		h.bus.Publish(helpers.MembershipChangedBusEventName, &membership.Model{
+		newInfo := &membership.Model{
 			MemberNumber: selfOrder,
 			TotalMembers: len(instances),
-		})
+		}
+
+		if newInfo.IsChanged(h.info) {
+			logger.Log.Debug("new info arrived for member: %v/%v", newInfo.MemberNumber, newInfo.TotalMembers)
+
+			h.bus.Publish(helpers.MembershipChangedBusEventName, newInfo)
+		}
 
 		h.lastActiveInstances = instances
 	}

dcp.go

@@ -84,14 +84,8 @@ func (s *dcp) SetEventHandler(eventHandler models.EventHandler) {
 	s.eventHandler = eventHandler
 }
 
-func (s *dcp) membershipChangedListener(m *membership.Model) {
-	currentInfo := &membership.Model{
-		MemberNumber: s.vBucketDiscovery.GetMetric().MemberNumber,
-		TotalMembers: s.vBucketDiscovery.GetMetric().TotalMembers,
-	}
-	if s.stream.IsOpen() && currentInfo.IsChanged(m) {
-		s.stream.Rebalance()
-	}
+func (s *dcp) membershipChangedListener(_ *membership.Model) {
+	s.stream.Rebalance()
 }
 
 //nolint:funlen

example/config_dynamic.yml

@@ -1,7 +1,7 @@
 hosts:
   - localhost:8091
 username: user
-password: password
+password: 123456
 bucketName: dcp-test
 dcp:
   group:

example/config_k8s_default.yml

@@ -1,7 +1,7 @@
 hosts:
   - couchbase.couchbase.svc.cluster.local:8091
 username: user
-password: password
+password: 123456
 bucketName: dcp-test
 dcp:
   group:

example/config_k8s_leader_election.yml

@@ -1,7 +1,7 @@
 hosts:
   - couchbase.couchbase.svc.cluster.local:8091
 username: user
-password: password
+password: 123456
 bucketName: dcp-test
 dcp:
   group:

membership/dynamic_membership.go

@@ -16,6 +16,7 @@ func (d *dynamicMembership) GetInfo() *Model {
 	if d.info != nil {
 		return d.info
 	}
+	logger.Log.Info("dynamic membership waiting first request")
 	return <-d.infoChan
 }
 

membership/membership.go

@@ -1,5 +1,7 @@
 package membership
 
+import "github.com/Trendyol/go-dcp/logger"
+
 type Membership interface {
 	GetInfo() *Model
 	Close()
@@ -23,5 +25,10 @@ func (s *Model) IsChanged(other *Model) bool {
 		return true
 	}
 
-	return s.MemberNumber != other.MemberNumber || s.TotalMembers != other.TotalMembers
+	res := s.MemberNumber != other.MemberNumber || s.TotalMembers != other.TotalMembers
+	if !res {
+		logger.Log.Info("membership info not changed")
+	}
+
+	return res
 }