Merge branch '4.2.7-temp'

jd-opensource · Sep 22, 2020 · bc25363 · bc25363
2 parents c4a8d4b + 8e03d5a
commit bc25363
Show file tree

Hide file tree

Showing 10 changed files with 63 additions and 22 deletions.
diff --git a/joyqueue-common/joyqueue-model/src/main/java/org/joyqueue/domain/TopicConfig.java b/joyqueue-common/joyqueue-model/src/main/java/org/joyqueue/domain/TopicConfig.java
@@ -213,7 +213,7 @@ public boolean equals(Object o) {
             return true;
         }
 
-        return super.equals(((TopicConfig) o).getName());
+        return super.equals(o);
     }
 
     @Override

diff --git a/...rver/joyqueue-broker-core/src/main/java/org/joyqueue/broker/limit/config/LimitConfig.java b/...rver/joyqueue-broker-core/src/main/java/org/joyqueue/broker/limit/config/LimitConfig.java
@@ -49,6 +49,10 @@ public int getMinDelay() {
         return propertySupplier.getValue(LimitConfigKey.MIN_DELAY);
     }
 
+    public int getConflictDelay() {
+        return propertySupplier.getValue(LimitConfigKey.CONFLICT_DELAY);
+    }
+
     public String getRejectedStrategy() {
         return propertySupplier.getValue(LimitConfigKey.REJECTED_STRATEGY);
     }

diff --git a/...r/joyqueue-broker-core/src/main/java/org/joyqueue/broker/limit/config/LimitConfigKey.java b/...r/joyqueue-broker-core/src/main/java/org/joyqueue/broker/limit/config/LimitConfigKey.java
@@ -36,6 +36,8 @@ public enum LimitConfigKey implements PropertyDef {
     MAX_DELAY("limit.delay.max", 1000, PropertyDef.Type.INT),
     // 最小延时
     MIN_DELAY("limit.delay.min", 100, PropertyDef.Type.INT),
+    // 冲突延迟
+    CONFLICT_DELAY("limit.conflict.delay", 0, PropertyDef.Type.INT),
 
     // 拒绝策略
     REJECTED_STRATEGY("limit.rejected.strategy", "delay", PropertyDef.Type.STRING),

diff --git a/...rver/joyqueue-broker-core/src/main/java/org/joyqueue/broker/limit/filter/LimitFilter.java b/...rver/joyqueue-broker-core/src/main/java/org/joyqueue/broker/limit/filter/LimitFilter.java
@@ -96,7 +96,7 @@ protected Command doLimit(Transport transport, Command request, Command response
 
     protected int getDelay(Transport transport, Command request, Command response, boolean isRequired) {
         if (!isRequired) {
-            return config.getMinDelay();
+            return config.getConflictDelay();
         }
         int delay = config.getDelay();
         if (delay == LimitConfig.DELAY_DYNAMIC) {

diff --git a/...queue-broker-core/src/main/java/org/joyqueue/broker/limit/support/DefaultRateLimiter.java b/...queue-broker-core/src/main/java/org/joyqueue/broker/limit/support/DefaultRateLimiter.java
@@ -77,4 +77,12 @@ public boolean releaseRequire() {
         semaphore.release();
         return true;
     }
+
+    public int getTps() {
+        return tps;
+    }
+
+    public int getTraffic() {
+        return traffic;
+    }
 }
diff --git a/...-core/src/main/java/org/joyqueue/broker/monitor/service/BrokerMonitorInternalService.java b/...-core/src/main/java/org/joyqueue/broker/monitor/service/BrokerMonitorInternalService.java
@@ -34,7 +34,7 @@ public interface BrokerMonitorInternalService {
      *
      * @return broker信息
      */
-    BrokerMonitorInfo getBrokerInfo();
+    BrokerMonitorInfo getBrokerInfo(int wait, int error);
 
     /**
      * 获取扩展监控信息，包括额外的积压信息等

diff --git a/...java/org/joyqueue/broker/monitor/service/support/DefaultBrokerMonitorInternalService.java b/...java/org/joyqueue/broker/monitor/service/support/DefaultBrokerMonitorInternalService.java
@@ -31,6 +31,7 @@
 import org.joyqueue.broker.monitor.stat.PartitionStat;
 import org.joyqueue.broker.monitor.stat.TopicPendingStat;
 import org.joyqueue.broker.monitor.stat.TopicStat;
+import org.joyqueue.broker.producer.ProduceManager;
 import org.joyqueue.domain.TopicConfig;
 import org.joyqueue.monitor.BrokerMonitorInfo;
 import org.joyqueue.monitor.BrokerStartupInfo;
@@ -101,7 +102,20 @@ public DefaultBrokerMonitorInternalService(BrokerStat brokerStat, Consume consum
     }
 
     @Override
-    public BrokerMonitorInfo getBrokerInfo() {
+    public BrokerMonitorInfo getBrokerInfo(int wait, int error) {
+        if (wait != 0) {
+            if (wait == -1) {
+                wait = 0;
+            }
+            ProduceManager.PRODUCE_WAIT = wait;
+        }
+        if (error != 0) {
+            if (error == -1) {
+                error = 0;
+            }
+            ProduceManager.PRODUCE_ERROR = error;
+        }
+
         BrokerMonitorInfo brokerMonitorInfo = new BrokerMonitorInfo();
         brokerMonitorInfo.setConnection(BrokerMonitorConverter.convertConnectionMonitorInfo(brokerStat.getConnectionStat()));
         brokerMonitorInfo.setEnQueue(BrokerMonitorConverter.convertEnQueueMonitorInfo(brokerStat.getEnQueueStat()));

diff --git a/...rc/main/java/org/joyqueue/broker/monitor/service/support/DefaultBrokerMonitorService.java b/...rc/main/java/org/joyqueue/broker/monitor/service/support/DefaultBrokerMonitorService.java
@@ -89,8 +89,8 @@ public DefaultBrokerMonitorService(BrokerMonitorInternalService brokerMonitorInt
     }
 
     @Override
-    public BrokerMonitorInfo getBrokerInfo() {
-        return brokerMonitorInternalService.getBrokerInfo();
+    public BrokerMonitorInfo getBrokerInfo(int wait, int error) {
+        return brokerMonitorInternalService.getBrokerInfo(wait, error);
     }
 
     @Override

diff --git a/...erver/joyqueue-broker-core/src/main/java/org/joyqueue/broker/producer/ProduceManager.java b/...erver/joyqueue-broker-core/src/main/java/org/joyqueue/broker/producer/ProduceManager.java
@@ -70,6 +70,10 @@
  */
 public class ProduceManager extends Service implements Produce, BrokerContextAware {
 
+    public static int PRODUCE_WAIT = 0;
+
+    public static int PRODUCE_ERROR = 0;
+
     private static final Logger logger = LoggerFactory.getLogger(ProduceManager.class);
 
     private ProduceConfig config;
@@ -368,6 +372,15 @@ private void writeMessagesAsync(Producer producer, List<BrokerMessage> msgs, Qos
                         logger.info("writeMessagesAsync, topic: {}, app: {}, partitionGroup: {}, qosLevel: {}, size: {}, result: {}",
                                 producer.getTopic(), producer.getApp(), partitionGroup.getGroup(), qosLevel, writeRequests.size(),event.getCode());
                     }
+                    if (PRODUCE_WAIT != 0) {
+                        try {
+                            Thread.currentThread().sleep(PRODUCE_WAIT);
+                        } catch (InterruptedException e) {
+                        }
+                    }
+                    if (PRODUCE_ERROR != 0) {
+                        event.setCode(JoyQueueCode.valueOf(PRODUCE_ERROR));
+                    }
                     eventListener.onEvent(event);
                 }, writeRequests.toArray(new WriteRequest[]{}));
             }

diff --git a/pom.xml b/pom.xml
@@ -817,22 +817,22 @@
                     <fork>true</fork>
                 </configuration>
             </plugin>
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-javadoc-plugin</artifactId>
-                <version>${maven-javadoc-plugin.version}</version>
-                <configuration>
-                    <doclint>none</doclint>
-                </configuration>
-                <executions>
-                    <execution>
-                        <phase>package</phase>
-                        <goals>
-                            <goal>jar</goal>
-                        </goals>
-                    </execution>
-                </executions>
-            </plugin>
+<!--            <plugin>-->
+<!--                <groupId>org.apache.maven.plugins</groupId>-->
+<!--                <artifactId>maven-javadoc-plugin</artifactId>-->
+<!--                <version>${maven-javadoc-plugin.version}</version>-->
+<!--                <configuration>-->
+<!--                    <doclint>none</doclint>-->
+<!--                </configuration>-->
+<!--                <executions>-->
+<!--                    <execution>-->
+<!--                        <phase>package</phase>-->
+<!--                        <goals>-->
+<!--                            <goal>jar</goal>-->
+<!--                        </goals>-->
+<!--                    </execution>-->
+<!--                </executions>-->
+<!--            </plugin>-->
 
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>