[CELEBORN-1634][FOLLOWUP] Simplify the logic of the `RpcSource.addTim…

…er` and `RpcSource.updateTimer`

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

As title

### Why are the changes needed?

As title

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

GA

Closes apache#2935 from cfmcgrady/CELEBORN-1634-followup.

Authored-by: Fu Chen <cfmcgrady@gmail.com>
Signed-off-by: Fu Chen <cfmcgrady@gmail.com>

common/src/main/scala/org/apache/celeborn/common/metrics/source/AbstractSource.scala

@@ -298,6 +298,9 @@ abstract class AbstractSource(conf: CelebornConf, role: String)
 
   def updateTimer(metricsName: String, value: Long, labels: Map[String, String]): Unit = {
     val metricNameWithLabel = metricNameWithCustomizedLabels(metricsName, labels)
+    if (!namedTimers.containsKey(metricNameWithLabel)) {
+      addTimer(metricsName, labels)
+    }
     val (namedTimer, _) = namedTimers.get(metricNameWithLabel)
     namedTimer.timer.update(value, TimeUnit.NANOSECONDS)
   }

common/src/main/scala/org/apache/celeborn/common/rpc/RpcSource.scala

@@ -17,31 +17,12 @@
 
 package org.apache.celeborn.common.rpc
 
-import java.util.concurrent.ConcurrentHashMap
-
 import org.apache.celeborn.common.CelebornConf
 import org.apache.celeborn.common.metrics.source.AbstractSource
 
 class RpcSource(conf: CelebornConf) extends AbstractSource(conf, RpcSource.ROLE_RPC) {
   override def sourceName: String = RpcSource.ROLE_RPC
 
-  private val msgNameSet = ConcurrentHashMap.newKeySet[String]()
-
-  override def updateTimer(name: String, value: Long): Unit = {
-    if (!msgNameSet.contains(name)) {
-      super.addTimer(name)
-      msgNameSet.add(name)
-    }
-    super.updateTimer(name, value)
-  }
-
-  override def addTimer(name: String): Unit = {
-    if (!msgNameSet.contains(name)) {
-      super.addTimer(name)
-      msgNameSet.add(name)
-    }
-  }
-
   startCleaner()
 }