Re: [PR] [CELEBORN-1182] Support application dimension ActiveConnectionCount metric to record the number of registered connections for each application [incubator-celeborn]

["FMX (via GitHub)" <gi...@apache.org>]

FMX commented on code in PR #2167:
URL: https://github.com/apache/incubator-celeborn/pull/2167#discussion_r1475595371


##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/WorkerSource.scala:
##########
@@ -69,6 +79,41 @@ class WorkerSource(conf: CelebornConf) extends AbstractSource(conf, MetricsSyste
     val metricNameWithLabel = metricNameWithCustomizedLabels(metricsName, Map.empty)
     namedCounters.get(metricNameWithLabel).counter.getCount
   }
+
+  def connectionActive(client: TransportClient): Unit = {
+    appActiveConnections.putIfAbsent(
+      client.getChannel.id().asLongText(),
+      new util.HashSet[String]())

Review Comment:
   I think util.HashSet is not thread-safe in your scenario. Maybe a concurrent hash set would be better.



##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/WorkerSource.scala:
##########
@@ -69,6 +79,41 @@ class WorkerSource(conf: CelebornConf) extends AbstractSource(conf, MetricsSyste
     val metricNameWithLabel = metricNameWithCustomizedLabels(metricsName, Map.empty)
     namedCounters.get(metricNameWithLabel).counter.getCount
   }
+
+  def connectionActive(client: TransportClient): Unit = {
+    appActiveConnections.putIfAbsent(
+      client.getChannel.id().asLongText(),
+      new util.HashSet[String]())
+    incCounter(ACTIVE_CONNECTION_COUNT, 1)
+  }
+
+  def connectionInactive(client: TransportClient): Unit = {
+    appActiveConnections.remove(client.getChannel.id().asLongText())
+    incCounter(ACTIVE_CONNECTION_COUNT, -1)
+  }
+
+  def recordAppActiveConnection(client: TransportClient, shuffleKey: String): Unit = {
+    val applicationIds = appActiveConnections.get(client.getChannel.id().asLongText())
+    val applicationId = Utils.splitShuffleKey(shuffleKey)._1
+    if (CollectionUtils.isNotEmpty(applicationIds) && !applicationIds.contains(applicationId)) {

Review Comment:
   This method would be called in different threads. This block is not thread-safe.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org