This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new f0d66d45624 [improve][broker] Added RoundRobinBrokerSelector (#19321)
f0d66d45624 is described below
commit f0d66d456249c42e48444765e11c34988b11c120
Author: Heesung Sohn <[email protected]>
AuthorDate: Fri Feb 3 04:40:17 2023 -0800
[improve][broker] Added RoundRobinBrokerSelector (#19321)
---
.../loadbalance/impl/RoundRobinBrokerSelector.java | 70 ++++++++++++++++++++++
.../ModularLoadManagerStrategyTest.java | 44 ++++++++++++++
2 files changed, 114 insertions(+)
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/RoundRobinBrokerSelector.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/RoundRobinBrokerSelector.java
new file mode 100644
index 00000000000..57804f40ba4
--- /dev/null
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/RoundRobinBrokerSelector.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.loadbalance.impl;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.loadbalance.LoadData;
+import org.apache.pulsar.broker.loadbalance.ModularLoadManagerStrategy;
+import org.apache.pulsar.policies.data.loadbalancer.BundleData;
+
+/**
+ * Simple Round Robin Broker Selection Strategy.
+ */
+public class RoundRobinBrokerSelector implements ModularLoadManagerStrategy {
+
+ final AtomicInteger count = new AtomicInteger();
+ final AtomicReference<List<String>> ref = new AtomicReference<>(List.of());
+
+ @Override
+ public Optional<String> selectBroker(Set<String> candidates, BundleData
bundleToAssign, LoadData loadData,
+ ServiceConfiguration conf) {
+ int candidateSize = candidates.size();
+ if (candidateSize == 0) {
+ return Optional.empty();
+ }
+
+ var cache = ref.get();
+ int cacheSize = cache.size();
+ int index = count.getAndUpdate(i -> i == Integer.MAX_VALUE ? 0 : i +
1) % candidateSize;
+ boolean updateCacheRef = false;
+
+ if (cacheSize <= index || candidateSize != cacheSize) {
+ cache = List.copyOf(candidates);
+ updateCacheRef = true;
+ }
+
+ var selected = cache.get(index);
+ if (!candidates.contains(selected)) {
+ cache = List.copyOf(candidates);
+ updateCacheRef = true;
+ selected = cache.get(index);
+ }
+
+ if (updateCacheRef) {
+ ref.set(cache);
+ }
+
+ return Optional.of(selected);
+ }
+}
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/ModularLoadManagerStrategyTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/ModularLoadManagerStrategyTest.java
index 4f86536c54a..b967deaa726 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/ModularLoadManagerStrategyTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/ModularLoadManagerStrategyTest.java
@@ -21,17 +21,22 @@ package org.apache.pulsar.broker.loadbalance;
import static org.testng.Assert.assertEquals;
import java.lang.reflect.Field;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Collections;
+import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.loadbalance.impl.LeastLongTermMessageRate;
import org.apache.pulsar.broker.loadbalance.impl.LeastResourceUsageWithWeight;
+import org.apache.pulsar.broker.loadbalance.impl.RoundRobinBrokerSelector;
import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
import org.apache.pulsar.policies.data.loadbalancer.ResourceUsage;
import org.apache.pulsar.policies.data.loadbalancer.BrokerData;
@@ -184,6 +189,45 @@ public class ModularLoadManagerStrategyTest {
assertEquals(strategy.selectBroker(candidates, bundleData, loadData,
conf), Optional.of("1"));
}
+ public void testRoundRobinBrokerSelector() throws IllegalAccessException {
+ Set<String> brokers = new LinkedHashSet(Arrays.asList("1", "2", "3"));
+ int n = brokers.size();
+ RoundRobinBrokerSelector strategy = new RoundRobinBrokerSelector();
+
+ assertEquals(strategy.selectBroker(Set.of(), null, null, null),
Optional.empty());
+
+ int i = 0;
+ for (; i < 10; i++) {
+ String id = (i % n) + 1 + "";
+ assertEquals(strategy.selectBroker(brokers, null, null, null),
Optional.of(id));
+ }
+
+ Set<String> brokers2 = new LinkedHashSet(Arrays.asList("2", "3", "1"));
+ for (; i < 20; i++) {
+ String id = (i % n) + 1 + "";
+ assertEquals(strategy.selectBroker(brokers2, null, null, null),
Optional.of(id));
+ }
+
+ Set<String> brokers3 = new LinkedHashSet(Arrays.asList("1", "2", "4"));
+ assertEquals(strategy.selectBroker(brokers3, null, null, null),
Optional.of("4"));
+ assertEquals(strategy.selectBroker(brokers3, null, null, null),
Optional.of("1"));
+ assertEquals(strategy.selectBroker(brokers3, null, null, null),
Optional.of("2"));
+ assertEquals(strategy.selectBroker(brokers3, null, null, null),
Optional.of("4"));
+ assertEquals(strategy.selectBroker(brokers3, null, null, null),
Optional.of("1"));
+ assertEquals(strategy.selectBroker(brokers3, null, null, null),
Optional.of("2"));
+
+ Set<String> brokers4 = new LinkedHashSet(Arrays.asList("2", "4"));
+ assertEquals(strategy.selectBroker(brokers4, null, null, null),
Optional.of("2"));
+ assertEquals(strategy.selectBroker(brokers4, null, null, null),
Optional.of("4"));
+ assertEquals(strategy.selectBroker(brokers4, null, null, null),
Optional.of("2"));
+ assertEquals(strategy.selectBroker(brokers4, null, null, null),
Optional.of("4"));
+
+
+ FieldUtils.writeDeclaredField(strategy, "count", new
AtomicInteger(Integer.MAX_VALUE), true);
+ assertEquals(strategy.selectBroker(brokers, null, null, null),
Optional.of((Integer.MAX_VALUE % n) + 1 + ""));
+ assertEquals(((AtomicInteger) FieldUtils.readDeclaredField(strategy,
"count", true)).get(), 0);
+ }
+
private BrokerData initBrokerData(double usage, double limit) {
LocalBrokerData localBrokerData = new LocalBrokerData();
localBrokerData.setCpu(new ResourceUsage(usage, limit));
