Fix authentication metrics for 4.0 (#1445)

Author: coderzc

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,15 +83,29 @@ 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);
         this.loadAsync();
     }
 
     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;
         }
     }

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>