[fix][test] Fix flaky test: PrometheusMetricsTest.testDuplicateMetric…

…TypeDefinitions (apache#18077) * [fix][test] Fix flaky test: PrometheusMetricsTest.testDuplicateMetricTypeDefinitions * fix
nodece · Oct 18, 2022 · eed8c74 · eed8c74
1 parent 48da869
commit eed8c74
Show file tree

Hide file tree

Showing 3 changed files with 24 additions and 2 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -562,6 +562,8 @@ public CompletableFuture<Void> closeAsync() {
             if (transactionExecutorProvider != null) {
                 transactionExecutorProvider.shutdownNow();
             }
+            MLPendingAckStoreProvider.closeBufferedWriterMetrics();
+            MLTransactionMetadataStoreProvider.closeBufferedWriterMetrics();
             if (this.offloaderStats != null) {
                 this.offloaderStats.close();
             }

diff --git a/.../java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreProvider.java b/.../java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreProvider.java
@@ -61,6 +61,16 @@ public static void initBufferedWriterMetrics(String brokerAdvertisedAddress){
         }
     }
 
+    public static void closeBufferedWriterMetrics() {
+        synchronized (MLPendingAckStoreProvider.class){
+            if (bufferedWriterMetrics == DisabledTxnLogBufferedWriterMetricsStats.DISABLED_BUFFERED_WRITER_METRICS){
+                return;
+            }
+            bufferedWriterMetrics.close();
+            bufferedWriterMetrics = DisabledTxnLogBufferedWriterMetricsStats.DISABLED_BUFFERED_WRITER_METRICS;
+        }
+    }
+
     @Override
     public CompletableFuture<PendingAckStore> newPendingAckStore(PersistentSubscription subscription) {
         CompletableFuture<PendingAckStore> pendingAckStoreFuture = new CompletableFuture<>();

diff --git a/...va/org/apache/pulsar/transaction/coordinator/impl/MLTransactionMetadataStoreProvider.java b/...va/org/apache/pulsar/transaction/coordinator/impl/MLTransactionMetadataStoreProvider.java
@@ -39,17 +39,27 @@ public class MLTransactionMetadataStoreProvider implements TransactionMetadataSt
             DisabledTxnLogBufferedWriterMetricsStats.DISABLED_BUFFERED_WRITER_METRICS;
 
     public static void initBufferedWriterMetrics(String brokerAdvertisedAddress){
-        if (bufferedWriterMetrics != DisabledTxnLogBufferedWriterMetricsStats.DISABLED_BUFFERED_WRITER_METRICS){
+        if (bufferedWriterMetrics != DisabledTxnLogBufferedWriterMetricsStats.DISABLED_BUFFERED_WRITER_METRICS) {
             return;
         }
         synchronized (MLTransactionMetadataStoreProvider.class){
-            if (bufferedWriterMetrics != DisabledTxnLogBufferedWriterMetricsStats.DISABLED_BUFFERED_WRITER_METRICS){
+            if (bufferedWriterMetrics != DisabledTxnLogBufferedWriterMetricsStats.DISABLED_BUFFERED_WRITER_METRICS) {
                 return;
             }
             bufferedWriterMetrics = new MLTransactionMetadataStoreBufferedWriterMetrics(brokerAdvertisedAddress);
         }
     }
 
+    public static void closeBufferedWriterMetrics() {
+        synchronized (MLTransactionMetadataStoreProvider.class){
+            if (bufferedWriterMetrics == DisabledTxnLogBufferedWriterMetricsStats.DISABLED_BUFFERED_WRITER_METRICS) {
+                return;
+            }
+            bufferedWriterMetrics.close();
+            bufferedWriterMetrics = DisabledTxnLogBufferedWriterMetricsStats.DISABLED_BUFFERED_WRITER_METRICS;
+        }
+    }
+
     @Override
     public CompletableFuture<TransactionMetadataStore> openStore(TransactionCoordinatorID transactionCoordinatorId,
                                                                  ManagedLedgerFactory managedLedgerFactory,