Merge pull request #5534 from yu199195/metrics-aysnc

Use Async thread save metrics.

sharding-metrics/sharding-metrics-configuration/src/main/java/org/apache/shardingsphere/metrics/configuration/config/MetricsConfiguration.java

@@ -37,6 +37,10 @@ public final class MetricsConfiguration {
     
     private Integer port;
     
+    private Boolean async;
+    
+    private Integer threadCount;
+    
     private Properties props;
 }
 

sharding-metrics/sharding-metrics-configuration/src/main/java/org/apache/shardingsphere/metrics/configuration/swapper/MetricsConfigurationYamlSwapper.java

@@ -32,6 +32,8 @@ public final class MetricsConfigurationYamlSwapper implements YamlSwapper<YamlMe
         configuration.setHost(metricsConfiguration.getHost());
         configuration.setName(metricsConfiguration.getMetricsName());
         configuration.setPort(metricsConfiguration.getPort());
+        configuration.setAsync(metricsConfiguration.getAsync());
+        configuration.setThreadCount(metricsConfiguration.getThreadCount());
         configuration.setProps(metricsConfiguration.getProps());
         return configuration;
     }
@@ -39,7 +41,10 @@ public final class MetricsConfigurationYamlSwapper implements YamlSwapper<YamlMe
     @Override
     public MetricsConfiguration swap(final YamlMetricsConfiguration metricsConfiguration) {
         return new MetricsConfiguration(metricsConfiguration.getName(), metricsConfiguration.getHost(),
-                null == metricsConfiguration.getPort() ? 9190 : metricsConfiguration.getPort(), metricsConfiguration.getProps());
+                null == metricsConfiguration.getPort() ? 9190 : metricsConfiguration.getPort(),
+                null == metricsConfiguration.getAsync(),
+                null == metricsConfiguration.getThreadCount() ? Runtime.getRuntime().availableProcessors() << 1 : metricsConfiguration.getThreadCount(),
+                metricsConfiguration.getProps());
     }
 }
 

sharding-metrics/sharding-metrics-configuration/src/main/java/org/apache/shardingsphere/metrics/configuration/yaml/YamlMetricsConfiguration.java

@@ -36,6 +36,10 @@ public final class YamlMetricsConfiguration implements YamlConfiguration {
     
     private Integer port;
     
+    private Boolean async;
+    
+    private Integer threadCount;
+    
     private Properties props = new Properties();
 }
 

sharding-metrics/sharding-metrics-configuration/src/test/java/org/apache/shardingsphere/metrics/configuration/swapper/MetricsConfigurationYamlSwapperTest.java

@@ -21,24 +21,55 @@ import org.apache.shardingsphere.metrics.configuration.config.MetricsConfigurati
 import org.apache.shardingsphere.metrics.configuration.yaml.YamlMetricsConfiguration;
 import org.junit.Test;
 
-import java.util.Properties;
-
 import static org.hamcrest.CoreMatchers.is;
 import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThat;
 
 public final class MetricsConfigurationYamlSwapperTest {
     
     @Test
-    public void swap() {
+    public void assertSwapDefault() {
+        MetricsConfigurationYamlSwapper swapper = new MetricsConfigurationYamlSwapper();
+        YamlMetricsConfiguration yaml = new YamlMetricsConfiguration();
+        yaml.setHost("127.0.0.1");
+        yaml.setName("prometheus");
+        
+        MetricsConfiguration metricsConfiguration = swapper.swap(yaml);
+        assertThat(metricsConfiguration.getPort(), is(9190));
+        assertThat(metricsConfiguration.getAsync(), is(true));
+        assertThat(metricsConfiguration.getThreadCount(), is(Runtime.getRuntime().availableProcessors() << 1));
+    
+        YamlMetricsConfiguration yamlSwap = swapper.swap(metricsConfiguration);
+        assertNotNull(yamlSwap);
+        assertThat(yamlSwap.getPort(), is(9190));
+        assertThat(yamlSwap.getName(), is("prometheus"));
+        assertThat(yamlSwap.getHost(), is("127.0.0.1"));
+        assertThat(yamlSwap.getAsync(), is(true));
+        assertThat(yamlSwap.getThreadCount(), is(Runtime.getRuntime().availableProcessors() << 1));
+    }
+    
+    @Test
+    public void assertSwapFull() {
         MetricsConfigurationYamlSwapper swapper = new MetricsConfigurationYamlSwapper();
-        YamlMetricsConfiguration yamlConfiguration = swapper.swap(new MetricsConfiguration("prometheus", "127.0.0.1", null, new Properties()));
-        assertNotNull(yamlConfiguration);
-        assertNull(yamlConfiguration.getPort());
-        MetricsConfiguration configuration = swapper.swap(yamlConfiguration);
-        assertNotNull(configuration);
-        assertThat(configuration.getPort(), is(9190));
+        YamlMetricsConfiguration yaml = new YamlMetricsConfiguration();
+        yaml.setHost("127.0.0.1");
+        yaml.setName("prometheus");
+        yaml.setPort(9195);
+        yaml.setThreadCount(8);
+        yaml.setAsync(false);
+        
+        MetricsConfiguration metricsConfiguration = swapper.swap(yaml);
+        assertThat(metricsConfiguration.getPort(), is(9195));
+        assertThat(metricsConfiguration.getAsync(), is(false));
+        assertThat(metricsConfiguration.getThreadCount(), is(8));
+        
+        YamlMetricsConfiguration yamlSwap = swapper.swap(metricsConfiguration);
+        assertNotNull(yamlSwap);
+        assertThat(yamlSwap.getPort(), is(9195));
+        assertThat(yamlSwap.getName(), is("prometheus"));
+        assertThat(yamlSwap.getHost(), is("127.0.0.1"));
+        assertThat(yamlSwap.getAsync(), is(false));
+        assertThat(yamlSwap.getThreadCount(), is(8));
     }
 }
 

sharding-metrics/sharding-metrics-facade/src/main/java/org/apache/shardingsphere/metrics/facade/MetricsTrackerFacade.java

@@ -19,19 +19,11 @@ package org.apache.shardingsphere.metrics.facade;
 
 import com.google.common.base.Preconditions;
 import lombok.Getter;
-import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.shardingsphere.metrics.api.CounterMetricsTracker;
-import org.apache.shardingsphere.metrics.api.GaugeMetricsTracker;
-import org.apache.shardingsphere.metrics.api.HistogramMetricsTracker;
 import org.apache.shardingsphere.metrics.api.HistogramMetricsTrackerDelegate;
-import org.apache.shardingsphere.metrics.api.MetricsTracker;
-import org.apache.shardingsphere.metrics.api.NoneHistogramMetricsTrackerDelegate;
-import org.apache.shardingsphere.metrics.api.NoneSummaryMetricsTrackerDelegate;
-import org.apache.shardingsphere.metrics.api.SummaryMetricsTracker;
 import org.apache.shardingsphere.metrics.api.SummaryMetricsTrackerDelegate;
 import org.apache.shardingsphere.metrics.configuration.config.MetricsConfiguration;
-import org.apache.shardingsphere.metrics.enums.MetricsTypeEnum;
+import org.apache.shardingsphere.metrics.facade.handler.MetricsTrackerHandler;
 import org.apache.shardingsphere.metrics.spi.MetricsTrackerManager;
 
 import java.util.HashMap;
@@ -53,7 +45,6 @@ public final class MetricsTrackerFacade {
     private MetricsTrackerManager metricsTrackerManager;
     
     @Getter
-    @Setter
     private volatile boolean enabled;
     
     private MetricsTrackerFacade() {
@@ -88,6 +79,7 @@ public final class MetricsTrackerFacade {
         metricsTrackerManager = findMetricsTrackerManager(metricsConfiguration.getMetricsName());
         Preconditions.checkNotNull(metricsTrackerManager, "Can not find metrics tracker manager with metrics name in metrics configuration.");
         metricsTrackerManager.start(metricsConfiguration);
+        MetricsTrackerHandler.getInstance().init(metricsConfiguration.getAsync(), metricsConfiguration.getThreadCount(), metricsTrackerManager);
         enabled = true;
     }
     
@@ -99,8 +91,7 @@ public final class MetricsTrackerFacade {
      */
     public void counterInc(final String metricsLabel, final String... labelValues) {
         if (enabled) {
-            metricsTrackerManager.getMetricsTrackerFactory().create(MetricsTypeEnum.COUNTER.name(), metricsLabel)
-                .ifPresent(metricsTracker -> ((CounterMetricsTracker) metricsTracker).inc(1.0, labelValues));
+            MetricsTrackerHandler.getInstance().counterInc(metricsLabel, labelValues);
         }
     }
     
@@ -112,8 +103,7 @@ public final class MetricsTrackerFacade {
      */
     public void gaugeInc(final String metricsLabel, final String... labelValues) {
         if (enabled) {
-            metricsTrackerManager.getMetricsTrackerFactory().create(MetricsTypeEnum.GAUGE.name(), metricsLabel)
-                .ifPresent(metricsTracker -> ((GaugeMetricsTracker) metricsTracker).inc(1.0, labelValues));
+            MetricsTrackerHandler.getInstance().gaugeInc(metricsLabel, labelValues);
         }
     }
     
@@ -125,8 +115,7 @@ public final class MetricsTrackerFacade {
      */
     public void gaugeDec(final String metricsLabel, final String... labelValues) {
         if (enabled) {
-            metricsTrackerManager.getMetricsTrackerFactory().create(MetricsTypeEnum.GAUGE.name(), metricsLabel)
-                .ifPresent(metricsTracker -> ((GaugeMetricsTracker) metricsTracker).dec(1.0, labelValues));
+            MetricsTrackerHandler.getInstance().gaugeDec(metricsLabel, labelValues);
         }
     }
     
@@ -141,8 +130,7 @@ public final class MetricsTrackerFacade {
         if (!enabled) {
             return Optional.empty();
         }
-        Optional<MetricsTracker> metricsTracker = metricsTrackerManager.getMetricsTrackerFactory().create(MetricsTypeEnum.HISTOGRAM.name(), metricsLabel);
-        return metricsTracker.map(tracker -> Optional.of(((HistogramMetricsTracker) tracker).startTimer(labelValues))).orElseGet(() -> Optional.of(new NoneHistogramMetricsTrackerDelegate()));
+        return MetricsTrackerHandler.getInstance().histogramStartTimer(metricsLabel, labelValues);
     }
     
     /**
@@ -151,7 +139,7 @@ public final class MetricsTrackerFacade {
      * @param delegate histogram metrics tracker delegate
      */
     public void histogramObserveDuration(final HistogramMetricsTrackerDelegate delegate) {
-        delegate.observeDuration();
+        MetricsTrackerHandler.getInstance().histogramObserveDuration(delegate);
     }
     
     /**
@@ -165,8 +153,7 @@ public final class MetricsTrackerFacade {
         if (!enabled) {
             return Optional.empty();
         }
-        Optional<MetricsTracker> metricsTracker = metricsTrackerManager.getMetricsTrackerFactory().create(MetricsTypeEnum.SUMMARY.name(), metricsLabel);
-        return metricsTracker.map(tracker -> Optional.of(((SummaryMetricsTracker) tracker).startTimer(labelValues))).orElseGet(() -> Optional.of(new NoneSummaryMetricsTrackerDelegate()));
+        return MetricsTrackerHandler.getInstance().summaryStartTimer(metricsLabel, labelValues);
     }
     
     /**
@@ -175,7 +162,7 @@ public final class MetricsTrackerFacade {
      * @param delegate summary metrics tracker delegate
      */
     public void summaryObserveDuration(final SummaryMetricsTrackerDelegate delegate) {
-        delegate.observeDuration();
+        MetricsTrackerHandler.getInstance().summaryObserveDuration(delegate);
     }
     
     private void loadMetricsManager() {

sharding-metrics/sharding-metrics-facade/src/main/java/org/apache/shardingsphere/metrics/facade/executor/MetricsThreadPoolExecutor.java

+/*
+ * 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.shardingsphere.metrics.facade.executor;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Metrics thread pool executor.
+ */
+@Slf4j
+public final class MetricsThreadPoolExecutor extends ThreadPoolExecutor {
+    
+    @Getter
+    private final String name;
+    
+    /**
+     * Instantiates a new Metrics thread pool executor.
+     *
+     * @param nameFormat  thread name format
+     * @param threadCount  core and max thread count
+     * @param queueSize  queue size
+     */
+    public MetricsThreadPoolExecutor(final String nameFormat, final int threadCount, final int queueSize) {
+        super(threadCount, threadCount, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(queueSize),
+                new ThreadFactoryBuilder().setDaemon(true).setNameFormat(nameFormat).build(), buildRejectedExecutionHandler(queueSize));
+        this.name = nameFormat;
+    }
+    
+    private static RejectedExecutionHandler buildRejectedExecutionHandler(final int size) {
+        return (r, executor) -> {
+            BlockingQueue<Runnable> queue = executor.getQueue();
+            while (queue.size() >= size) {
+                if (executor.isShutdown()) {
+                    throw new RejectedExecutionException("metrics thread pool executor closed");
+                }
+                ((MetricsThreadPoolExecutor) executor).onRejected();
+            }
+            if (!executor.isShutdown()) {
+                executor.execute(r);
+            }
+        };
+    }
+    
+    private void onRejected() {
+        log.info("...thread:{}, Saturation occurs, actuator:{}", Thread.currentThread().getName(), name);
+    }
+}
+

sharding-metrics/sharding-metrics-facade/src/main/java/org/apache/shardingsphere/metrics/facade/handler/MetricsTrackerHandler.java

+/*
+ * 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.shardingsphere.metrics.facade.handler;
+
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.shardingsphere.metrics.api.CounterMetricsTracker;
+import org.apache.shardingsphere.metrics.api.GaugeMetricsTracker;
+import org.apache.shardingsphere.metrics.api.HistogramMetricsTracker;
+import org.apache.shardingsphere.metrics.api.HistogramMetricsTrackerDelegate;
+import org.apache.shardingsphere.metrics.api.MetricsTracker;
+import org.apache.shardingsphere.metrics.api.NoneHistogramMetricsTrackerDelegate;
+import org.apache.shardingsphere.metrics.api.NoneSummaryMetricsTrackerDelegate;
+import org.apache.shardingsphere.metrics.api.SummaryMetricsTracker;
+import org.apache.shardingsphere.metrics.api.SummaryMetricsTrackerDelegate;
+import org.apache.shardingsphere.metrics.enums.MetricsTypeEnum;
+import org.apache.shardingsphere.metrics.facade.executor.MetricsThreadPoolExecutor;
+import org.apache.shardingsphere.metrics.spi.MetricsTrackerManager;
+
+import java.util.Arrays;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * Metrics tracker handler.
+ */
+@Slf4j
+public final class MetricsTrackerHandler {
+    
+    private static final String NAME_FORMAT = "ShardingSphere-Metrics-%d";
+    
+    private static final MetricsTrackerHandler INSTANCE = new MetricsTrackerHandler();
+    
+    private static final int FUTURE_GET_TIME_OUT_MILLISECONDS = 500;
+    
+    private static final int QUEUE_SIZE = 5000;
+    
+    @Getter
+    private MetricsTrackerManager metricsTrackerManager;
+    
+    @Getter
+    private ExecutorService executorService;
+    
+    private volatile boolean async;
+    
+    /**
+     * Gets instance.
+     *
+     * @return the instance
+     */
+    public static MetricsTrackerHandler getInstance() {
+        return INSTANCE;
+    }
+    
+    /**
+     * Init for metrics tracker handler.
+     *
+     * @param async async
+     * @param threadCount thread count
+     * @param metricsTrackerManager metrics tracker manager
+     */
+    public void init(final boolean async, final int threadCount, final MetricsTrackerManager metricsTrackerManager) {
+        this.async = async;
+        this.metricsTrackerManager = metricsTrackerManager;
+        if (async) {
+            executorService = new MetricsThreadPoolExecutor(NAME_FORMAT, threadCount, QUEUE_SIZE);
+        }
+    }
+    
+    /**
+     * Increment of counter metrics tracker.
+     *
+     * @param metricsLabel metrics label
+     * @param labelValues  label values
+     */
+    public void counterInc(final String metricsLabel, final String... labelValues) {
+        if (async) {
+            executorService.execute(() -> handlerCounter(metricsLabel, labelValues));
+        } else {
+            handlerCounter(metricsLabel, labelValues);
+        }
+    }
+    
+    /**
+     * Increment of gauge metrics tracker.
+     *
+     * @param metricsLabel metrics label
+     * @param labelValues  label values
+     */
+    public void gaugeInc(final String metricsLabel, final String... labelValues) {
+        if (async) {
+            executorService.execute(() -> handlerGaugeInc(metricsLabel, labelValues));
+        } else {
+            handlerGaugeInc(metricsLabel, labelValues);
+        }
+    }
+    
+    /**
+     * Decrement of gauge metrics tracker.
+     *
+     * @param metricsLabel metrics label
+     * @param labelValues  label values
+     */
+    public void gaugeDec(final String metricsLabel, final String... labelValues) {
+        if (async) {
+            executorService.execute(() -> handlerGaugeDec(metricsLabel, labelValues));
+        } else {
+            handlerGaugeDec(metricsLabel, labelValues);
+        }
+    }
+    
+    /**
+     * Start timer of histogram metrics tracker.
+     *
+     * @param metricsLabel metrics label
+     * @param labelValues  label values
+     * @return histogram metrics tracker delegate
+     */
+    public Optional<HistogramMetricsTrackerDelegate> histogramStartTimer(final String metricsLabel, final String... labelValues) {
+        if (async) {
+            try {
+                return executorService.submit(() -> handlerHistogramStartTimer(metricsLabel, labelValues)).get(FUTURE_GET_TIME_OUT_MILLISECONDS, TimeUnit.MILLISECONDS);
+            } catch (InterruptedException | ExecutionException | TimeoutException e) {
+                throw new IllegalStateException(String.format("Error while fetching histogram metric with metricsLabel= %s and labelValues=%s", metricsLabel, Arrays.toString(labelValues)), e);
+            }
+        } else {
+            return handlerHistogramStartTimer(metricsLabel, labelValues);
+        }
+    }
+    
+    /**
+     * Observe amount of time since start time with histogram metrics tracker.
+     *
+     * @param delegate histogram metrics tracker delegate
+     */
+    public void histogramObserveDuration(final HistogramMetricsTrackerDelegate delegate) {
+        delegate.observeDuration();
+    }
+    
+    /**
+     * Start timer of summary metrics tracker.
+     *
+     * @param metricsLabel metrics label
+     * @param labelValues  label values
+     * @return summary metrics tracker delegate
+     */
+    public Optional<SummaryMetricsTrackerDelegate> summaryStartTimer(final String metricsLabel, final String... labelValues) {
+        if (async) {
+            try {
+                return executorService.submit(() -> handlerSummaryStartTimer(metricsLabel, labelValues)).get(FUTURE_GET_TIME_OUT_MILLISECONDS, TimeUnit.SECONDS);
+            } catch (InterruptedException | ExecutionException | TimeoutException e) {
+                throw new IllegalStateException(String.format("Error while fetching summary metric with metricsLabel= %s and labelValues=%s", metricsLabel, Arrays.toString(labelValues)), e);
+            }
+        } else {
+            return handlerSummaryStartTimer(metricsLabel, labelValues);
+        }
+    }
+    
+    /**
+     * Observe amount of time since start time with summary metrics tracker.
+     *
+     * @param delegate summary metrics tracker delegate
+     */
+    public void summaryObserveDuration(final SummaryMetricsTrackerDelegate delegate) {
+        delegate.observeDuration();
+    }
+    
+    /**
+     * Executor service close.
+     */
+    public void close() {
+        if (null != executorService && !executorService.isShutdown()) {
+            executorService.shutdown();
+        }
+    }
+    
+    private void handlerCounter(final String metricsLabel, final String... labelValues) {
+        metricsTrackerManager.getMetricsTrackerFactory().create(MetricsTypeEnum.COUNTER.name(), metricsLabel)
+                .ifPresent(metricsTracker -> ((CounterMetricsTracker) metricsTracker).inc(1.0, labelValues));
+    }
+    
+    private void handlerGaugeInc(final String metricsLabel, final String... labelValues) {
+        metricsTrackerManager.getMetricsTrackerFactory().create(MetricsTypeEnum.GAUGE.name(), metricsLabel)
+                .ifPresent(metricsTracker -> ((GaugeMetricsTracker) metricsTracker).inc(1.0, labelValues));
+    }
+    
+    /**
+     * Handler gauge dec.
+     *
+     * @param metricsLabel the metrics label
+     * @param labelValues  the label values
+     */
+    public void handlerGaugeDec(final String metricsLabel, final String... labelValues) {
+        metricsTrackerManager.getMetricsTrackerFactory().create(MetricsTypeEnum.GAUGE.name(), metricsLabel)
+                .ifPresent(metricsTracker -> ((GaugeMetricsTracker) metricsTracker).dec(1.0, labelValues));
+    }
+    
+    private Optional<HistogramMetricsTrackerDelegate> handlerHistogramStartTimer(final String metricsLabel, final String... labelValues) {
+        Optional<MetricsTracker> metricsTracker = metricsTrackerManager.getMetricsTrackerFactory().create(MetricsTypeEnum.HISTOGRAM.name(), metricsLabel);
+        return metricsTracker.map(tracker -> Optional.of(((HistogramMetricsTracker) tracker).startTimer(labelValues))).orElseGet(() -> Optional.of(new NoneHistogramMetricsTrackerDelegate()));
+    }
+    
+    private Optional<SummaryMetricsTrackerDelegate> handlerSummaryStartTimer(final String metricsLabel, final String... labelValues) {
+        Optional<MetricsTracker> metricsTracker = metricsTrackerManager.getMetricsTrackerFactory().create(MetricsTypeEnum.SUMMARY.name(), metricsLabel);
+        return metricsTracker.map(tracker -> Optional.of(((SummaryMetricsTracker) tracker).startTimer(labelValues))).orElseGet(() -> Optional.of(new NoneSummaryMetricsTrackerDelegate()));
+    }
+}
+

sharding-metrics/sharding-metrics-facade/src/test/java/org/apache/shardingsphere/metrics/facade/MetricsTrackerFacadeTest.java

@@ -22,9 +22,8 @@ import org.apache.shardingsphere.metrics.api.NoneHistogramMetricsTrackerDelegate
 import org.apache.shardingsphere.metrics.api.NoneSummaryMetricsTrackerDelegate;
 import org.apache.shardingsphere.metrics.api.SummaryMetricsTrackerDelegate;
 import org.apache.shardingsphere.metrics.configuration.config.MetricsConfiguration;
-import org.apache.shardingsphere.metrics.facade.fixture.FirstMetricsTrackerFactoryFixture;
-import org.apache.shardingsphere.metrics.facade.fixture.SecondMetricsTrackerFactoryFixture;
 import org.apache.shardingsphere.metrics.facade.fixture.SecondMetricsTrackerManagerFixture;
+import org.apache.shardingsphere.metrics.facade.util.FieldUtil;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -44,9 +43,14 @@ public final class MetricsTrackerFacadeTest {
     
     @Before
     public void setUp() {
-        MetricsConfiguration metricsConfiguration = new MetricsConfiguration("fixture", null, null, null);
+        MetricsConfiguration metricsConfiguration = new MetricsConfiguration("fixture", null, null, false, 8, null);
+        metricsTrackerFacade.init(metricsConfiguration);
+    }
+    
+    @Test
+    public void assertInit() {
+        MetricsConfiguration metricsConfiguration = new MetricsConfiguration("fixture", null, null, false, 8, null);
         metricsTrackerFacade.init(metricsConfiguration);
-        assertThat(metricsTrackerFacade.getMetricsTrackerManager().getClass().getName(), is(SecondMetricsTrackerManagerFixture.class.getName()));
         assertThat(metricsTrackerFacade.isEnabled(), is(true));
     }
     
@@ -57,60 +61,48 @@ public final class MetricsTrackerFacadeTest {
     }
     
     @Test
-    public void counterInc() {
+    public void assertCounterInc() {
         metricsTrackerFacade.counterInc("request_total");
     }
     
     @Test
-    public void gaugeInc() {
+    public void assertGaugeInc() {
         metricsTrackerFacade.gaugeInc("request_total");
     }
     
     @Test
-    public void gaugeDec() {
+    public void assertGaugeDec() {
         metricsTrackerFacade.gaugeDec("request_total");
     }
     
     @Test
-    public void histogram() {
+    public void assertHistogram() {
         assertThat(metricsTrackerFacade.getMetricsTrackerManager().getClass().getName(), is(SecondMetricsTrackerManagerFixture.class.getName()));
-        ((SecondMetricsTrackerManagerFixture) metricsTrackerFacade.getMetricsTrackerManager()).setMetricsTrackerFactory(new SecondMetricsTrackerFactoryFixture());
         Optional<HistogramMetricsTrackerDelegate> histogramDelegate = metricsTrackerFacade.histogramStartTimer("request");
         assertThat(histogramDelegate.isPresent(), is(true));
         histogramDelegate.ifPresent(delegate -> {
             metricsTrackerFacade.histogramObserveDuration(delegate);
             assertThat(delegate.getClass().getName(), is(NoneHistogramMetricsTrackerDelegate.class.getName()));
         });
+    
+        FieldUtil.setField(metricsTrackerFacade, "enabled", false);
+        Optional<HistogramMetricsTrackerDelegate> empty = metricsTrackerFacade.histogramStartTimer("request");
+        assertThat(empty, is(Optional.empty()));
     }
     
     @Test
     public void summary() {
         assertThat(metricsTrackerFacade.getMetricsTrackerManager().getClass().getName(), is(SecondMetricsTrackerManagerFixture.class.getName()));
-        ((SecondMetricsTrackerManagerFixture) metricsTrackerFacade.getMetricsTrackerManager()).setMetricsTrackerFactory(new SecondMetricsTrackerFactoryFixture());
         Optional<SummaryMetricsTrackerDelegate> summaryDelegate = metricsTrackerFacade.summaryStartTimer("request");
         assertThat(summaryDelegate.isPresent(), is(true));
         summaryDelegate.ifPresent(delegate -> {
             metricsTrackerFacade.summaryObserveDuration(delegate);
             assertThat(delegate.getClass().getName(), is(NoneSummaryMetricsTrackerDelegate.class.getName()));
         });
-    }
     
-    @Test
-    public void testNoneDelegate() {
-        ((SecondMetricsTrackerManagerFixture) metricsTrackerFacade.getMetricsTrackerManager()).setMetricsTrackerFactory(new FirstMetricsTrackerFactoryFixture());
-        Optional<SummaryMetricsTrackerDelegate> summaryDelegate = metricsTrackerFacade.summaryStartTimer("request");
-        summaryDelegate.ifPresent(delegate -> assertThat(delegate.getClass().getName(), is(NoneSummaryMetricsTrackerDelegate.class.getName())));
-        Optional<HistogramMetricsTrackerDelegate> histogramDelegate = metricsTrackerFacade.histogramStartTimer("request");
-        histogramDelegate.ifPresent(delegate -> assertThat(delegate.getClass().getName(), is(NoneHistogramMetricsTrackerDelegate.class.getName())));
-    }
-    
-    @Test
-    public void testNotInit() {
-        metricsTrackerFacade.setEnabled(false);
-        Optional<HistogramMetricsTrackerDelegate> histogramDelegate = metricsTrackerFacade.histogramStartTimer("request");
-        assertThat(histogramDelegate, is(Optional.empty()));
-        Optional<SummaryMetricsTrackerDelegate> summaryDelegate = metricsTrackerFacade.summaryStartTimer("request");
-        assertThat(summaryDelegate, is(Optional.empty()));
+        FieldUtil.setField(metricsTrackerFacade, "enabled", false);
+        Optional<SummaryMetricsTrackerDelegate> empty = metricsTrackerFacade.summaryStartTimer("request");
+        assertThat(empty, is(Optional.empty()));
     }
 }
 

sharding-metrics/sharding-metrics-facade/src/test/java/org/apache/shardingsphere/metrics/facade/executor/MetricsThreadPoolExecutorTest.java

+/*
+ * 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.shardingsphere.metrics.facade.executor;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+@RunWith(MockitoJUnitRunner.class)
+public final class MetricsThreadPoolExecutorTest {
+    
+    @Test
+    public void assertInit() {
+        MetricsThreadPoolExecutor executor = new MetricsThreadPoolExecutor("test", 8, 100);
+        assertThat(executor.getName(), is("test"));
+        assertThat(executor.getCorePoolSize(), is(8));
+        assertThat(executor.getMaximumPoolSize(), is(8));
+        assertThat(executor.getQueue().getClass().getName(), is(LinkedBlockingQueue.class.getName()));
+        assertThat(executor.isShutdown(), is(false));
+        assertThat(executor.getKeepAliveTime(TimeUnit.MILLISECONDS), is(0L));
+        executor.shutdown();
+    }
+    
+    @Test
+    public void assertRejectedExecutionHandler() {
+        MetricsThreadPoolExecutor executor = new MetricsThreadPoolExecutor("test", 1, 1);
+        for (int i = 0; i < 10; i++) {
+            executor.execute(() -> {
+            
+            });
+        }
+        executor.shutdown();
+    }
+}
+

sharding-metrics/sharding-metrics-facade/src/test/java/org/apache/shardingsphere/metrics/facade/handler/MetricsTrackerHandlerTest.java

+/*
+ * 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.shardingsphere.metrics.facade.handler;
+
+import org.apache.shardingsphere.metrics.api.HistogramMetricsTrackerDelegate;
+import org.apache.shardingsphere.metrics.api.NoneHistogramMetricsTrackerDelegate;
+import org.apache.shardingsphere.metrics.api.NoneSummaryMetricsTrackerDelegate;
+import org.apache.shardingsphere.metrics.api.SummaryMetricsTrackerDelegate;
+import org.apache.shardingsphere.metrics.facade.fixture.SecondMetricsTrackerManagerFixture;
+import org.apache.shardingsphere.metrics.facade.util.FieldUtil;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Optional;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+public final class MetricsTrackerHandlerTest {
+    
+    private static final String METRICS_LABEL = "test";
+    
+    private final MetricsTrackerHandler handler = MetricsTrackerHandler.getInstance();
+    
+    @Before
+    public void init() {
+        handler.init(true, Runtime.getRuntime().availableProcessors() << 1, new SecondMetricsTrackerManagerFixture());
+    }
+    
+    @Test
+    public void counterInc() {
+        handler.counterInc(METRICS_LABEL);
+        FieldUtil.setField(handler, "async", false);
+        handler.counterInc(METRICS_LABEL);
+    }
+    
+    @Test
+    public void gaugeInc() {
+        handler.gaugeInc(METRICS_LABEL);
+        FieldUtil.setField(handler, "async", false);
+        handler.gaugeInc(METRICS_LABEL);
+    }
+    
+    @Test
+    public void gaugeDec() {
+        handler.gaugeDec(METRICS_LABEL);
+        FieldUtil.setField(handler, "async", false);
+        handler.gaugeDec(METRICS_LABEL);
+    }
+    
+    @Test
+    public void assertHistogram() {
+        Optional<HistogramMetricsTrackerDelegate> histogramDelegate = handler.histogramStartTimer(METRICS_LABEL);
+        assertThat(histogramDelegate.isPresent(), is(true));
+        histogramDelegate.ifPresent(delegate -> {
+            handler.histogramObserveDuration(delegate);
+            assertThat(delegate.getClass().getName(), is(NoneHistogramMetricsTrackerDelegate.class.getName()));
+        });
+        FieldUtil.setField(handler, "async", false);
+        Optional<HistogramMetricsTrackerDelegate> syncHistogram = handler.histogramStartTimer(METRICS_LABEL);
+        assertThat(syncHistogram.isPresent(), is(true));
+        syncHistogram.ifPresent(delegate -> assertThat(delegate.getClass().getName(), is(NoneHistogramMetricsTrackerDelegate.class.getName())));
+    }
+    
+    @Test
+    public void summary() {
+        Optional<SummaryMetricsTrackerDelegate> summaryDelegate = handler.summaryStartTimer(METRICS_LABEL);
+        assertThat(summaryDelegate.isPresent(), is(true));
+        summaryDelegate.ifPresent(delegate -> {
+            handler.summaryObserveDuration(delegate);
+            assertThat(delegate.getClass().getName(), is(NoneSummaryMetricsTrackerDelegate.class.getName()));
+        });
+        
+        FieldUtil.setField(handler, "async", false);
+        Optional<SummaryMetricsTrackerDelegate> syncSummary = handler.summaryStartTimer(METRICS_LABEL);
+        assertThat(syncSummary.isPresent(), is(true));
+        syncSummary.ifPresent(delegate -> assertThat(delegate.getClass().getName(), is(NoneSummaryMetricsTrackerDelegate.class.getName())));
+    }
+    
+    @After
+    public void assertClose() {
+        handler.close();
+        assertThat(handler.getExecutorService().isShutdown(), is(true));
+    }
+}
+

sharding-metrics/sharding-metrics-facade/src/test/java/org/apache/shardingsphere/metrics/facade/util/FieldUtil.java

+/*
+ * 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.shardingsphere.metrics.facade.util;
+
+import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
+
+import java.lang.reflect.Field;
+
+/**
+ * Field util.
+ */
+@RequiredArgsConstructor
+public final class FieldUtil {
+    
+    /**
+     * Set field.
+     *
+     * @param target target to be settled
+     * @param fieldName field name to be settled
+     * @param fieldValue field value to be settled
+     */
+    @SneakyThrows
+    public static void setField(final Object target, final String fieldName, final Object fieldValue) {
+        Field field = target.getClass().getDeclaredField(fieldName);
+        field.setAccessible(true);
+        field.set(target, fieldValue);
+    }
+}

sharding-metrics/sharding-metrics-prometheus/src/test/java/org/apache/shardingsphere/metrics/prometheus/PrometheusMetricsTrackerManagerTest.java

@@ -33,7 +33,7 @@ public final class PrometheusMetricsTrackerManagerTest {
     @Test
     public void startNoHost() {
         PrometheusMetricsTrackerManager manager = new PrometheusMetricsTrackerManager();
-        MetricsConfiguration metricsConfiguration = new MetricsConfiguration("metricsName", "", 9191, null);
+        MetricsConfiguration metricsConfiguration = new MetricsConfiguration("metricsName", "", 9191, false, 8, null);
         manager.start(metricsConfiguration);
         HTTPServer server = manager.getServer();
         assertNotNull(server);
@@ -44,7 +44,7 @@ public final class PrometheusMetricsTrackerManagerTest {
     @Test
     public void startHost() {
         PrometheusMetricsTrackerManager manager = new PrometheusMetricsTrackerManager();
-        MetricsConfiguration metricsConfiguration = new MetricsConfiguration("metricsName", "127.0.0.1", 9195, null);
+        MetricsConfiguration metricsConfiguration = new MetricsConfiguration("metricsName", "127.0.0.1", 9195, false, 8, null);
         manager.start(metricsConfiguration);
         HTTPServer server = manager.getServer();
         assertThat(server.getPort(), is(9195));