MINOR: Remove fetchQuotaMetrics and copyQuotaMetrics on close (#20394)

- Changes: Remove  fetchQuotaMetrics and copyQuotaMetrics in RemoteLogManager on close

from: #20342 (comment)

Reviewers: Kamal Chandraprakash <[email protected]>

Author: jim0987795064

storage/src/main/java/org/apache/kafka/server/log/remote/quota/RLMQuotaMetrics.java

@@ -24,7 +24,7 @@
 
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
-public class RLMQuotaMetrics {
+public class RLMQuotaMetrics implements AutoCloseable {
 
     private final SensorAccess sensorAccess;
     private final Metrics metrics;
@@ -51,4 +51,9 @@ public Sensor sensor() {
                 String.format(descriptionFormat, "maximum")), new Max());
         });
     }
+
+    @Override
+    public void close() {
+        this.metrics.removeSensor(name);
+    }
 }

storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java

@@ -314,6 +314,8 @@ private void removeMetrics() {
         metricsGroup.removeMetric(REMOTE_LOG_MANAGER_TASKS_AVG_IDLE_PERCENT_METRIC);
         metricsGroup.removeMetric(REMOTE_LOG_READER_FETCH_RATE_AND_TIME_METRIC);
         remoteStorageReaderThreadPool.removeMetrics();
+        Utils.closeQuietly(fetchQuotaMetrics, "fetchQuotaMetrics");
+        Utils.closeQuietly(copyQuotaMetrics, "copyQuotaMetrics");
     }
 
     // Visible for testing
@@ -2044,17 +2046,18 @@ public void close() {
                 followerThreadPool.close();
                 try {
                     shutdownAndAwaitTermination(remoteStorageReaderThreadPool, "RemoteStorageReaderThreadPool", 10, TimeUnit.SECONDS);
+
+                    leaderCopyRLMTasks.clear();
+                    leaderExpirationRLMTasks.clear();
+                    followerRLMTasks.clear();
+
+                    Utils.closeQuietly(indexCache, "RemoteIndexCache");
+                    Utils.closeQuietly(remoteLogMetadataManagerPlugin, "remoteLogMetadataManagerPlugin");
+                    Utils.closeQuietly(remoteStorageManagerPlugin, "remoteStorageManagerPlugin");
+                    closed = true;
                 } finally {
                     removeMetrics();
                 }
-                leaderCopyRLMTasks.clear();
-                leaderExpirationRLMTasks.clear();
-                followerRLMTasks.clear();
-
-                Utils.closeQuietly(indexCache, "RemoteIndexCache");
-                Utils.closeQuietly(remoteLogMetadataManagerPlugin, "remoteLogMetadataManagerPlugin");
-                Utils.closeQuietly(remoteStorageManagerPlugin, "remoteStorageManagerPlugin");
-                closed = true;
             }
         }
     }

storage/src/test/java/org/apache/kafka/server/log/remote/quota/RLMQuotaMetricsTest.java

@@ -28,6 +28,8 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
 
 public class RLMQuotaMetricsTest {
     private final MockTime time = new MockTime();
@@ -49,4 +51,22 @@ public void testNewSensorWhenExpired() {
         Sensor newSensor = rlmQuotaMetrics.sensor();
         assertNotEquals(sensor, newSensor);
     }
+
+    @Test
+    public void testClose() {
+        RLMQuotaMetrics quotaMetrics = new RLMQuotaMetrics(metrics, "metric", "group", "format", 5);
+
+        // Register the sensor
+        quotaMetrics.sensor();
+        var avg = metrics.metricName("metric" + "-avg", "group", String.format("format", "average"));
+
+        // Verify that metrics are created
+        assertNotNull(metrics.metric(avg));
+
+        // Close the quotaMetrics instance
+        quotaMetrics.close();
+
+        // After closing, the metrics should be removed
+        assertNull(metrics.metric(avg));
+    }
 }