handle single broker case
This commit is contained in:
@@ -126,14 +126,30 @@ func EnsureAssignmentsToActiveBrokers(activeBrokers cmap.ConcurrentMap[string, *
|
|||||||
i++
|
i++
|
||||||
hasChanges = true
|
hasChanges = true
|
||||||
}
|
}
|
||||||
|
|
||||||
|
hasEmptyFollowers := false
|
||||||
j := 0
|
j := 0
|
||||||
for ; j<len(assignment.FollowerBrokers); j++ {
|
for ; j<len(assignment.FollowerBrokers); j++ {
|
||||||
if assignment.FollowerBrokers[j] == "" {
|
if assignment.FollowerBrokers[j] == "" {
|
||||||
assignment.FollowerBrokers[j] = pickedBrokers[i]
|
|
||||||
i++
|
|
||||||
hasChanges = true
|
hasChanges = true
|
||||||
|
if i < len(pickedBrokers) {
|
||||||
|
assignment.FollowerBrokers[j] = pickedBrokers[i]
|
||||||
|
i++
|
||||||
|
} else {
|
||||||
|
hasEmptyFollowers = true
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
if hasEmptyFollowers {
|
||||||
|
var followerBrokers []string
|
||||||
|
for _, follower := range assignment.FollowerBrokers {
|
||||||
|
if follower != "" {
|
||||||
|
followerBrokers = append(followerBrokers, follower)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
assignment.FollowerBrokers = followerBrokers
|
||||||
|
}
|
||||||
|
|
||||||
if i < len(pickedBrokers) {
|
if i < len(pickedBrokers) {
|
||||||
assignment.FollowerBrokers = append(assignment.FollowerBrokers, pickedBrokers[i:]...)
|
assignment.FollowerBrokers = append(assignment.FollowerBrokers, pickedBrokers[i:]...)
|
||||||
hasChanges = true
|
hasChanges = true
|
||||||
|
|||||||
@@ -83,6 +83,8 @@ func TestEnsureAssignmentsToActiveBrokersX(t *testing.T) {
|
|||||||
lowActiveBrokers := cmap.New[*BrokerStats]()
|
lowActiveBrokers := cmap.New[*BrokerStats]()
|
||||||
lowActiveBrokers.SetIfAbsent("localhost:1", &BrokerStats{})
|
lowActiveBrokers.SetIfAbsent("localhost:1", &BrokerStats{})
|
||||||
lowActiveBrokers.SetIfAbsent("localhost:2", &BrokerStats{})
|
lowActiveBrokers.SetIfAbsent("localhost:2", &BrokerStats{})
|
||||||
|
singleActiveBroker := cmap.New[*BrokerStats]()
|
||||||
|
singleActiveBroker.SetIfAbsent("localhost:1", &BrokerStats{})
|
||||||
tests := []struct {
|
tests := []struct {
|
||||||
name string
|
name string
|
||||||
args args
|
args args
|
||||||
@@ -207,6 +209,23 @@ func TestEnsureAssignmentsToActiveBrokersX(t *testing.T) {
|
|||||||
},
|
},
|
||||||
hasChanges: false,
|
hasChanges: false,
|
||||||
},
|
},
|
||||||
|
{
|
||||||
|
name: "test single active broker",
|
||||||
|
args: args{
|
||||||
|
activeBrokers: singleActiveBroker,
|
||||||
|
followerCount: 3,
|
||||||
|
assignments: []*mq_pb.BrokerPartitionAssignment{
|
||||||
|
{
|
||||||
|
LeaderBroker: "localhost:1",
|
||||||
|
Partition: &mq_pb.Partition{},
|
||||||
|
FollowerBrokers: []string{
|
||||||
|
"localhost:2",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
hasChanges: true,
|
||||||
|
},
|
||||||
}
|
}
|
||||||
for _, tt := range tests {
|
for _, tt := range tests {
|
||||||
t.Run(tt.name, func(t *testing.T) {
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
|||||||
Reference in New Issue
Block a user