fix lost prometheus metric in OrderedExecutor

bookkeeper-common/src/main/java/org/apache/bookkeeper/common/util/OrderedExecutor.java

@@ -391,6 +391,10 @@ protected OrderedExecutor(String baseName, int numThreads, ThreadFactory threadF
             ExecutorService thread = createSingleThreadExecutor(
                     new ThreadFactoryBuilder().setNameFormat(name + "-" + getClass().getSimpleName() + "-" + i + "-%d")
                     .setThreadFactory(threadFactory).build());
+            SingleThreadExecutor ste = null;
+            if (thread instanceof SingleThreadExecutor) {
+                ste = (SingleThreadExecutor) thread;
+            }
 
             if (traceTaskExecution || preserveMdcForTaskExecution) {
                 thread = addExecutorDecorators(thread);
@@ -425,48 +429,8 @@ protected OrderedExecutor(String baseName, int numThreads, ThreadFactory threadF
                 throw new RuntimeException("Couldn't start thread " + i, e);
             }
 
-            if (thread instanceof SingleThreadExecutor) {
-                SingleThreadExecutor ste = (SingleThreadExecutor) thread;
+            if (ste != null) {
                 ste.registerMetrics(statsLogger);
-            } else if (thread instanceof ThreadPoolExecutor) {
-                ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) thread;
-                // Register gauges
-                statsLogger.scopeLabel("thread", String.valueOf(idx))
-                        .registerGauge(String.format("%s-queue", name), new Gauge<Number>() {
-                            @Override
-                            public Number getDefaultValue() {
-                                return 0;
-                            }
-
-                            @Override
-                            public Number getSample() {
-                                return threadPoolExecutor.getQueue().size();
-                            }
-                        });
-                statsLogger.scopeLabel("thread", String.valueOf(idx))
-                        .registerGauge(String.format("%s-completed-tasks", name), new Gauge<Number>() {
-                            @Override
-                            public Number getDefaultValue() {
-                                return 0;
-                            }
-
-                            @Override
-                            public Number getSample() {
-                                return threadPoolExecutor.getCompletedTaskCount();
-                            }
-                        });
-                statsLogger.scopeLabel("thread", String.valueOf(idx))
-                        .registerGauge(String.format("%s-total-tasks", name), new Gauge<Number>() {
-                            @Override
-                            public Number getDefaultValue() {
-                                return 0;
-                            }
-
-                            @Override
-                            public Number getSample() {
-                                return threadPoolExecutor.getTaskCount();
-                            }
-                        });
             }
         }
 

bookkeeper-common/src/test/java/org/apache/bookkeeper/common/util/TestOrderedExecutor.java

@@ -0,0 +1,58 @@
+/*
+ *
+ * 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.bookkeeper.common.util;
+
+import org.apache.bookkeeper.test.TestStatsProvider;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Test OrderedExecutor/Scheduler .
+ */
+public class TestOrderedExecutor {
+
+    @Test
+    public void testOrderExecutorPrometheusMetric() {
+        testGenerateMetric(false);
+        testGenerateMetric(true);
+    }
+
+    private void testGenerateMetric(boolean isTraceTaskExecution) {
+        TestStatsProvider provider = new TestStatsProvider();
+
+        TestStatsProvider.TestStatsLogger rootStatsLogger = provider.getStatsLogger("");
+        TestStatsProvider.TestStatsLogger bookieStats =
+                (TestStatsProvider.TestStatsLogger) rootStatsLogger.scope("bookkeeper_server");
+
+        OrderedExecutor executor = OrderedExecutor.newBuilder().statsLogger(bookieStats)
+                .name("test").numThreads(1).traceTaskExecution(isTraceTaskExecution).build();
+
+        TestStatsProvider.TestStatsLogger testStatsLogger = (TestStatsProvider.TestStatsLogger)
+                bookieStats.scope("thread_test_OrderedExecutor_0_0");
+
+        Assert.assertNotNull(testStatsLogger.getGauge("thread_executor_queue").getSample());
+        Assert.assertNotNull(testStatsLogger.getGauge("thread_executor_completed").getSample());
+        Assert.assertNotNull(testStatsLogger.getGauge("thread_executor_tasks_completed").getSample());
+        Assert.assertNotNull(testStatsLogger.getGauge("thread_executor_tasks_rejected").getSample());
+        Assert.assertNotNull(testStatsLogger.getGauge("thread_executor_tasks_failed").getSample());
+    }
+}