[GitHub] [pulsar] Demogorgon314 commented on a diff in pull request #19813: [improve][broker] PIP-192 Improved Auto Unload Logic

["Demogorgon314 (via GitHub)" <gi...@apache.org>]

Demogorgon314 commented on code in PR #19813:
URL: https://github.com/apache/pulsar/pull/19813#discussion_r1141831538


##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedderTest.java:
##########
@@ -134,25 +143,29 @@ public void init() throws MetadataStoreException {
                     (upperEndpoint.equals(NamespaceBundles.FULL_UPPER_BOUND)) ? BoundType.CLOSED : BoundType.OPEN);
             return new NamespaceBundle(NamespaceName.get(namespace), hashRange, factory);
         }).when(factory).getBundle(anyString(), anyString());
-        doReturn(true).when(antiAffinityGroupPolicyHelper).canUnload(any(), any(), any(), any());
+        doReturn(new AtomicReference(loadManagerWrapper)).when(pulsar).getLoadManager();

Review Comment:
   ```suggestion
           doReturn(new AtomicReference<>(loadManagerWrapper)).when(pulsar).getLoadManager();
   ```



##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedderTest.java:
##########
@@ -567,8 +616,7 @@ private void setIsolationPolicies(SimpleResourceAllocationPolicies policies,
         }).when(policies).shouldFailoverToSecondaries(eq(namespaceName), anyInt());
     }
 
-    private PulsarService getMockPulsar() {
-        var pulsar = mock(PulsarService.class);
+    private PulsarService getMockPulsar(PulsarService pulsar) {

Review Comment:
   This method is never used right now. Please remove it.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedder.java:
##########
@@ -195,47 +209,65 @@ Optional<UnloadDecision.Reason> update(final LoadDataStore<BrokerLoadData> loadS
 
                 double load = localBrokerData.getWeightedMaxEMA();
 
-                minBrokers.offer(broker);
-                if (minBrokers.size() > maxTransfers) {
-                    minBrokers.poll();
-                }
-                maxBrokers.offer(broker);
-                if (maxBrokers.size() > maxTransfers) {
-                    maxBrokers.poll();
-                }
                 sum += load;
                 sqSum += load * load;
                 totalBrokers++;
             }
 
-
             if (totalBrokers == 0) {
                 if (decisionReason == null) {
                     decisionReason = NoBrokers;
                 }
                 update(0.0, 0.0, 0);
+                if (debug) {
+                    log.info("There is no broker load data.");
+                }
+                return Optional.of(decisionReason);
+            }
+
+            if (!missingLoadDataBrokers.isEmpty()) {
+                decisionReason = NoLoadData;
+                update(0.0, 0.0, 0);
+                if (debug) {
+                    log.info("There is missing load data from brokers:{}", missingLoadDataBrokers);
+                }
                 return Optional.of(decisionReason);
             }
 
             update(sum, sqSum, totalBrokers);
             return Optional.empty();
         }
 
-        boolean hasTransferableBrokers() {
-            return !(maxBrokers.isEmpty() || minBrokers.isEmpty()
-                    || maxBrokers.peekLast().equals(minBrokers().peekLast()));
-        }
-
         void setLoadDataStore(LoadDataStore<BrokerLoadData> loadDataStore) {
             this.loadDataStore = loadDataStore;
+            brokersSortedByLoad.addAll(loadDataStore.entrySet());
+            Collections.sort(brokersSortedByLoad, (a, b) -> Double.compare(
+                    a.getValue().getWeightedMaxEMA(),
+                    b.getValue().getWeightedMaxEMA()));

Review Comment:
   nit:
   ```suggestion
               brokersSortedByLoad.sort(Comparator.comparingDouble(a -> a.getValue().getWeightedMaxEMA()));
   ```
   



##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedderTest.java:
##########
@@ -589,8 +637,9 @@ private PulsarService getMockPulsar() {
 
     @Test
     public void testBundlesWithAntiAffinityGroup() throws IllegalAccessException, MetadataStoreException {
-        var pulsar = getMockPulsar();
-        TransferShedder transferShedder = new TransferShedder(pulsar, antiAffinityGroupPolicyHelper);
+        //var pulsar = getMockPulsar();

Review Comment:
   We can remove it.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/models/TopKBundles.java:
##########
@@ -66,7 +66,7 @@ public void update(Map<String, NamespaceBundleStats> bundleStats, int topk) {
         topk = Math.min(topk, arr.size());
         partitionSort(arr, topk);
 
-        for (int i = 0; i < topk; i++) {
+        for (int i = topk - 1; i >= 0; i--) {

Review Comment:
   Why reverse the topK bundles order?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org