From 60f27679ff6d677f9e1ad5603afd1424a684675d Mon Sep 17 00:00:00 2001
From: Cong Zhao <zhaocong@apache.org>
Date: Mon, 9 Sep 2024 13:33:13 +0800
Subject: [PATCH] Fix authentication metrics for 4.0 (#1445)

---
 .../AuthenticationProviderMTls.java           | 23 +++++++++++++++----
 pom.xml                                       |  2 +-
 2 files changed, 20 insertions(+), 5 deletions(-)

diff --git a/mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/identitypool/AuthenticationProviderMTls.java b/mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/identitypool/AuthenticationProviderMTls.java
index 906f9d8d..6cad681c 100644
--- a/mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/identitypool/AuthenticationProviderMTls.java
+++ b/mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/identitypool/AuthenticationProviderMTls.java
@@ -69,6 +69,7 @@ public class AuthenticationProviderMTls implements AuthenticationProvider {
     @VisibleForTesting
     private final ConcurrentHashMap<String, ExpressionCompiler> poolMap = new ConcurrentHashMap<>();
     private boolean needCloseMetaData = false;
+    private AuthenticationMetrics metrics;
 
     private enum ErrorCode {
         UNKNOWN,
@@ -82,8 +83,19 @@ private enum ErrorCode {
     }
 
     @Override
-    public void initialize(ServiceConfiguration config) throws IOException {
-        this.metadataStore = createLocalMetadataStore(config);
+    public void initialize(ServiceConfiguration conf) throws IOException {
+        initialize(Context.builder().config(conf).build());
+    }
+
+    @Override
+    public void initialize(Context context) throws IOException {
+        metrics = new AuthenticationMetrics(
+                context.getOpenTelemetry(), getClass().getSimpleName(), "mtls");
+        init(context.getConfig());
+    }
+
+    private void init(ServiceConfiguration conf) throws MetadataStoreException {
+        this.metadataStore = createLocalMetadataStore(conf);
         this.needCloseMetaData = true;
         this.metadataStore.registerListener(this::handleMetadataChanges);
         this.poolResources = new OIDCPoolResources(metadataStore);
@@ -91,6 +103,9 @@ public void initialize(ServiceConfiguration config) throws IOException {
     }
 
     public void initialize(MetadataStore metadataStore) {
+        Context context = Context.builder().build();
+        this.metrics = new AuthenticationMetrics(
+                context.getOpenTelemetry(), getClass().getSimpleName(), "mtls");
         this.metadataStore = metadataStore;
         this.metadataStore.registerListener(this::handleMetadataChanges);
         this.poolResources = new OIDCPoolResources(metadataStore);
@@ -219,10 +234,10 @@ public String authenticate(AuthenticationDataSource authData) throws Authenticat
                 errorCode = ErrorCode.NO_MATCH_POOL;
                 throw new AuthenticationException("No matched identity pool from the client certificate");
             }
-            AuthenticationMetrics.authenticateSuccess(this.getClass().getSimpleName(), this.getAuthMethodName());
+            metrics.recordSuccess();
             return principal;
         } catch (AuthenticationException e) {
-            this.incrementFailureMetric(errorCode);
+            metrics.recordFailure(errorCode);
             throw e;
         }
     }
diff --git a/pom.xml b/pom.xml
index 35e126a9..d5041e5b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -49,7 +49,7 @@
         <mockito.version>2.22.0</mockito.version>
         <testng.version>6.14.3</testng.version>
         <awaitility.version>4.0.2</awaitility.version>
-        <pulsar.version>4.0.0-SNAPSHOT</pulsar.version>
+        <pulsar.version>4.0.0-ursa-5-SNAPSHOT</pulsar.version>
         <mqtt.codec.version>4.1.94.Final</mqtt.codec.version>
         <log4j2.version>2.18.0</log4j2.version>
         <fusesource.client.version>1.16</fusesource.client.version>