Clear metrics when closing JournalStateMachine

### What changes are proposed in this pull request?

Clear metrics when closing JournalStateMachine,
so when a new instance of JournalStateMachine is created, those metrics
can be refreshed.


https://github.com/Alluxio/alluxio/blob/bd30e25d7b0103e30dd8de611da250ced85cdf27/core/server/common/src/main/java/alluxio/master/journal/raft/RaftJournalSystem.java#L293-L297

Fix #16704.

### Why are the changes needed?

Fix a bug that after a primacy change, metrics of JournalStateMachine
will get stuck.

### Does this PR introduce any user facing changes?

No.

pr-link: #16693
change-id: cid-572df304ceeeebe78f36d22784d65a5088c58693

core/server/common/src/main/java/alluxio/master/journal/raft/JournalStateMachine.java

@@ -367,6 +367,11 @@ public CompletableFuture<Message> query(Message request) {
   @Override
   public void close() {
     mClosed = true;
+    MetricsSystem.removeMetrics(MetricKey.MASTER_EMBEDDED_JOURNAL_SNAPSHOT_LAST_INDEX.getName());
+    MetricsSystem.removeMetrics(MetricKey.MASTER_JOURNAL_ENTRIES_SINCE_CHECKPOINT.getName());
+    MetricsSystem.removeMetrics(MetricKey.MASTER_JOURNAL_LAST_CHECKPOINT_TIME.getName());
+    MetricsSystem.removeMetrics(MetricKey.MASTER_JOURNAL_LAST_APPLIED_COMMIT_INDEX.getName());
+    MetricsSystem.removeMetrics(MetricKey.MASTER_JOURNAL_CHECKPOINT_WARN.getName());
     synchronized (mSnapshotManager) {
       mSnapshotManager.notifyAll();
     }

core/server/common/src/main/java/alluxio/master/journal/raft/RaftJournalSystem.java

@@ -1192,6 +1192,11 @@ synchronized RaftServer getRaftServer() {
     return mServer;
   }
 
+  @VisibleForTesting
+  ConcurrentHashMap<String, RaftJournal> getJournals() {
+    return mJournals;
+  }
+
   /**
    * Updates raft group with the current values from raft server.
    */

core/server/master/src/test/java/alluxio/master/journal/raft/RaftJournalSystemMetricsTest.java

@@ -12,6 +12,8 @@
 package alluxio.master.journal.raft;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 
 import alluxio.conf.Configuration;
 import alluxio.conf.PropertyKey;
@@ -43,6 +45,39 @@ public void after() {
     Configuration.reloadProperties();
   }
 
+  @Test
+  public void journalStateMachineMetrics() throws Exception {
+    Configuration.set(PropertyKey.MASTER_EMBEDDED_JOURNAL_ADDRESSES,
+        "localhost:29200,localhost:29201,localhost:29202");
+    Configuration.set(PropertyKey.MASTER_HOSTNAME, "localhost");
+    Configuration.set(PropertyKey.MASTER_EMBEDDED_JOURNAL_PORT, 29200);
+    RaftJournalSystem system =
+        new RaftJournalSystem(mFolder.newFolder().toURI(), ServiceType.MASTER_RAFT);
+    String[] metricsNames = new String[] {
+        MetricKey.MASTER_EMBEDDED_JOURNAL_SNAPSHOT_LAST_INDEX.getName(),
+        MetricKey.MASTER_JOURNAL_ENTRIES_SINCE_CHECKPOINT.getName(),
+        MetricKey.MASTER_JOURNAL_LAST_CHECKPOINT_TIME.getName(),
+        MetricKey.MASTER_JOURNAL_LAST_APPLIED_COMMIT_INDEX.getName(),
+        MetricKey.MASTER_JOURNAL_CHECKPOINT_WARN.getName(),
+    };
+    JournalStateMachine stateMachine = new JournalStateMachine(system.getJournals(), system);
+    for (String name : metricsNames) {
+      assertNotNull(MetricsSystem.METRIC_REGISTRY.getGauges().get(name));
+    }
+    stateMachine.close();
+    for (String name : metricsNames) {
+      assertNull(MetricsSystem.METRIC_REGISTRY.getGauges().get(name));
+    }
+    JournalStateMachine newStateMachine = new JournalStateMachine(system.getJournals(), system);
+    for (String name : metricsNames) {
+      assertNotNull(MetricsSystem.METRIC_REGISTRY.getGauges().get(name));
+    }
+    newStateMachine.close();
+    for (String name : metricsNames) {
+      assertNull(MetricsSystem.METRIC_REGISTRY.getGauges().get(name));
+    }
+  }
+
   @Test
   public void metrics() throws Exception {
     Configuration.set(PropertyKey.MASTER_EMBEDDED_JOURNAL_ADDRESSES,