[improve][broker] PIP-192 Improved Auto Unload Logic #19813

heesung-sn · 2023-03-15T01:37:20Z

Master Issue: Master Issue: #16691, #18099

Motivation

Raising a PR to implement Master Issue: #16691, #18099

We want to reduce unload frequencies from flaky traffic.

Modifications

This PR

Introduced a config loadBalancerSheddingConditionHitCountThreshold to further restrict shedding conditions based on the hit count.
Normalized offload traffic
Lowered the default loadBalanceSheddingDelayInSeconds value from 600 to 180, as 10 mins are too long. 3 mins can be long enough to catch the new load after unloads.
Changed the config loadBalancerBundleLoadReportPercentage to loadBalancerMaxNumberOfBundlesInBundleLoadReport to make the topk bundle count absolute instead of relative.
Renamed loadBalancerNamespaceBundleSplitConditionThreshold to loadBalancerNamespaceBundleSplitConditionHitCountThreshold to be consistent with *ConditionHitCountThreshold.
Renamed loadBalancerMaxNumberOfBrokerTransfersPerCycle to loadBalancerMaxNumberOfBrokerSheddingPerCycle.
Added LoadDataStore cleanup logic in BSC monitor.
Added msgThroughputEMA in BrokerLoadData to smooth the broker throughput info.
Updated Topk bundles sorted in a ascending order (instead of descending)
Update some info logs to only show in the debug mode.
Added load data tombstone upon Own, Releasing, Splitting
Added the bundle ownership(isOwned) check upon split and unload.
Added swap unload logic

Verifying this change

Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

*Updated unit tests.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

Documentation

doc
doc-required
doc-not-needed
doc-complete

We will have separate PRs to update the Doc later.

Matching PR in forked repository

PR in forked repository: heesung-sn#39

heesung-sn · 2023-03-16T05:02:17Z

Please pause the review. There will be more updates.

heesung-sn · 2023-03-18T01:51:16Z

Please pause the review. There will be more updates.

Please continue the review.

Demogorgon314 · 2023-03-19T04:02:23Z

Please resolve the conflicts and rebase it into the master branch.

heesung-sn · 2023-03-19T04:58:29Z

Yes, I will rebase after the pr for excluding bundles with policies.

Demogorgon314 · 2023-03-20T09:25:37Z

...src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedder.java

+            Collections.sort(brokersSortedByLoad, (a, b) -> Double.compare(
+                    a.getValue().getWeightedMaxEMA(),
+                    b.getValue().getWeightedMaxEMA()));


nit:

Suggested change

Collections.sort(brokersSortedByLoad, (a, b) -> Double.compare(

a.getValue().getWeightedMaxEMA(),

b.getValue().getWeightedMaxEMA()));

brokersSortedByLoad.sort(Comparator.comparingDouble(a -> a.getValue().getWeightedMaxEMA()));

Demogorgon314 · 2023-03-20T09:26:18Z

...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();


Suggested change

doReturn(new AtomicReference(loadManagerWrapper)).when(pulsar).getLoadManager();

doReturn(new AtomicReference<>(loadManagerWrapper)).when(pulsar).getLoadManager();

Demogorgon314 · 2023-03-20T09:27:39Z

...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) {


This method is never used right now. Please remove it.

Demogorgon314 · 2023-03-20T09:28:00Z

...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();


We can remove it.

Demogorgon314 · 2023-03-20T10:05:22Z

...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--) {


Why reverse the topK bundles order?

We want to transfer lower-loaded ones first to meet the target threshold. If the highest is too much traffic, we can't unload at all.

heesung-sn · 2023-03-22T00:56:28Z

Rebase is done. Please continue the review.

Demogorgon314 · 2023-03-23T06:59:55Z

...in/java/org/apache/pulsar/broker/loadbalance/extensions/reporter/BrokerLoadDataReporter.java

+
+    @Override
+    public void handleEvent(String serviceUnit, ServiceUnitStateData data, Throwable t) {
+        executor.execute(() -> {


Is there any reason to use executor here? I did not see any block operating here.

The intention was to control this handleEvents in a single-threaded manner. I agree that we don't have to use executor here. Removing it.

Demogorgon314 · 2023-03-23T07:01:17Z

.../src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/SplitScheduler.java

@@ -156,6 +159,13 @@ public void start() {
        task = loadManagerExecutor.scheduleAtFixedRate(() -> {
            try {
                execute();
+                if (conf.isLoadBalancerDebugModeEnabled()) {


Why don't we use debugMode here?

Demogorgon314 · 2023-03-23T11:13:05Z

...src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedder.java

        private LoadDataStore<BrokerLoadData> loadDataStore;
-
+        private  List<Map.Entry<String, BrokerLoadData>> brokersSortedByLoad;


Suggested change

private List<Map.Entry<String, BrokerLoadData>> brokersSortedByLoad;

private List<Map.Entry<String, BrokerLoadData>> brokersSortedByLoad;

...in/java/org/apache/pulsar/broker/loadbalance/extensions/reporter/BrokerLoadDataReporter.java

...ava/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java

...src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedder.java

heesung-sn · 2023-03-28T04:57:46Z

@Demogorgon314 can we merge this pr?

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Mar 15, 2023

heesung-sn mentioned this pull request Mar 15, 2023

PIP-192: New Pulsar Broker Load Balancer #16691

Closed

Demogorgon314 self-requested a review March 15, 2023 03:14

Demogorgon314 assigned heesung-sn Mar 15, 2023

heesung-sn force-pushed the pip-192-unload-improve branch 4 times, most recently from 1af3e15 to 31a3c3c Compare March 17, 2023 23:19

Demogorgon314 added type/enhancement The enhancements for the existing features or docs. e.g. reduce memory usage of the delayed messages area/broker labels Mar 20, 2023

Demogorgon314 reviewed Mar 20, 2023

View reviewed changes

heesung-sn force-pushed the pip-192-unload-improve branch from 31a3c3c to 0ca5ae3 Compare March 21, 2023 02:05

heesung-sn added 7 commits March 21, 2023 09:31

[improve][broker] PIP-192 improved TransferShedder logic

1a5df8f

normalized the offload

836aee1

Adjusted the threhsold

9289204

update

8d1a765

temp

8e0ecd0

added swap

4357d31

rebase

f310c4f

heesung-sn force-pushed the pip-192-unload-improve branch from 0ca5ae3 to f310c4f Compare March 21, 2023 23:46

fixed the test failure

7e157e8

heesung-sn force-pushed the pip-192-unload-improve branch 3 times, most recently from d619eb9 to cbac501 Compare March 22, 2023 22:22

fixed test failure

1cc8aab

heesung-sn force-pushed the pip-192-unload-improve branch from cbac501 to 1cc8aab Compare March 22, 2023 22:22

Demogorgon314 reviewed Mar 23, 2023

View reviewed changes

resolved comments

9337913

gaoran10 reviewed Mar 25, 2023

View reviewed changes

resolved comments

459631b

heesung-sn force-pushed the pip-192-unload-improve branch from a9b4171 to 459631b Compare March 25, 2023 20:42

Demogorgon314 approved these changes Mar 26, 2023

View reviewed changes

Technoboy- added the ready-to-test label Mar 27, 2023

Technoboy- added this to the 3.0.0 milestone Mar 27, 2023

Technoboy- closed this Mar 27, 2023

Technoboy- reopened this Mar 27, 2023

gaoran10 approved these changes Mar 27, 2023

View reviewed changes

Demogorgon314 merged commit ee5ac86 into apache:master Mar 29, 2023

heesung-sn deleted the pip-192-unload-improve branch April 2, 2024 17:44

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[improve][broker] PIP-192 Improved Auto Unload Logic #19813

[improve][broker] PIP-192 Improved Auto Unload Logic #19813

heesung-sn commented Mar 15, 2023 •

edited

Loading

heesung-sn commented Mar 16, 2023

heesung-sn commented Mar 18, 2023

Demogorgon314 commented Mar 19, 2023

heesung-sn commented Mar 19, 2023

Demogorgon314 Mar 20, 2023

heesung-sn Mar 22, 2023

Demogorgon314 Mar 20, 2023

heesung-sn Mar 22, 2023

Demogorgon314 Mar 20, 2023

heesung-sn Mar 22, 2023

Demogorgon314 Mar 20, 2023

heesung-sn Mar 22, 2023

Demogorgon314 Mar 20, 2023

heesung-sn Mar 21, 2023

heesung-sn commented Mar 22, 2023

Demogorgon314 Mar 23, 2023

heesung-sn Mar 23, 2023

Demogorgon314 Mar 23, 2023

heesung-sn Mar 23, 2023

Demogorgon314 Mar 23, 2023

heesung-sn Mar 23, 2023

heesung-sn commented Mar 28, 2023

	doReturn(new AtomicReference(loadManagerWrapper)).when(pulsar).getLoadManager();
	doReturn(new AtomicReference<>(loadManagerWrapper)).when(pulsar).getLoadManager();

		private LoadDataStore<BrokerLoadData> loadDataStore;

		private List<Map.Entry<String, BrokerLoadData>> brokersSortedByLoad;

[improve][broker] PIP-192 Improved Auto Unload Logic #19813

[improve][broker] PIP-192 Improved Auto Unload Logic #19813

Conversation

heesung-sn commented Mar 15, 2023 • edited Loading

Motivation

Modifications

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Matching PR in forked repository

heesung-sn commented Mar 16, 2023

heesung-sn commented Mar 18, 2023

Demogorgon314 commented Mar 19, 2023

heesung-sn commented Mar 19, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

heesung-sn commented Mar 22, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

heesung-sn commented Mar 28, 2023

heesung-sn commented Mar 15, 2023 •

edited

Loading