Repository: skiptests/astraea
Pull request #516 (2022-07-28 12:22:01)

Fix testConsumerRebalanceListener
這是我在 #513 沒寫好的地方，不小心留了一隻 thread 在操作 consumer


Patch:
diff --git a/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java b/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java
index 38dab18a..ff4587b4 100644
--- a/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java
+++ b/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java
@@ -158,54 +158,6 @@ public final class KafkaMetrics {
     }
   }
 
-  public enum Purgatory {
-    AlterAcls("AlterAcls"),
-    DeleteRecords("DeleteRecords"),
-    ElectLeader("ElectLeader"),
-    Fetch("Fetch"),
-    Heartbeat("Heartbeat"),
-    Produce("Produce"),
-    Rebalance("Rebalance");
-
-    private final String metricName;
-
-    Purgatory(String name) {
-      this.metricName = name;
-    }
-
-    public String metricName() {
-      return metricName;
-    }
-
-    public Collection<HasBeanObject> fetch(MBeanClient mBeanClient) {
-      return mBeanClient
-          .queryBeans(
-              BeanQuery.builder()
-                  .domainName("kafka.server")
-                  .property("type", "DelayedOperationPurgatory")
-                  .property("delayedOperation", metricName)
-                  .property("name", "PurgatorySize")
-                  .build())
-          .stream()
-          .map(HasValue::of)
-          .collect(Collectors.toUnmodifiableList());
-    }
-
-    public int size(MBeanClient mBeanClient) {
-      return (int)
-          mBeanClient
-              .queryBean(
-                  BeanQuery.builder()
-                      .domainName("kafka.server")
-                      .property("type", "DelayedOperationPurgatory")
-                      .property("delayedOperation", this.name())
-                      .property("name", "PurgatorySize")
-                      .build())
-              .attributes()
-              .get("Value");
-    }
-  }
-
   public enum Request {
     Produce,
     FetchConsumer,
diff --git a/app/src/main/java/org/astraea/app/metrics/broker/ServerMetrics.java b/app/src/main/java/org/astraea/app/metrics/broker/ServerMetrics.java
new file mode 100644
index 00000000..8af7a1ab
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/metrics/broker/ServerMetrics.java
@@ -0,0 +1,89 @@
+/*
+ * 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.astraea.app.metrics.broker;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.stream.Collectors;
+import org.astraea.app.metrics.jmx.BeanObject;
+import org.astraea.app.metrics.jmx.BeanQuery;
+import org.astraea.app.metrics.jmx.MBeanClient;
+
+public final class ServerMetrics {
+
+  public enum DelayedOperationPurgatory {
+    AlterAcls("AlterAcls"),
+    DeleteRecords("DeleteRecords"),
+    ElectLeader("ElectLeader"),
+    Fetch("Fetch"),
+    Heartbeat("Heartbeat"),
+    Produce("Produce"),
+    Rebalance("Rebalance");
+
+    private final String metricName;
+
+    DelayedOperationPurgatory(String name) {
+      this.metricName = name;
+    }
+
+    public String metricName() {
+      return metricName;
+    }
+
+    public Collection<Size> fetch(MBeanClient mBeanClient) {
+      return mBeanClient
+          .queryBeans(
+              BeanQuery.builder()
+                  .domainName("kafka.server")
+                  .property("type", "DelayedOperationPurgatory")
+                  .property("delayedOperation", metricName)
+                  .property("name", "PurgatorySize")
+                  .build())
+          .stream()
+          .map(Size::new)
+          .collect(Collectors.toUnmodifiableList());
+    }
+
+    public static DelayedOperationPurgatory of(String metricName) {
+      return Arrays.stream(DelayedOperationPurgatory.values())
+          .filter(metric -> metric.metricName().equalsIgnoreCase(metricName))
+          .findFirst()
+          .orElseThrow(() -> new IllegalArgumentException("No such metric: " + metricName));
+    }
+
+    public static class Size implements HasValue {
+      private final BeanObject beanObject;
+
+      public Size(BeanObject beanObject) {
+        this.beanObject = beanObject;
+      }
+
+      public String metricsName() {
+        return beanObject().properties().get("delayedOperation");
+      }
+
+      public DelayedOperationPurgatory type() {
+        return DelayedOperationPurgatory.of(metricsName());
+      }
+
+      @Override
+      public BeanObject beanObject() {
+        return beanObject;
+      }
+    }
+  }
+}


Test patch:
diff --git a/app/src/test/java/org/astraea/app/consumer/RebalanceListenerTest.java b/app/src/test/java/org/astraea/app/consumer/RebalanceListenerTest.java
index 9765124a..074b8567 100644
--- a/app/src/test/java/org/astraea/app/consumer/RebalanceListenerTest.java
+++ b/app/src/test/java/org/astraea/app/consumer/RebalanceListenerTest.java
@@ -18,7 +18,6 @@ package org.astraea.app.consumer;
 
 import java.time.Duration;
 import java.util.Set;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.astraea.app.common.Utils;
 import org.astraea.app.service.RequireBrokerCluster;
@@ -35,7 +34,6 @@ public class RebalanceListenerTest extends RequireBrokerCluster {
             .bootstrapServers(bootstrapServers())
             .consumerRebalanceListener(ignore -> getAssignment.incrementAndGet())
             .build()) {
-      CompletableFuture.runAsync(() -> consumer.poll(Duration.ofSeconds(10)));
       Utils.waitFor(
           () -> {
             consumer.poll(Duration.ofSeconds(1));
diff --git a/app/src/test/java/org/astraea/app/metrics/KafkaMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/KafkaMetricsTest.java
index 75199eac..5632e7d2 100644
--- a/app/src/test/java/org/astraea/app/metrics/KafkaMetricsTest.java
+++ b/app/src/test/java/org/astraea/app/metrics/KafkaMetricsTest.java
@@ -108,13 +108,6 @@ class KafkaMetricsTest extends RequireBrokerCluster {
     assertDoesNotThrow(result::rateUnit);
   }
 
-  @ParameterizedTest()
-  @EnumSource(value = KafkaMetrics.Purgatory.class)
-  void testPurgatorySize(KafkaMetrics.Purgatory request) {
-    // act assert type casting correct and field exists
-    assertDoesNotThrow(() -> request.size(mBeanClient));
-  }
-
   @ParameterizedTest()
   @EnumSource(value = KafkaMetrics.Request.class)
   void testRequestTotalTimeMs(KafkaMetrics.Request request) {
diff --git a/app/src/test/java/org/astraea/app/metrics/broker/ServerMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/broker/ServerMetricsTest.java
new file mode 100644
index 00000000..b5312648
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/metrics/broker/ServerMetricsTest.java
@@ -0,0 +1,31 @@
+/*
+ * 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.astraea.app.metrics.broker;
+
+import org.astraea.app.metrics.jmx.MBeanClient;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+public class ServerMetricsTest {
+
+  @ParameterizedTest()
+  @EnumSource(value = ServerMetrics.DelayedOperationPurgatory.class)
+  void testPurgatorySize(ServerMetrics.DelayedOperationPurgatory request) {
+    request.fetch(MBeanClient.local()).forEach(s -> Assertions.assertTrue(s.value() >= 0));
+  }
+}