feat(core): refactor AbstractGoal for better scalability (#924)

Signed-off-by: Shichao Nie <niesc@automq.com>

core/src/main/java/kafka/autobalancer/goals/AbstractGoal.java

@@ -17,7 +17,6 @@
 import kafka.autobalancer.common.AutoBalancerConstants;
 import kafka.autobalancer.model.BrokerUpdater;
 import kafka.autobalancer.model.ClusterModelSnapshot;
-import kafka.autobalancer.model.ModelUtils;
 import kafka.autobalancer.model.TopicPartitionReplicaUpdater;
 import org.slf4j.Logger;
 
@@ -117,9 +116,9 @@ protected double calculateAcceptanceScore(BrokerUpdater.Broker srcBrokerBefore,
         }
 
         if (!isSrcBrokerAcceptedBefore && !isSrcBrokerAcceptedAfter) {
-            return score <= POSITIVE_ACTION_SCORE_THRESHOLD ? 0.0 : score;
+            return score < POSITIVE_ACTION_SCORE_THRESHOLD ? 0.0 : score;
         } else if (!isDestBrokerAcceptedBefore && !isDestBrokerAcceptedAfter) {
-            return score <= POSITIVE_ACTION_SCORE_THRESHOLD ? 0.0 : score;
+            return score < POSITIVE_ACTION_SCORE_THRESHOLD ? 0.0 : score;
         }
         return score;
     }
@@ -162,24 +161,16 @@ public double actionAcceptanceScore(Action action, ClusterModelSnapshot cluster)
         BrokerUpdater.Broker destBrokerBefore = cluster.broker(action.getDestBrokerId());
         BrokerUpdater.Broker srcBrokerAfter = srcBrokerBefore.copy();
         BrokerUpdater.Broker destBrokerAfter = destBrokerBefore.copy();
-        TopicPartitionReplicaUpdater.TopicPartitionReplica srcReplica = cluster.replica(action.getSrcBrokerId(), action.getSrcTopicPartition());
 
-        switch (action.getType()) {
-            case MOVE:
-                ModelUtils.moveReplicaLoad(srcBrokerAfter, destBrokerAfter, srcReplica);
-                break;
-            case SWAP:
-                ModelUtils.moveReplicaLoad(srcBrokerAfter, destBrokerAfter, srcReplica);
-                ModelUtils.moveReplicaLoad(destBrokerAfter, srcBrokerAfter,
-                        cluster.replica(action.getDestBrokerId(), action.getDestTopicPartition()));
-                break;
-            default:
-                return 0.0;
+        if (!moveReplica(action, cluster, srcBrokerAfter, destBrokerAfter)) {
+            return 0.0;
         }
 
         return calculateAcceptanceScore(srcBrokerBefore, destBrokerBefore, srcBrokerAfter, destBrokerAfter);
     }
 
+    protected abstract boolean moveReplica(Action action, ClusterModelSnapshot cluster, BrokerUpdater.Broker src, BrokerUpdater.Broker dest);
+
     @Override
     public Set<BrokerUpdater.Broker> getEligibleBrokers(ClusterModelSnapshot cluster) {
         return cluster.brokers().stream().filter(BrokerUpdater.Broker::isActive).collect(Collectors.toSet());

core/src/main/java/kafka/autobalancer/goals/AbstractResourceGoal.java

@@ -18,6 +18,7 @@
 import kafka.autobalancer.common.Resource;
 import kafka.autobalancer.model.BrokerUpdater;
 import kafka.autobalancer.model.ClusterModelSnapshot;
+import kafka.autobalancer.model.ModelUtils;
 import kafka.autobalancer.model.TopicPartitionReplicaUpdater;
 import org.slf4j.Logger;
 
@@ -55,6 +56,24 @@ protected Optional<Action> trySwapPartitionOut(ClusterModelSnapshot cluster,
         return getAcceptableAction(candidateActionScores);
     }
 
+    @Override
+    protected boolean moveReplica(Action action, ClusterModelSnapshot cluster, BrokerUpdater.Broker src, BrokerUpdater.Broker dest) {
+        TopicPartitionReplicaUpdater.TopicPartitionReplica srcReplica = cluster.replica(action.getSrcBrokerId(), action.getSrcTopicPartition());
+        switch (action.getType()) {
+            case MOVE:
+                ModelUtils.moveReplicaLoad(src, dest, srcReplica);
+                break;
+            case SWAP:
+                TopicPartitionReplicaUpdater.TopicPartitionReplica destReplica = cluster.replica(action.getDestBrokerId(), action.getDestTopicPartition());
+                ModelUtils.moveReplicaLoad(src, dest, srcReplica);
+                ModelUtils.moveReplicaLoad(dest, src, destReplica);
+                break;
+            default:
+                return false;
+        }
+        return true;
+    }
+
     /**
      * Try to reduce resource load by move or swap replicas out.
      *