Integrate BeanColloctor into Partitioner #138

wycccccc · 2021-12-03T18:34:29Z

繼續 #113 的完善
將BeanColloctor整合進了Partitioner中。
簡化了使用者配置Jmx addresses的行為，不再需要將NodeID與Jmx addresses一同配置。
Code更加乾淨了。

wycccccc · 2021-12-03T18:37:09Z

還有一些小問題需要請教 #137
一些程式中的print()會在bug解決後刪除
NodeLoadClient還會再補充一些測試。

…nAgain

wycccccc · 2021-12-05T08:27:49Z

#137

另一個角度是說這個問題先緩一緩先把重構的部份處理好再回來

這個問題應該緩不了，因為我們的測試環境就是一個host上有三個node。算是上述提到的半個情景，我有嘗試解決，現在可以支持同一個host上有多個node的情況。.
程式的基本框架已經完成，之後的其他功能會在合併後通過其他pr添加。

chia7712 · 2021-12-05T14:21:28Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/BeanCollectorFactory.java

+import org.astraea.metrics.HasBeanObject;
+
+/** create a single BeanCollector to avoid waste of resources */
+public class BeanCollectorFactory {


我不太確定是否還需要用這種方式，之前用這個是因為要考慮"cluster"的分組問題，不過現在BeanCollector並沒有cluster的概念，BeanCollector就是維護一個jmx server and metrics的關係。換言之，我們可以簡單用static memebr來處理"singleton"就好

如果目前做法對效能沒有太大影響，我想可以暫時保留這種處理方式，一方面是如果改用static memebr可能還是需要一個類似邏輯的程式判斷BeanCollector何時應該關閉。
另一方面，我未來是打算將BeanCollector能夠根據實際的情況進行客制化創建（例如可以存儲不同時間量的數據，或是根據面對jmxserver的數量決定要開設幾條線程處理它們）。這部分我還沒想好具體要怎麼做，但是一個Factory可以保證我生產出想要的BeanCollector。

所以現在partitioner和BeanCollector的關係為何？是什麼情況會共用什麼時候又不會共用？

用同樣的config創建多個producer時候這些producer是共用的。用不同的config這樣就是不同的了。

不同的config

這個可能也要開個議題追蹤一下，因為太容易有不同的config 但是面對同一個叢集的狀態

這樣聽起來不太秒，我直接改成維護一個 static BeanCollector的形式。已完成。

wycccccc · 2021-12-10T15:40:50Z

緊接着 #112 進行討論
我試圖創建一個Map<producer,partitioner>來將producer與partitoner進行匹配方便控制partitioner中的state。
我盡力想通過producer調出partitioner，或是通過partitioner匹配到producer。但似乎都做不到。
現在我們的使用者將需要多配置一個參數producerID在producerConfig中。
每一個producerID都需要唯一，幫助進行partitioner與producer的匹配。

使用者目前實現dependency的方法為。

var dependencyClient = new DependencyClient();
dependencyClient.initializeDependency(props);
dependencyClient.beginDependency(props);
producer.send();
dependencyClient.finishDependency(props);

如果這種做法可行的話我會再補齊注釋。

chia7712 · 2021-12-11T04:37:35Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/BeanCollectorFactory.java

+  public void close() {
+    synchronized (lock) {
+      if (count == 1) {
+        beanCollector.close();


這樣設計會有問題，例如當同一個JVM關掉最後一個partitioner，然後又開一個新的，這樣就會用到已經被關閉的beanCollector

已修復該問題，現在關掉後再創建不會再用到已關閉的beanCollector

chia7712 · 2021-12-11T04:38:16Z

app/src/main/java/org/astraea/metrics/BeanCollector.java

@@ -25,7 +25,7 @@ public static Builder builder() {
    return new Builder();
  }

-  static class Builder {
+  public static class Builder {


請使用builder()取代開放此建構

這個地方我不太知道改用哪種寫法。如果刪掉public在我build它的時候會出現Cannot be accessed from outside package的錯誤。

BeanCollector.build() 這樣就會建立Builder了

去掉pubilc後它會回報上述錯誤，我是不知道這個錯誤該怎麼避免。

喔喔那是我看錯了你改得沒錯在別的package要存取的話

好的，沒問題就行

wycccccc · 2021-12-11T08:11:42Z

如果producerID的做法沒有問題的話，會有以下一個問題
現在的producerID為了程序正常運轉是必須需要設置的，但是實際上只有當用dependency的時候才需要這個參數。
為了讓使用者更自由確定是否設置producerID，還需要加一套producerID的防呆機制。
我先將這個問題記錄在這裡，如果確定producerID是可行的未來方向。我再開始處理這個問題。

chia7712 · 2021-12-11T15:55:44Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java

+  }
+
+  public void close() {
+    FACTORY.close();


在多個partitioner共用的狀況下，close並不會釋放任何東西。但實際上的確有東西要釋放掉，那就是beanCollector.register()執行完後的物件，該物件連結了在beanCollector裡面的資源，那就是呼叫註冊的執行緒必須去釋放的東西

因此這邊的邏輯應該要改成搜集執行緒過程建立的Unregister，然後在這裡把它釋放掉就好

請見#150，現在不需要去追蹤FACTORY，而是把自己拿到的Receivers斷掉就好

有修改成斷掉維護Receivers的thread pool，這樣有算把資源釋放掉嘛。

chia7712 · 2021-12-11T18:57:39Z

如果producerID的做法沒有問題的話，會有以下一個問題
現在的producerID為了程序正常運轉是必須需要設置的，但是實際上只有當用dependency的時候才需要這個參數。
為了讓使用者更自由確定是否設置producerID，還需要加一套producerID的防呆機制。
我先將這個問題記錄在這裡，如果確定producerID是可行的未來方向。我再開始處理這個問題。

麻煩開另一個議題來追蹤，這應該跟這隻PR的關係不大

chia7712

@wycccccc 因為現在演算法過程用的資料結構太難閱讀，我希望看如何調整BeanCollector讓你不需要用那麼複雜的方式來保存資料，也希望能簡化你這隻在多執行緒和關閉等議題上變得比較輕鬆

麻煩先看一下我回的意見，以及#150，謝謝

chia7712 · 2021-12-11T18:55:32Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java

-      this.nodeMetadataCollection.add(
-          new NodeMetadata(entry.getKey(), createNodeMetrics(entry.getKey(), entry.getValue())));
+      this.beanCollector = FACTORY.beanCollector();
+      beanCollector


我看了一下你的用法，重新調整了BeanCollector (see #150)，在新版的用法下，這段可以改成

var receivers = List.of( beanCollector .register() .host(entry.getKey()) .port(Integer.parseInt(entry.getValue())) .metricsGetter(KafkaMetrics.BrokerTopic.BytesOutPerSec::fetch) .build(), beanCollector .register() .host(entry.getKey()) .port(Integer.parseInt(entry.getValue())) .metricsGetter(KafkaMetrics.BrokerTopic.BytesInPerSec::fetch) .build())

已完成修改。

chia7712 · 2021-12-11T18:59:12Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java

-    for (NodeMetadata nodeMetadata : nodeMetadataCollection) {
-      NodeMetrics nodeMetrics = nodeMetadata.getNodeMetrics();
-      Utils.close(nodeMetrics.getKafkaMetricClient());
+    for (Map.Entry<Integer, Map.Entry<String, Integer>> hostPort : addresses.entrySet()) {


麻煩看看能否以Receiver -> List<HasBeanObject>的資料結構來重新整理這段，現在這隻PR用的這些資料結構太難閱讀（因為層層圈套），如果無法單純用Receiver -> List<HasBeanObject>這樣的方式來處理，那我們再討論一下該如何調整APIs

我寫的時候也很艱辛，像是在玩一個邏輯積木。我會嘗試用新方法去實現，應該不會再變成這樣。今天需要做論文簡報，明天睡前應該可以改好。

可以將複雜的資料結構寫成一個新的class，給各個變數新的名字，一來好閱讀二來也好維護。另外也建議多寫一些comment幫助理解現在的要用的資料結構

有學到，不過使用Receiver後資料結構得以簡化不少，目前應該保持Map還可以接受。

chia7712 · 2021-12-11T18:59:46Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java

+  }
+
+  public void close() {
+    FACTORY.close();


請見#150，現在不需要去追蹤FACTORY，而是把自己拿到的Receivers斷掉就好

chia7712

@wycccccc 感謝更新，一樣幾個重要的設計討論請看一下

chia7712 · 2021-12-15T07:35:24Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/BeanCollectorFactory.java

+        for (Map.Entry<String, Integer> entry : jmxAddresses.entrySet()) {
+          var mbeanClient = MBeanClient.jndi(entry.getKey(), entry.getValue());
+          BiFunction<String, Integer, MBeanClient> clientCreator = (host, port) -> mbeanClient;
+          var beanCollector =


現在BeanCollector已經不佔用資源，而是作為一個“共同“儲存媒介給各個Receiver，換言之，跟現在的BeanCollectorFactory角色有點重疊，另外BeanCollector已經是thread-safe，這邊在用lock有點多餘

chia7712 · 2021-12-15T07:36:10Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/BeanCollectorFactory.java

+                  .interval(Duration.ofSeconds(1))
+                  .numberOfObjectsPerNode(10)
+                  .clientCreator(clientCreator)
+                  .build();


這邊每次都建立新的BeanCollector會使得不同的receiver不能共享bean objects，除非你是故意要達到這樣的效果，否則應該要由同一個BeanCollector來取得Receiver

chia7712 · 2021-12-15T07:37:24Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/BeanCollectorFactory.java

+            ThreadPool.builder()
+                .runnables(
+                    receiversList.stream()
+                        .map(receiver -> ((Runnable) receiver::current))


一樣需要執行緒嗎？如果要仰賴額外執行緒來取得bean，又會掉入之前處理多個partitioner共用下要如何優雅關閉的問題

wycccccc · 2021-12-16T17:57:08Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/receiverFactory.java

+                  .metricsGetter(KafkaMetrics.BrokerTopic.BytesOutPerSec::fetch)
+                  .build());
+          count.put(entry, 1);
+        } else {


出現了一個問題，我按照這樣的形式創建我的receiver們，可是獲得的receiver獲取的metrics都會變成第一個receiver設置的metrics。

我不確定是我創建的方式有問題，還是這是一個隱蔽的bug。

你是跑哪一個test出現這個問題？

我很快看了一下，這邊用的資料結構是Map<String, Receiver>，這樣在有多個brokers的時候就會撞到key重複。

學長指的是nodeLoadClient中的Map<String, Receiver>嘛。這個應該不會是問題的所在，我跑的是partitionerTest，因為我在創建這個資料結構前，在我一開始拿到receiverList的時候，我就直接對其中的兩個receiver，拿取他們的hasBeanObject進行對比。
發現兩個receiver記錄的metrics是一樣的。這代表問題應該在receiverFactory創建receiver中就已經出現。所以我在懷疑是我創建的方式有問題還是其他原因。
上圖我截圖的代碼是我用來抓取receiverList中每個receiver的hasBeanObject的metricsname。

好的，我了解問題了，請看#154

已測試完畢，現在它能正常運行了。

wycccccc · 2021-12-18T12:25:57Z

上述的建議都已修改，現在代碼應該精煉了不少，再麻煩學長review。

chia7712

@wycccccc 現在程式碼裡面很多可以用Map.forEach來改寫，麻煩看一下

chia7712 · 2021-12-18T18:08:35Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/BrokersWeight.java

-      }
-    }
+  public synchronized void brokerHashMap(Map<Integer, Double> poissonMap) {
+    poissonMap


poissonMap.forEach( (key, value) -> { if (!brokerHashMap.containsKey(key)) { brokerHashMap.put(key, new int[] {(int) ((1 - value) * 20), 0}); } else { brokerHashMap.put(key, new int[] {(int) ((1 - value) * 20), brokerHashMap.get(key)[1]}); } });

已完成提到的所有修改，麻煩再看一下給予意見。

chia7712 · 2021-12-18T18:10:26Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/ReceiverFactory.java

+   */
+  public List<Receiver> receiversList(Map<String, Integer> jmxAddresses) {
+    synchronized (lock) {
+      jmxAddresses


麻煩用jmxAddresses.forEach((host, port) -> {來處理

chia7712 · 2021-12-18T18:21:52Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/BrokersWeight.java

-  public BrokersWeight(LoadPoisson loadPoisson) {
-    this.loadPoisson = loadPoisson;
-  }
+  private static Map<Integer, int[]> brokerHashMap = new HashMap<>();


這種用法容易產生誤會，因為使用者必須“建立BrokersWeight”這個物件，但這個物件內部的資料卻是static...

這邊我這樣做的原因，是因為我想讓不同的partitioner可以共享一個權重，這樣新的partitioner一進來就會按目前集群的狀況來塞資料。或許我可以把它搬進partitioer中。將BrokerWeight變為一個內部類，然後在partitioner中維護一個static BrokersWeight，這樣感覺就比較自然。

我又思考了一下，所有partitioner都用一個權重好像也沒有明顯的優勢。反而會在維護權重上產生不少問題。所以我直接移除了static。

chia7712

@wycccccc thanks for updating code. a couple of comments below. Also, I'm waiting the response about producer id

chia7712 · 2021-12-20T16:39:30Z

app/src/test/java/org/astraea/partitioner/nodeLoadMetric/receiverFactoryTest.java

+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class receiverFactoryTest extends RequireBrokerCluster {


ReceiverFactoryTest

chia7712 · 2021-12-20T16:40:26Z

app/src/test/java/org/astraea/partitioner/nodeLoadMetric/receiverFactoryTest.java

+
+  @Test
+  void testSingleton() {
+    initProConfig();


The returned object is not used so I guess this method is not necessary

chia7712 · 2021-12-20T16:42:21Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/LoadPoisson.java

+
+  private int avgLoadCount(Map<Integer, Integer> overLoadCount) {
+    var avgLoadCount =
+        overLoadCount.values().stream().mapToDouble(Integer::doubleValue).average().getAsDouble();


verLoadCount.values().stream().mapToDouble(Integer::doubleValue).average().orElse(0);

chia7712 · 2021-12-20T16:42:49Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/LoadPoisson.java

    return ans;
  }

+  // visible for test
  public long factorial(long number) {


If it is used for testing only, it should be package-private rather than public modifier

了解，相關的其他地方也已一並修改。

chia7712 · 2021-12-20T16:44:30Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/LoadPoisson.java

-      int x = nodeLoadClient.getBinOneCount(entry.getValue());
-      poissonMap.put(entry.getKey(), doPoisson(lambda, x));
-    }
+  public synchronized HashMap<Integer, Double> allPoisson(Map<Integer, Integer> overLoadCount) {


Why it requires synchronization? You should make overLoadCount thread-safe instead.

sor，我沒有修改到，現在的架構確實已經不需要了，已修正。

chia7712 · 2021-12-20T16:45:58Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java

-      e.printStackTrace();
-    }
-    return null;
+  private static final ReceiverFactory FACTORY = new ReceiverFactory();


RECEIVER_FACTORY

chia7712 · 2021-12-20T16:48:05Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java

+    for (Map.Entry<Integer, Map.Entry<String, Integer>> hostPort : addresses.entrySet()) {
+      List<Receiver> matchingNode;
+      if (!hostPort.getValue().getKey().matches(regex)) {
+        var localhost =


why we need to check localhost?

應該不是loaclhost，而是host。
這裡是因為我發現測試中啟動的kafka抓到的host是host name，而不是它的ip地址。
那時候一直在解決這個問題，所以心中一直想着loaclhost就取成了這個名。
已修改名稱。

chia7712 · 2021-12-20T16:48:31Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java

+                                                    n.stream().findAny().get().port(),
+                                                    hostPort.getValue().getValue()))
+                                        .anyMatch(n -> n.equals(true)))
+                                || Objects.equals(nodeIDReceiver.size(), 0))


we should evaluate Objects.equals(nodeIDReceiver.size(), 0) first since it is cheaper.

有學到！

chia7712 · 2021-12-20T16:49:17Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java

+    var addresses =
+        cluster.nodes().stream()
+            .collect(
+                Collectors.toMap(node -> node.id(), node -> Map.entry(node.host(), node.port())));


chia7712 · 2021-12-20T16:49:38Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java

+            receiverList.stream()
+                .filter(
+                    receiver ->
+                        ((nodeIDReceiver.values().stream()


please use helper to handle this duplicate code

雖然它們之間長的很像但是實際上是在比較不同的東西，所以用不了helper。
不過現在我有改變一下思路，進而拿掉了那些duplicate code。

chia7712 · 2021-12-21T15:00:19Z

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java

+
+    for (Map.Entry<Integer, Map.Entry<String, Integer>> hostPort : addresses.entrySet()) {
+      List<Receiver> matchingNode;
+      var host = "-1.-1.-1.-1";


這看起來是要把hostname轉成IP address，麻煩用個helper method來處理

已修改。

chia7712

@wycccccc 感謝更新，針對message order的設計還有一些疑問，請看一下

chia7712 · 2021-12-24T15:55:56Z

app/src/main/java/org/astraea/partitioner/smoothPartitioner/DependencyClient.java

+
+  private SmoothWeightPartitioner partitionerOfProducer(KafkaProducer producer)
+      throws NoSuchFieldException, IllegalAccessException {
+    var field = producer.getClass().getDeclaredField("partitioner");


麻煩把這個動作放到Utils裡面

已移動到Utils中

chia7712 · 2021-12-24T15:56:21Z

app/src/main/java/org/astraea/partitioner/smoothPartitioner/DependencyClient.java

+    smoothWeightPartitioner.finishDependency();
+  }
+
+  private SmoothWeightPartitioner partitionerOfProducer(KafkaProducer producer)


KafkaProducer<?, ?>

chia7712 · 2021-12-24T15:57:37Z

app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java

+      Objects.requireNonNull(mapAddress, "You must configure jmx_servers correctly.");
+
+      nodeLoadClient = new NodeLoadClient(mapAddress);
+      dependencyManager = new DependencyManager();


這可以直接初始化就好
private final DependencyManager dependencyManager = new DependencyManager();

chia7712 · 2021-12-24T15:58:34Z

app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java

+        } catch (UnknownHostException e) {
+          e.printStackTrace();
+        }
+        assert overLoadCount != null;


為何不是拋出錯誤而是要用assert?

修改成拋出錯誤，類似問題也都已修改。

chia7712 · 2021-12-24T15:59:30Z

app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java

+
+        assert overLoadCount != null;
+        brokerHashMap(loadPoisson.allPoisson(overLoadCount));
+        AtomicReference<Map.Entry<Integer, int[]>> maxWeightServer = new AtomicReference<>();


這邊用Map就好

這邊我只需要找到一個最大權重的server就好，感覺資料結構上Map沒有Map.Entry好處理。

chia7712 · 2021-12-24T16:00:07Z

app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java

+
+  private NodeLoadClient nodeLoadClient;
+  private DependencyManager dependencyManager;
+  private int targetPartition;


這個特別的參數需要註解

chia7712 · 2021-12-24T16:03:32Z

app/src/main/java/org/astraea/partitioner/smoothPartitioner/DependencyClient.java

+public class DependencyClient {
+  private final SmoothWeightPartitioner smoothWeightPartitioner;
+
+  public DependencyClient(KafkaProducer kafkaProducer)


可否讓這個建構式交由SmoothWeightPartitioner處理？也就是使用者只要認SmoothWeightPartitioner就好，而不是用特別去記這個看起來不容易察覺關聯性的名字？例如

SmoothWeightPartitioner.startGroup(producer); try { ... } finally { SmoothWeightPartitioner.endGroup(producer); }

chia7712 · 2021-12-24T16:04:14Z

app/src/main/java/org/astraea/partitioner/smoothPartitioner/DependencyClient.java

+    this.smoothWeightPartitioner = partitionerOfProducer(kafkaProducer);
+  }
+
+  public synchronized void initializeDependency() {


承上，為何需要有READY?

備用，現在Dependency大概只是能夠按要求動的程度，留個坑以後好補些邏輯做優化。

除非必要，否則APi越簡潔越好

好，已經移除。

wycccccc · 2021-12-25T09:37:18Z

app/src/main/java/org/astraea/Utils.java

+    field.setAccessible(true);
+    return (SmoothWeightPartitioner) field.get(producer);
+  }
+


現在user使用Utils中的partitionerOfProducermethod調用到producer對應的partitioner。
以下為範例代碼

KafkaProducer producer = new KafkaProducer(props); SmoothWeightPartitioner smoothWeightPartitioner = Utils.partitionerOfProducer(producer); try{ dependencyClient.beginDependency(); producer.send(); } finally{ dependencyClient.finishDependency(); }

SmoothWeightPartitioner smoothWeightPartitioner = Utils.partitionerOfProducer(producer);

理想上使用者只需要認SmoothWeightPartitioner這隻class就好，一來方便二來好記憶，因此會建議改成:

var orderControl = SmoothWeightPartitioner.orderControl(producer); try { orderControl.begin(); producer.send(); producer.send(); producer.send(); } finally { orderControl.finish(); }

好，已修改成上述案例樣式。

chia7712

@wycccccc 感謝更新，還有一些設計討論，請看一下

chia7712 · 2021-12-27T14:31:43Z

app/src/main/java/org/astraea/Utils.java

+   * @param fieldName reflected field name.
+   * @return Required field.
+   */
+  public static Field reflectionField(Object object, String fieldName) {


public static Object requireField(Object object, String fieldName) { try { var field = object.getClass().getDeclaredField(fieldName); field.setAccessible(true); return field.get(object); } catch (NoSuchFieldException | IllegalAccessException e) { throw new IllegalArgumentException(e); } }

chia7712 · 2021-12-27T14:36:50Z

app/src/test/java/org/astraea/partitioner/nodeLoadMetric/DependencyPartitionTest.java

+            .collect(Collectors.toMap(e -> e.getKey().toString(), Map.Entry::getValue));
+    try (var producer = Producer.builder().configs(props).build()) {
+      SmoothWeightPartitioner smoothWeightPartitioner =
+          SmoothWeightPartitioner.orderControl(producer.kafkaProducer());


這邊也是前後兩行的呼叫可以得知，拿到smoothWeightPartitioner後馬上要呼叫beginDependency，那麼我們就可以把這兩行合併在一起

另外也要考慮是否要建議一個新的介面來避免直接使用SmoothWeightPartitioner

有建立了一個interface DependencyClient來避免使用者能直接調用到partitioner。然後是合併的問題，這樣如果使用者在一個程式里對一個producer反覆開關dependency不就會創建很多沒有用的object了。還是說為了讓使用者方便，這種程度的資源浪費是可以接受的？

一個producer反覆開關dependency不就會創建很多沒有用的object了。還是說為了讓使用者方便，這種程度的資源浪費是可以接受的？

別太小看現代JVM了而且相比這支程式要處理的邏輯這個簡單的物件開關不會是大事情

了解，已經修改完畢，現在它們被整合到了一起。

chia7712 · 2021-12-27T14:37:19Z

app/src/main/java/org/astraea/producer/Producer.java

@@ -11,6 +12,8 @@

  void close();

+  KafkaProducer kafkaProducer();


KafkaProducer<Key, Value> kafkaProducer();

chia7712

@wycccccc a coupe of major questions below.

chia7712 · 2022-01-03T17:29:23Z

app/src/main/java/org/astraea/partitioner/smoothPartitioner/DependencyClient.java

@@ -0,0 +1,5 @@
+package org.astraea.partitioner.smoothPartitioner;
+
+public interface DependencyClient {


麻煩加上註解

chia7712 · 2022-01-03T17:30:05Z

app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java

+    return (DependencyClient) Utils.requireField(producer, "partitioner");
+  }
+
+  public synchronized void finishDependency() {


@Override

chia7712 · 2022-01-03T17:31:39Z

app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java

+  }
+
+  public static synchronized DependencyClient beginDependency(KafkaProducer<?, ?> producer) {
+    dependencyManager.beginDependency();


如果同時有多個執行緒同時在呼叫beginDependency和finishDependency，這樣會執行緒安全嗎

目前的話，這樣會因為錯誤的狀態轉換而報錯，我將在下個PR一併優化該問題。

wycccccc · 2022-01-05T10:26:14Z

我提議能夠先將此PR meger，因為我已基於這支PR完成了partitoner v1.0.0,添加了不少新東西。再在此push顯得過於臃腫，不太適合繼續在此幹活了。

chia7712 · 2022-01-05T11:02:03Z

我提議能夠先將此PR meger，因為我已基於這支PR完成了partitoner v1.0.0,添加了不少新東西。再在此push顯得過於臃腫，不太適合繼續在此幹活了。

麻煩把整個討論串中提到的題目都先開好issue

wycccccc added 2 commits December 3, 2021 23:22

ask questions

90b588a

ask questions

b2b4052

wycccccc mentioned this pull request Dec 3, 2021

Make the code clear. #113

Closed

wycccccc added 2 commits December 4, 2021 20:43

Merge branch 'main' of https://github.com/skiptests/astraea into clea…

5576ea5

…nAgain

mission completed

3f02758

chia7712 reviewed Dec 5, 2021

View reviewed changes

wycccccc added 2 commits December 6, 2021 17:39

Little perfect

dc3ed59

add test

fd50b9e

wycccccc mentioned this pull request Dec 7, 2021

Ensure that the partitioner finds the node where the partition exists #107

Closed

wycccccc added 2 commits December 10, 2021 23:29

dependency send

8e2b28d

spotless

cdd6197

static beancolloctor

610bbc9

chia7712 reviewed Dec 11, 2021

View reviewed changes

fix BeanCollectorFactory

0ab1a15

chia7712 reviewed Dec 11, 2021

View reviewed changes

wycccccc added 3 commits December 15, 2021 00:36

use Receiver replace beanCollector

c21366d

Redundant code

105e9b9

private static

27dd6e6

wycccccc mentioned this pull request Dec 14, 2021

The processing of dependency data partitioner #112

Closed

chia7712 reviewed Dec 15, 2021

View reviewed changes

a question

a8a0ddf

wycccccc commented Dec 16, 2021

View reviewed changes

chia7712 mentioned this pull request Dec 17, 2021

The receiver should keep independent objects #154

Merged

lambda

b0b786a

chia7712 reviewed Dec 18, 2021

View reviewed changes

wycccccc added 2 commits December 20, 2021 21:53

integrate SmoothWeightPartitioner and brokersWeight

a0ebab7

remove static of brokerHashMap

f10cced

chia7712 reviewed Dec 20, 2021

View reviewed changes

wycccccc added 2 commits December 21, 2021 22:13

Some improvements

a265516

remove duplicate code

832f6ee

chia7712 reviewed Dec 21, 2021

View reviewed changes

wycccccc added 2 commits December 22, 2021 18:16

reflection of dependency

0ae477f

Edit comment

93372ed

chia7712 reviewed Dec 24, 2021

View reviewed changes

some improvements

255ef3a

wycccccc commented Dec 25, 2021

View reviewed changes

wycccccc added 3 commits December 25, 2021 17:39

Annotation modification

c5a81a1

remove no need state

6710422

dependency Control

1404866

chia7712 reviewed Dec 27, 2021

View reviewed changes

chia7712 mentioned this pull request Dec 27, 2021

Enable transactional send #136

Merged

wycccccc added 2 commits December 28, 2021 14:49

Dependency client interface

1a504ee

Integration beginDependency

7ccf9ac

chia7712 reviewed Jan 3, 2022

View reviewed changes

reset

e0ff448

wycccccc force-pushed the cleanAgain branch from ba1c44d to e0ff448 Compare January 5, 2022 10:14

wycccccc added 2 commits January 5, 2022 18:16

spotlessApply

2d98146

TODO

a873c68

chia7712 merged commit d6ab715 into opensource4you:main Jan 5, 2022

wycccccc mentioned this pull request Jan 5, 2022

關於相依性資料partitioner的處理方式 #79

Closed

		@@ -11,6 +12,8 @@

		void close();

		KafkaProducer kafkaProducer();

		@@ -0,0 +1,5 @@
		package org.astraea.partitioner.smoothPartitioner;

		public interface DependencyClient {

Integrate BeanColloctor into Partitioner #138

Integrate BeanColloctor into Partitioner #138

Conversation

wycccccc commented Dec 3, 2021

wycccccc commented Dec 3, 2021

wycccccc commented Dec 5, 2021 • edited Loading

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

wycccccc Dec 10, 2021 • edited Loading

Choose a reason for hiding this comment

wycccccc commented Dec 10, 2021 • edited Loading

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

wycccccc commented Dec 11, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chia7712 commented Dec 11, 2021

chia7712 left a 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

chia7712 left a 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

Choose a reason for hiding this comment

wycccccc commented Dec 18, 2021

chia7712 left a 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

wycccccc Dec 20, 2021 • edited Loading

Choose a reason for hiding this comment

chia7712 left a 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

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

wycccccc Dec 21, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wycccccc commented Dec 5, 2021 •

edited

Loading

wycccccc Dec 10, 2021 •

edited

Loading

wycccccc commented Dec 10, 2021 •

edited

Loading

wycccccc Dec 20, 2021 •

edited

Loading

wycccccc Dec 21, 2021 •

edited

Loading