[Colocation] Fix Colocation balance endless loop bug

fe/fe-core/src/main/java/org/apache/doris/clone/ColocateTableBalancer.java

@@ -546,6 +546,11 @@ private boolean balance(GroupId groupId, List<Long> allAvailBackendIds, Colocate
             // sort backends with replica num
             List<Map.Entry<Long, Long>> backendWithReplicaNum =
                     getSortedBackendReplicaNumPairs(allAvailBackendIds, flatBackendsPerBucketSeq);
+            // if there is only one available backend, end the outer loop
+            if (backendWithReplicaNum.size() == 1) {
+                LOG.info("there is only one available backend, end the outer loop in colocate group {}", groupId);
+                break;
+            }
 
             int i = 0;
             int j = backendWithReplicaNum.size() - 1;
@@ -600,11 +605,18 @@ private boolean balance(GroupId groupId, List<Long> allAvailBackendIds, Colocate
                 }
 
                 if (!isThisRoundChanged) {
-                    // select another load backend and try again
                     LOG.info("unable to replace backend {} with backend {} in colocate group {}",
                             srcBeId, destBeId, groupId);
-                    j--;
-                    continue;
+                    if (--j == i) {
+                        // if all backends are checked but this round is not changed,
+                        // we should end the outer loop to avoid endless loops
+                        LOG.info("all backends are checked but this round is not changed, " +
+                                         "end outer loop in colocate group {}", groupId);
+                        break OUT;
+                    } else {
+                        // select another load backend and try again
+                        continue;
+                    }
                 }
 
                 break;

fe/fe-core/src/test/java/org/apache/doris/clone/ColocateTableBalancerTest.java

@@ -151,4 +151,37 @@ public void testBalance() {
         Assert.assertFalse(changed);
         Assert.assertTrue(balancedBackendsPerBucketSeq.isEmpty());
     }
+
+    @Test
+    public void testFixBalanceEndlessLoop() {
+        GroupId groupId = new GroupId(10000, 10001);
+        List<Column> distributionCols = Lists.newArrayList();
+        distributionCols.add(new Column("k1", PrimitiveType.INT));
+        ColocateGroupSchema groupSchema = new ColocateGroupSchema(groupId, distributionCols, 5, (short) 1);
+        Map<GroupId, ColocateGroupSchema> group2Schema = Maps.newHashMap();
+        group2Schema.put(groupId, groupSchema);
+
+        // 1. only one available backend
+        // [[7], [7], [7], [7], [7]]
+        ColocateTableIndex colocateTableIndex = createColocateIndex(groupId, Lists.newArrayList(7L, 7L, 7L, 7L, 7L));
+        Deencapsulation.setField(colocateTableIndex, "group2Schema", group2Schema);
+
+        List<List<Long>> balancedBackendsPerBucketSeq = Lists.newArrayList();
+        List<Long> allAvailBackendIds = Lists.newArrayList(7L);
+        boolean changed = Deencapsulation.invoke(balancer, "balance", groupId, allAvailBackendIds,
+                                                 colocateTableIndex, infoService, balancedBackendsPerBucketSeq);
+        Assert.assertFalse(changed);
+
+        // 2. all backends are checked but this round is not changed
+        // [[7], [7], [7], [7], [7]]
+        // and add new backends 8, 9 that are on the same host with 7
+        colocateTableIndex = createColocateIndex(groupId, Lists.newArrayList(7L, 7L, 7L, 7L, 7L));
+        Deencapsulation.setField(colocateTableIndex, "group2Schema", group2Schema);
+
+        balancedBackendsPerBucketSeq = Lists.newArrayList();
+        allAvailBackendIds = Lists.newArrayList(7L, 8L, 9L);
+        changed = Deencapsulation.invoke(balancer, "balance", groupId, allAvailBackendIds,
+                                         colocateTableIndex, infoService, balancedBackendsPerBucketSeq);
+        Assert.assertFalse(changed);
+    }
 }