simplify

9 files changed, 41 insertions, 63 deletions

diff --git a/weed/mq/broker/broker_grpc_assign.go b/weed/mq/broker/broker_grpc_assign.go
index 264565b7b..48a20e334 100644
--- a/weed/mq/broker/broker_grpc_assign.go
+++ b/weed/mq/broker/broker_grpc_assign.go
@@ -14,7 +14,6 @@ import (
 // AssignTopicPartitions Runs on the assigned broker, to execute the topic partition assignment
 func (b *MessageQueueBroker) AssignTopicPartitions(c context.Context, request *mq_pb.AssignTopicPartitionsRequest) (*mq_pb.AssignTopicPartitionsResponse, error) {
 	ret := &mq_pb.AssignTopicPartitionsResponse{}
-	self := pb.ServerAddress(fmt.Sprintf("%s:%d", b.option.Ip, b.option.Port))
 
 	// drain existing topic partition subscriptions
 	for _, assignment := range request.BrokerPartitionAssignments {
@@ -27,7 +26,7 @@ func (b *MessageQueueBroker) AssignTopicPartitions(c context.Context, request *m
 		} else {
 			var localPartition *topic.LocalPartition
 			if localPartition = b.localTopicManager.GetTopicPartition(t, partition); localPartition == nil {
-				localPartition = topic.FromPbBrokerPartitionAssignment(self, partition, assignment, b.genLogFlushFunc(t, assignment.Partition), b.genLogOnDiskReadFunc(t, assignment.Partition))
+				localPartition = topic.NewLocalPartition(partition, b.genLogFlushFunc(t, assignment.Partition), b.genLogOnDiskReadFunc(t, assignment.Partition))
 				b.localTopicManager.AddTopicPartition(t, localPartition)
 			}
 		}
diff --git a/weed/mq/broker/broker_topic_conf_read_write.go b/weed/mq/broker/broker_topic_conf_read_write.go
index 987c60243..dfdf9a708 100644
--- a/weed/mq/broker/broker_topic_conf_read_write.go
+++ b/weed/mq/broker/broker_topic_conf_read_write.go
@@ -88,7 +88,7 @@ func (b *MessageQueueBroker) genLocalPartitionFromFiler(t topic.Topic, partition
 	self := b.option.BrokerAddress()
 	for _, assignment := range conf.BrokerPartitionAssignments {
 		if assignment.LeaderBroker == string(self) && partition.Equals(topic.FromPbPartition(assignment.Partition)) {
-			localPartition = topic.FromPbBrokerPartitionAssignment(b.option.BrokerAddress(), partition, assignment, b.genLogFlushFunc(t, assignment.Partition), b.genLogOnDiskReadFunc(t, assignment.Partition))
+			localPartition = topic.NewLocalPartition(partition, b.genLogFlushFunc(t, assignment.Partition), b.genLogOnDiskReadFunc(t, assignment.Partition))
 			b.localTopicManager.AddTopicPartition(t, localPartition)
 			isGenerated = true
 			break
diff --git a/weed/mq/pub_balancer/balance_brokers_test.go b/weed/mq/pub_balancer/balance_brokers_test.go
index 54667d154..122984f0d 100644
--- a/weed/mq/pub_balancer/balance_brokers_test.go
+++ b/weed/mq/pub_balancer/balance_brokers_test.go
@@ -21,8 +21,6 @@ func TestBalanceTopicPartitionOnBrokers(t *testing.T) {
 			Topic:     topic.Topic{Namespace: "topic1", Name: "topic1"},
 			Partition: topic.Partition{RangeStart: 0, RangeStop: 512, RingSize: 1024},
 		},
-		ConsumerCount: 1,
-		IsLeader:      true,
 	})
 	broker2Stats := &BrokerStats{
 		TopicPartitionCount: 2,
@@ -35,16 +33,12 @@ func TestBalanceTopicPartitionOnBrokers(t *testing.T) {
 			Topic:     topic.Topic{Namespace: "topic1", Name: "topic1"},
 			Partition: topic.Partition{RangeStart: 512, RangeStop: 1024, RingSize: 1024},
 		},
-		ConsumerCount: 1,
-		IsLeader:      true,
 	})
 	broker2Stats.TopicPartitionStats.Set("topic2:0", &TopicPartitionStats{
 		TopicPartition: topic.TopicPartition{
 			Topic:     topic.Topic{Namespace: "topic2", Name: "topic2"},
 			Partition: topic.Partition{RangeStart: 0, RangeStop: 1024, RingSize: 1024},
 		},
-		ConsumerCount: 1,
-		IsLeader:      true,
 	})
 	brokers.Set("broker1", broker1Stats)
 	brokers.Set("broker2", broker2Stats)
diff --git a/weed/mq/pub_balancer/broker_stats.go b/weed/mq/pub_balancer/broker_stats.go
index b4bb28e42..00f1f80ca 100644
--- a/weed/mq/pub_balancer/broker_stats.go
+++ b/weed/mq/pub_balancer/broker_stats.go
@@ -9,15 +9,16 @@ import (
 
 type BrokerStats struct {
 	TopicPartitionCount int32
-	ConsumerCount       int32
+	PublisherCount      int32
+	SubscriberCount     int32
 	CpuUsagePercent     int32
 	TopicPartitionStats cmap.ConcurrentMap[string, *TopicPartitionStats] // key: topic_partition
 	Topics              []topic.Topic
 }
 type TopicPartitionStats struct {
 	topic.TopicPartition
-	ConsumerCount int32
-	IsLeader      bool
+	PublisherCount int32
+	SubscriberCount int32
 }
 
 func NewBrokerStats() *BrokerStats {
@@ -26,15 +27,15 @@ func NewBrokerStats() *BrokerStats {
 	}
 }
 func (bs *BrokerStats) String() string {
-	return fmt.Sprintf("BrokerStats{TopicPartitionCount:%d, ConsumerCount:%d, CpuUsagePercent:%d, Stats:%+v}",
-		bs.TopicPartitionCount, bs.ConsumerCount, bs.CpuUsagePercent, bs.TopicPartitionStats.Items())
+	return fmt.Sprintf("BrokerStats{TopicPartitionCount:%d, Publishers:%d, Subscribers:%d CpuUsagePercent:%d, Stats:%+v}",
+		bs.TopicPartitionCount, bs.PublisherCount, bs.SubscriberCount, bs.CpuUsagePercent, bs.TopicPartitionStats.Items())
 }
 
 func (bs *BrokerStats) UpdateStats(stats *mq_pb.BrokerStats) {
 	bs.TopicPartitionCount = int32(len(stats.Stats))
 	bs.CpuUsagePercent = stats.CpuUsagePercent
 
-	var consumerCount int32
+	var publisherCount, subscriberCount int32
 	currentTopicPartitions := bs.TopicPartitionStats.Items()
 	for _, topicPartitionStats := range stats.Stats {
 		tps := &TopicPartitionStats{
@@ -47,10 +48,11 @@ func (bs *BrokerStats) UpdateStats(stats *mq_pb.BrokerStats) {
 					UnixTimeNs: topicPartitionStats.Partition.UnixTimeNs,
 				},
 			},
-			ConsumerCount: topicPartitionStats.ConsumerCount,
-			IsLeader:      topicPartitionStats.IsLeader,
+			PublisherCount: topicPartitionStats.PublisherCount,
+			SubscriberCount: topicPartitionStats.SubscriberCount,
 		}
-		consumerCount += topicPartitionStats.ConsumerCount
+		publisherCount += topicPartitionStats.PublisherCount
+		subscriberCount += topicPartitionStats.SubscriberCount
 		key := tps.TopicPartition.String()
 		bs.TopicPartitionStats.Set(key, tps)
 		delete(currentTopicPartitions, key)
@@ -59,8 +61,8 @@ func (bs *BrokerStats) UpdateStats(stats *mq_pb.BrokerStats) {
 	for key := range currentTopicPartitions {
 		bs.TopicPartitionStats.Remove(key)
 	}
-	bs.ConsumerCount = consumerCount
-
+	bs.PublisherCount = publisherCount
+	bs.SubscriberCount = subscriberCount
 }
 
 func (bs *BrokerStats) RegisterAssignment(t *mq_pb.Topic, partition *mq_pb.Partition, isAdd bool) {
@@ -74,8 +76,8 @@ func (bs *BrokerStats) RegisterAssignment(t *mq_pb.Topic, partition *mq_pb.Parti
 				UnixTimeNs: partition.UnixTimeNs,
 			},
 		},
-		ConsumerCount: 0,
-		IsLeader:      true,
+		PublisherCount: 0,
+		SubscriberCount: 0,
 	}
 	key := tps.TopicPartition.String()
 	if isAdd {
diff --git a/weed/mq/pub_balancer/repair.go b/weed/mq/pub_balancer/repair.go
index 0ab1a5ea9..0f307c9eb 100644
--- a/weed/mq/pub_balancer/repair.go
+++ b/weed/mq/pub_balancer/repair.go
@@ -14,8 +14,7 @@ func (balancer *Balancer) RepairTopics() []BalanceAction {
 }
 
 type TopicPartitionInfo struct {
-	Leader    string
-	Followers []string
+	Broker string
 }
 
 // RepairMissingTopicPartitions check the stats of all brokers,
@@ -38,11 +37,7 @@ func RepairMissingTopicPartitions(brokers cmap.ConcurrentMap[string, *BrokerStat
 				tpi = &TopicPartitionInfo{}
 				topicPartitionToInfo[topicPartitionStat.Partition] = tpi
 			}
-			if topicPartitionStat.IsLeader {
-				tpi.Leader = broker
-			} else {
-				tpi.Followers = append(tpi.Followers, broker)
-			}
+			tpi.Broker = broker
 		}
 	}
 
diff --git a/weed/mq/topic/local_manager.go b/weed/mq/topic/local_manager.go
index aa2eefcdc..02334a27e 100644
--- a/weed/mq/topic/local_manager.go
+++ b/weed/mq/topic/local_manager.go
@@ -96,8 +96,9 @@ func (manager *LocalTopicManager) CollectStats(duration time.Duration) *mq_pb.Br
 					Namespace: string(localTopic.Namespace),
 					Name:      localTopic.Name,
 				},
-				Partition:     localPartition.Partition.ToPbPartition(),
-				ConsumerCount: localPartition.ConsumerCount,
+				Partition:       localPartition.Partition.ToPbPartition(),
+				PublisherCount:  int32(localPartition.Publishers.Size()),
+				SubscriberCount: int32(localPartition.Subscribers.Size()),
 			}
 			// fmt.Printf("collect topic %+v partition %+v\n", topicPartition, localPartition.Partition)
 		}
diff --git a/weed/mq/topic/local_partition.go b/weed/mq/topic/local_partition.go
index 5df2535a6..8b9970f20 100644
--- a/weed/mq/topic/local_partition.go
+++ b/weed/mq/topic/local_partition.go
@@ -22,10 +22,7 @@ type LocalPartition struct {
 	ListenersCond *sync.Cond
 
 	Partition
-	isLeader        bool
-	FollowerBrokers []pb.ServerAddress
 	LogBuffer       *log_buffer.LogBuffer
-	ConsumerCount   int32
 	Publishers      *LocalPartitionPublishers
 	Subscribers     *LocalPartitionSubscribers
 	FollowerId      int32
@@ -37,11 +34,9 @@ type LocalPartition struct {
 
 var TIME_FORMAT = "2006-01-02-15-04-05"
 
-func NewLocalPartition(partition Partition, isLeader bool, followerBrokers []pb.ServerAddress, logFlushFn log_buffer.LogFlushFuncType, readFromDiskFn log_buffer.LogReadFromDiskFuncType) *LocalPartition {
+func NewLocalPartition(partition Partition, logFlushFn log_buffer.LogFlushFuncType, readFromDiskFn log_buffer.LogReadFromDiskFuncType) *LocalPartition {
 	lp := &LocalPartition{
 		Partition:       partition,
-		isLeader:        isLeader,
-		FollowerBrokers: followerBrokers,
 		Publishers:  NewLocalPartitionPublishers(),
 		Subscribers: NewLocalPartitionSubscribers(),
 	}
@@ -121,15 +116,6 @@ func (p *LocalPartition) GetEarliestInMemoryMessagePosition() log_buffer.Message
 	return p.LogBuffer.GetEarliestPosition()
 }
 
-func FromPbBrokerPartitionAssignment(self pb.ServerAddress, partition Partition, assignment *mq_pb.BrokerPartitionAssignment, logFlushFn log_buffer.LogFlushFuncType, readFromDiskFn log_buffer.LogReadFromDiskFuncType) *LocalPartition {
-	isLeader := assignment.LeaderBroker == string(self)
-	followers := make([]pb.ServerAddress, len(assignment.FollowerBrokers))
-	for i, followerBroker := range assignment.FollowerBrokers {
-		followers[i] = pb.ServerAddress(followerBroker)
-	}
-	return NewLocalPartition(partition, isLeader, followers, logFlushFn, readFromDiskFn)
-}
-
 func (p *LocalPartition) closePublishers() {
 	p.Publishers.SignalShutdown()
 }
diff --git a/weed/pb/mq.proto b/weed/pb/mq.proto
index 45253059f..999985b7d 100644
--- a/weed/pb/mq.proto
+++ b/weed/pb/mq.proto
@@ -97,8 +97,8 @@ message BrokerStats {
 message TopicPartitionStats {
     Topic topic = 1;
     Partition partition = 2;
-    int32 consumer_count = 3;
-    bool is_leader = 4;
+    int32 publisher_count = 3;
+    int32 subscriber_count = 4;
 }
 
 
diff --git a/weed/pb/mq_pb/mq.pb.go b/weed/pb/mq_pb/mq.pb.go
index b5182e6af..8beccd946 100644
--- a/weed/pb/mq_pb/mq.pb.go
+++ b/weed/pb/mq_pb/mq.pb.go
@@ -476,10 +476,10 @@ type TopicPartitionStats struct {
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
-	Topic         *Topic     `protobuf:"bytes,1,opt,name=topic,proto3" json:"topic,omitempty"`
-	Partition     *Partition `protobuf:"bytes,2,opt,name=partition,proto3" json:"partition,omitempty"`
-	ConsumerCount int32      `protobuf:"varint,3,opt,name=consumer_count,json=consumerCount,proto3" json:"consumer_count,omitempty"`
-	IsLeader      bool       `protobuf:"varint,4,opt,name=is_leader,json=isLeader,proto3" json:"is_leader,omitempty"`
+	Topic           *Topic     `protobuf:"bytes,1,opt,name=topic,proto3" json:"topic,omitempty"`
+	Partition       *Partition `protobuf:"bytes,2,opt,name=partition,proto3" json:"partition,omitempty"`
+	PublisherCount  int32      `protobuf:"varint,3,opt,name=publisher_count,json=publisherCount,proto3" json:"publisher_count,omitempty"`
+	SubscriberCount int32      `protobuf:"varint,4,opt,name=subscriber_count,json=subscriberCount,proto3" json:"subscriber_count,omitempty"`
 }
 
 func (x *TopicPartitionStats) Reset() {
@@ -528,18 +528,18 @@ func (x *TopicPartitionStats) GetPartition() *Partition {
 	return nil
 }
 
-func (x *TopicPartitionStats) GetConsumerCount() int32 {
+func (x *TopicPartitionStats) GetPublisherCount() int32 {
 	if x != nil {
-		return x.ConsumerCount
+		return x.PublisherCount
 	}
 	return 0
 }
 
-func (x *TopicPartitionStats) GetIsLeader() bool {
+func (x *TopicPartitionStats) GetSubscriberCount() int32 {
 	if x != nil {
-		return x.IsLeader
+		return x.SubscriberCount
 	}
-	return false
+	return 0
 }
 
 type PublisherToPubBalancerRequest struct {
@@ -2762,7 +2762,7 @@ var file_mq_proto_rawDesc = []byte{
 	0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x21, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61,
 	0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x50, 0x61, 0x72,
 	0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x53, 0x74, 0x61, 0x74, 0x73, 0x52, 0x05, 0x76, 0x61, 0x6c,
-	0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0xbb, 0x01, 0x0a, 0x13, 0x54, 0x6f, 0x70, 0x69, 0x63,
+	0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0xcb, 0x01, 0x0a, 0x13, 0x54, 0x6f, 0x70, 0x69, 0x63,
 	0x50, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x53, 0x74, 0x61, 0x74, 0x73, 0x12, 0x29,
 	0x0a, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e,
 	0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x54, 0x6f, 0x70,
@@ -2770,11 +2770,12 @@ var file_mq_proto_rawDesc = []byte{
 	0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x17, 0x2e, 0x6d,
 	0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x50, 0x61, 0x72, 0x74,
 	0x69, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e,
-	0x12, 0x25, 0x0a, 0x0e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6d, 0x65, 0x72, 0x5f, 0x63, 0x6f, 0x75,
-	0x6e, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0d, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6d,
-	0x65, 0x72, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, 0x1b, 0x0a, 0x09, 0x69, 0x73, 0x5f, 0x6c, 0x65,
-	0x61, 0x64, 0x65, 0x72, 0x18, 0x04, 0x20, 0x01, 0x28, 0x08, 0x52, 0x08, 0x69, 0x73, 0x4c, 0x65,
-	0x61, 0x64, 0x65, 0x72, 0x22, 0xd3, 0x01, 0x0a, 0x1d, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68,
+	0x12, 0x27, 0x0a, 0x0f, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x65, 0x72, 0x5f, 0x63, 0x6f,
+	0x75, 0x6e, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0e, 0x70, 0x75, 0x62, 0x6c, 0x69,
+	0x73, 0x68, 0x65, 0x72, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, 0x29, 0x0a, 0x10, 0x73, 0x75, 0x62,
+	0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x72, 0x5f, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x04, 0x20,
+	0x01, 0x28, 0x05, 0x52, 0x0f, 0x73, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x72, 0x43,
+	0x6f, 0x75, 0x6e, 0x74, 0x22, 0xd3, 0x01, 0x0a, 0x1d, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68,
 	0x65, 0x72, 0x54, 0x6f, 0x50, 0x75, 0x62, 0x42, 0x61, 0x6c, 0x61, 0x6e, 0x63, 0x65, 0x72, 0x52,
 	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x4d, 0x0a, 0x04, 0x69, 0x6e, 0x69, 0x74, 0x18, 0x01,
 	0x20, 0x01, 0x28, 0x0b, 0x32, 0x37, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67,