Repository: skiptests/astraea
Pull request #270 (2022-03-02 14:06:46)

改善JMX server參數的使用方式
#262
現在的邏輯是這樣的，如果配置了jmx.servers，那麼會先配置這些broker的jmx port，如果還有broker沒有配置到jmx port，那麼統一用jmx.defaultPort進行配置。而如果有broker沒有被配置到，而同時也沒有配置jmx.defaultPort那麼就會報錯。

chia7712: @wycccccc 感謝這麼快準備好patch，不過這邊我們可能要提早想一下給未來其他partitioner都能用的方式。因為我們不太可能說每個partitioner要用的參數通通都擺到`Performance`上面，這樣會很亂又難以維護。

我一個很快的想法是，`Performance`要能吃一個叫做`--configs`的參數，該目的類似於`props.file`，用來自由設置任何參數，但比起使用property file又簡單一點，如此，我們在測試不同的partitioner的時候，就能透過`--configs`傳遞客製化參數給各個不同的partitioner
chia7712: > 我一個很快的想法是，Performance要能吃一個叫做--configs的參數，該目的類似於props.file，用來自由設置任何參數，但比起使用property file又簡單一點，如此，我們在測試不同的partitioner的時候，就能透過--configs傳遞客製化參數給各個不同的partitioner

see #271

Patch:
diff --git a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java
index 34281d9c..861d309f 100644
--- a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java
@@ -9,6 +9,8 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
+import java.util.TreeMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicReference;
@@ -20,6 +22,7 @@ import org.astraea.metrics.collector.Receiver;
 import org.astraea.metrics.java.HasJvmMemory;
 import org.astraea.metrics.kafka.KafkaMetrics;
 import org.astraea.partitioner.ClusterInfo;
+import org.astraea.partitioner.NodeInfo;
 
 /**
  * this clas is responsible for obtaining jmx metrics from BeanCollector and calculating the
@@ -27,9 +30,8 @@ import org.astraea.partitioner.ClusterInfo;
  */
 public class NodeLoadClient {
 
-  private final List<Receiver> receiverList;
-  private final Map<String, Integer> currentJmxAddresses;
   private final Lock lock = new ReentrantLock();
+  private final Optional<Integer> jmxPortDefault;
   private static final ReceiverFactory RECEIVER_FACTORY = new ReceiverFactory();
   private long lastTime = -1;
   // visible for testing
@@ -38,14 +40,16 @@ public class NodeLoadClient {
   private int referenceBrokerID;
   private int totalInput;
   private int totalOutput;
+  private Map<Integer, Integer> jmxOfBrokerID;
+  private Map<String, Integer> jmxAddress;
 
   private static final String regex =
       "((25[0-5]|2[0-4]\\d|1\\d{2}|[1-9]?\\d)\\.){3}" + "(25[0-5]|2[0-4]\\d|1\\d{2}|[1-9]?\\d)$";
 
-  public NodeLoadClient(Map<String, Integer> jmxAddresses) {
-    this.receiverList = RECEIVER_FACTORY.receiversList(jmxAddresses);
-    receiverList.forEach(receiver -> Utils.waitFor(() -> receiver.current().size() > 0));
-    currentJmxAddresses = jmxAddresses;
+  public NodeLoadClient(Map<Integer, Integer> jmxAddresses, int jmxPortDefault) {
+    if (jmxPortDefault != -1) this.jmxPortDefault = Optional.of(jmxPortDefault);
+    else this.jmxPortDefault = Optional.empty();
+    jmxOfBrokerID = jmxAddresses;
   }
 
   /**
@@ -69,6 +73,29 @@ public class NodeLoadClient {
     return currentLoadNode;
   }
 
+  Map<String, Integer> jmxAddress(ClusterInfo cluster) {
+    var iterator = jmxOfBrokerID.keySet().iterator();
+    var currentJmxAddresses = new TreeMap<String, Integer>();
+    var idHost = cluster.nodes().stream().collect(Collectors.toMap(NodeInfo::id, NodeInfo::host));
+    while (iterator.hasNext()) {
+      var id = iterator.next();
+      if (idHost.containsKey(id)) {
+        currentJmxAddresses.put(ipAddress(idHost.get(id)), jmxOfBrokerID.get(id));
+        idHost.remove(id);
+      }
+    }
+
+    if (idHost.size() > 0) {
+      jmxPortDefault.ifPresentOrElse(
+          port -> idHost.forEach((key, value) -> currentJmxAddresses.put(ipAddress(value), port)),
+          () -> {
+            var notFindJMX = new ArrayList<>(idHost.keySet());
+            throw new RuntimeException(notFindJMX + "jmx servers not found.");
+          });
+    }
+    return currentJmxAddresses;
+  }
+
   /** @param cluster the cluster from the partitioner */
   private void nodesOverLoad(ClusterInfo cluster) {
     if (lastTime == -1) {
@@ -78,6 +105,9 @@ public class NodeLoadClient {
               .map(node -> new Broker(node.id(), node.host(), node.port()))
               .collect(Collectors.toList());
       referenceBrokerID = brokers.stream().findFirst().get().brokerID;
+      jmxAddress = jmxAddress(cluster);
+      var receiverList = RECEIVER_FACTORY.receiversList(jmxAddress);
+      receiverList.forEach(receiver -> Utils.waitFor(() -> receiver.current().size() > 0));
       var nodeIDReceiver = new HashMap<Integer, List<Receiver>>();
       receiverList.forEach(
           receiver -> {
@@ -137,7 +167,7 @@ public class NodeLoadClient {
               Long.parseLong(
                   broker
                       .metrics
-                      .get(KafkaMetrics.BrokerTopic.BytesInPerSec.name())
+                      .get(KafkaMetrics.BrokerTopic.BytesOutPerSec.name())
                       .current()
                       .get(0)
                       .beanObject()
@@ -236,7 +266,7 @@ public class NodeLoadClient {
   }
 
   public void close() {
-    RECEIVER_FACTORY.close(currentJmxAddresses);
+    RECEIVER_FACTORY.close(jmxAddress);
   }
 
   private List<Integer> brokerIDOfReceiver(String host) {
diff --git a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/PartitionerUtils.java b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/PartitionerUtils.java
index 9036314c..b557cd74 100644
--- a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/PartitionerUtils.java
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/PartitionerUtils.java
@@ -1,7 +1,10 @@
 package org.astraea.partitioner.nodeLoadMetric;
 
+import java.io.FileInputStream;
+import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Properties;
 
 /** do poisson for node's load situation */
 public class PartitionerUtils {
@@ -45,20 +48,13 @@ public class PartitionerUtils {
     return (int) Math.round(avgLoadCount);
   }
 
-  /**
-   * A cheap way to deterministically convert a number to a positive value. When the input is
-   * positive, the original value is returned. When the input number is negative, the returned
-   * positive value is the original value bit AND against 0x7fffffff which is not its absolute
-   * value.
-   *
-   * <p>Note: changing this method in the future will possibly cause partition selection not to be
-   * compatible with the existing messages already placed on a partition since it is used in
-   * producer's {@link org.apache.kafka.clients.producer.internals.DefaultPartitioner}
-   *
-   * @param number a given number
-   * @return a positive number.
-   */
-  public static int toPositive(int number) {
-    return number & 0x7fffffff;
+  public static Properties partitionerConfig(Map<String, ?> configs) {
+    var properties = new Properties();
+    try {
+      properties.load(new FileInputStream((String) configs.get("partitioner.config")));
+    } catch (IOException e) {
+      e.printStackTrace();
+    }
+    return properties;
   }
 }
diff --git a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/ReceiverFactory.java b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/ReceiverFactory.java
index bdc8bc98..fb6f7ac6 100644
--- a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/ReceiverFactory.java
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/ReceiverFactory.java
@@ -69,9 +69,9 @@ public class ReceiverFactory {
     return receiversList;
   }
 
-  public void close(Map<String, Integer> jmxAddresses) {
+  public void close(Map<String, Integer> jmxAddress) {
     synchronized (lock) {
-      jmxAddresses.forEach(
+      jmxAddress.forEach(
           (host, port) -> {
             if (count.containsKey(nodeKey(host, port))) {
               if (count.get(nodeKey(host, port)) == 1) {
diff --git a/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java b/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java
index b1632239..5d94d7a1 100644
--- a/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java
+++ b/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java
@@ -2,22 +2,23 @@ package org.astraea.partitioner.smoothPartitioner;
 
 import static org.astraea.Utils.overSecond;
 import static org.astraea.partitioner.nodeLoadMetric.PartitionerUtils.allPoisson;
+import static org.astraea.partitioner.nodeLoadMetric.PartitionerUtils.partitionerConfig;
 import static org.astraea.partitioner.nodeLoadMetric.PartitionerUtils.weightPoisson;
 
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Iterator;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Random;
+import java.util.TreeMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
 import org.apache.kafka.clients.producer.Partitioner;
 import org.apache.kafka.common.Cluster;
 import org.astraea.partitioner.ClusterInfo;
+import org.astraea.partitioner.Configuration;
 import org.astraea.partitioner.nodeLoadMetric.NodeLoadClient;
 
 /**
@@ -26,14 +27,14 @@ import org.astraea.partitioner.nodeLoadMetric.NodeLoadClient;
  * Finally, the result of poisson is used as the weight to perform smooth weighted RoundRobin.
  */
 public class SmoothWeightPartitioner implements Partitioner {
+  private static final String JMX_PORT = "jmx.port";
 
-  /**
-   * Record the current weight of each node according to Poisson calculation and the weight after
-   * partitioner calculation.
-   */
+  //  Record the current weight of each node according to Poisson calculation and the weight after
+  // partitioner calculation.
   private final ConcurrentMap<Integer, SmoothWeightServer> brokersWeight =
       new ConcurrentHashMap<>();
 
+  private final Map<Integer, Integer> jmxPorts = new TreeMap<>();
   private NodeLoadClient nodeLoadClient;
   private long lastTime = -1;
   private final Random rand = new Random();
@@ -49,7 +50,7 @@ public class SmoothWeightPartitioner implements Partitioner {
 
     SmoothWeightServer maxWeightServer = null;
 
-    Iterator<Integer> iterator = brokersWeight.keySet().iterator();
+    var iterator = brokersWeight.keySet().iterator();
     var currentWrightSum = weightSum;
     while (iterator.hasNext()) {
       var smoothWeightServer = brokersWeight.get(iterator.next());
@@ -80,19 +81,26 @@ public class SmoothWeightPartitioner implements Partitioner {
 
   @Override
   public void configure(Map<String, ?> configs) {
-    var jmxAddresses =
-        Objects.requireNonNull(
-            configs.get("jmx_servers").toString(), "You must configure jmx_servers correctly");
-    var list = Arrays.asList((jmxAddresses).split(","));
-    HashMap<String, Integer> mapAddress = new HashMap<>();
-    list.forEach(
-        str ->
-            mapAddress.put(
-                Arrays.asList(str.split(":")).get(0),
-                Integer.parseInt(Arrays.asList(str.split(":")).get(1))));
-    Objects.requireNonNull(mapAddress, "You must configure jmx_servers correctly.");
-
-    nodeLoadClient = new NodeLoadClient(mapAddress);
+    var properties = partitionerConfig(configs);
+    var config =
+        Configuration.of(
+            properties.entrySet().stream()
+                .collect(
+                    Collectors.toMap(e -> e.getKey().toString(), e -> e.getValue().toString())));
+    var jmxPort = config.integer(JMX_PORT);
+
+    jmxPort = config.integer(JMX_PORT);
+
+    // seeks for custom jmx ports.
+    config.entrySet().stream()
+        .filter(e -> e.getKey().startsWith("broker."))
+        .filter(e -> e.getKey().endsWith(JMX_PORT))
+        .forEach(
+            e ->
+                jmxPorts.put(
+                    Integer.parseInt(e.getKey().split("\\.")[1]), Integer.parseInt(e.getValue())));
+
+    nodeLoadClient = new NodeLoadClient(jmxPorts, jmxPort.orElse(-1));
   }
 
   /** Change the weight of the node according to the current Poisson. */
diff --git a/app/src/main/java/org/astraea/performance/Performance.java b/app/src/main/java/org/astraea/performance/Performance.java
index 8b521746..58f1d924 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -285,7 +285,7 @@ public class Performance {
     @Parameter(
         names = {"--jmx.servers"},
         description =
-            "String: server to get jmx metrics <jmx_server>@<broker_id>[,<jmx_server>@<broker_id>]*",
+            "String: server to get jmx metrics <broker_id>.<jmx_serverPort>[,<broker_id>.<jmx_serverPort>]*",
         validateWith = NonEmptyStringField.class)
     String jmxServers = "";
 
diff --git a/config/partitionerConfig.properties b/config/partitionerConfig.properties
new file mode 100644
index 00000000..e31beda9
--- /dev/null
+++ b/config/partitionerConfig.properties
@@ -0,0 +1,20 @@
+#The properties are used to configure SmoothWeightPartitioner
+#
+#If broker.id.jmx.port is configured, then the jmx port of these brokers
+# will be configured first. If there are brokers that are not configured
+# to jmx port, then jmx.port will be used for configuration. And if any
+# broker is not configured, and jmx.port is not configured at the same
+# time, then an error will be reported.
+#
+###########################Smooth Partitioner Basics############################
+
+#If no jmx port is specified for the broker,jmx.port will be used.
+jmx.port=Default;
+
+# list of brokers used for jmx port about the rest of the cluster
+# format: broker.{brokerID}.jmx.port=jmx.port
+#broker.0.jmx.port=9999
+#broker.1.jmx.port=8888
+#......
+bootstrap.servers=localhost:9092
+broker.id.jmx.port=Default;


Test patch:
diff --git a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClientTest.java b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClientTest.java
index 3c754173..4b85e90a 100644
--- a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClientTest.java
+++ b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClientTest.java
@@ -2,7 +2,6 @@ package org.astraea.partitioner.nodeLoadMetric;
 
 import static org.mockito.Mockito.when;
 
-import java.io.IOException;
 import java.lang.reflect.Field;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -24,10 +23,10 @@ public class NodeLoadClientTest extends RequireBrokerCluster {
   private NodeLoadClient nodeLoadClient;
 
   @BeforeAll
-  void setUp() throws IOException {
-    var map = new HashMap<String, Integer>();
-    map.put(jmxServiceURL().getHost(), jmxServiceURL().getPort());
-    nodeLoadClient = new NodeLoadClient(map);
+  void setUp() {
+    var map = new HashMap<Integer, Integer>();
+    map.put(0, jmxServiceURL().getPort());
+    nodeLoadClient = new NodeLoadClient(map, jmxServiceURL().getPort());
   }
 
   @AfterAll
@@ -89,7 +88,6 @@ public class NodeLoadClientTest extends RequireBrokerCluster {
           nodes.add(NodeInfo.of(count.get(), hostPort[0], Integer.parseInt(hostPort[1])));
           count.getAndIncrement();
         });
-
     var cluster = Mockito.mock(ClusterInfo.class);
     when(cluster.nodes()).thenReturn(nodes);
     var load = nodeLoadClient.loadSituation(cluster);
diff --git a/app/src/test/java/org/astraea/partitioner/smoothPartitioner/PartitionerTest.java b/app/src/test/java/org/astraea/partitioner/smoothPartitioner/PartitionerTest.java
index af24a097..c655bfa4 100644
--- a/app/src/test/java/org/astraea/partitioner/smoothPartitioner/PartitionerTest.java
+++ b/app/src/test/java/org/astraea/partitioner/smoothPartitioner/PartitionerTest.java
@@ -6,6 +6,9 @@ import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
 import java.lang.reflect.Field;
 import java.time.Duration;
 import java.util.HashMap;
@@ -49,12 +52,24 @@ public class PartitionerTest extends RequireBrokerCluster {
     props.put(ProducerConfig.CLIENT_ID_CONFIG, "id1");
     props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, SmoothWeightPartitioner.class.getName());
     props.put("producerID", 1);
-    props.put("jmx_servers", jmxServiceURL().getHost() + ":" + jmxServiceURL().getPort());
+    var file = new File(PartitionerTest.class.getResource("").getPath() + "PartitionerConfigTest");
+    try {
+      var fileWriter = new FileWriter(file);
+      fileWriter.write("jmx.port=" + jmxServiceURL().getPort());
+      fileWriter.flush();
+      fileWriter.close();
+    } catch (IOException e) {
+      e.printStackTrace();
+    }
+    props.put(
+        "partitioner.config",
+        PartitionerTest.class.getResource("").getPath() + "PartitionerConfigTest");
     return props;
   }
 
   @Test
   void testSetBrokerHashMap() {
+    initProConfig();
     var nodeLoadClient = Mockito.mock(NodeLoadClient.class);
     when(nodeLoadClient.thoughPutComparison(anyInt())).thenReturn(1.0);
     var poissonMap = new HashMap<Integer, Double>();
@@ -218,6 +233,57 @@ public class PartitionerTest extends RequireBrokerCluster {
     Assertions.assertEquals(smoothWeightServer.originalWeight(), -9);
   }
 
+  @Test
+  void testJmxConfig() {
+    var props = initProConfig();
+    var file = new File(PartitionerTest.class.getResource("").getPath() + "PartitionerConfigTest");
+    try {
+      var fileWriter = new FileWriter(file);
+      fileWriter.write(
+          "broker.0.jmx.port="
+              + jmxServiceURL().getPort()
+              + "\n"
+              + "broker.1.jmx.port="
+              + jmxServiceURL().getPort()
+              + "\n"
+              + "broker.2.jmx.port="
+              + jmxServiceURL().getPort()
+              + "\n");
+      fileWriter.flush();
+      fileWriter.close();
+    } catch (IOException e) {
+      e.printStackTrace();
+    }
+    var topicName = "addressN";
+    admin.creator().topic(topicName).numberOfPartitions(10).create();
+    var key = "tainan";
+    var timestamp = System.currentTimeMillis() + 10;
+    var header = Header.of("a", "b".getBytes());
+
+    try (var producer =
+        Producer.builder()
+            .keySerializer(Serializer.STRING)
+            .configs(
+                props.entrySet().stream()
+                    .collect(Collectors.toMap(e -> e.getKey().toString(), Map.Entry::getValue)))
+            .build()) {
+      var metadata =
+          producer
+              .sender()
+              .topic(topicName)
+              .key(key)
+              .timestamp(timestamp)
+              .headers(List.of(header))
+              .run()
+              .toCompletableFuture()
+              .get();
+      assertEquals(topicName, metadata.topic());
+      assertEquals(timestamp, metadata.timestamp());
+    } catch (ExecutionException | InterruptedException e) {
+      e.printStackTrace();
+    }
+  }
+
   private Executor producerExecutor(
       Producer<String, byte[]> producer, String topic, String key, Header header, long timeStamp) {
     return new Executor() {