More synchronization on metrics addition (#2177)

TouK · Sep 13, 2021 · 4c4258d · 4c4258d
1 parent 1c1dcdd
commit 4c4258d
Showing 1 changed file with 12 additions and 2 deletions.
diff --git a/.../src/main/scala/pl/touk/nussknacker/engine/util/service/GenericTimeMeasuringService.scala b/.../src/main/scala/pl/touk/nussknacker/engine/util/service/GenericTimeMeasuringService.scala
@@ -38,8 +38,18 @@ trait GenericTimeMeasuringService {
     case Failure(_) => Some("FAIL")
   }
 
-  private def getOrCreateTimer(tags: Map[String, String], meterType: String) : EspTimer = metrics.getOrElseUpdate(meterType,
-    espTimer(tags + ("serviceName" -> serviceName), metricName :+ meterType))
+  private def getOrCreateTimer(tags: Map[String, String], meterType: String) : EspTimer = {
+    //TrieMap.getOrElseUpdate alone is not enough, as e.g. in Flink "espTimer" can be invoked only once - otherwise
+    //Metric may be already registered, which results in refusal to register metric without feedback. In such case
+    //we can end up using not-registered metric.
+    //The first check is for optimization purposes - to synchronize only at the beginnning
+    metrics.get(meterType) match {
+      case Some(value) => value
+      case None => synchronized {
+        metrics.getOrElseUpdate(meterType, espTimer(tags + ("serviceName" -> serviceName), metricName :+ meterType))
+      }
+    }
+  }
 
 
   def espTimer(tags: Map[String, String], metricName: NonEmptyList[String]) : EspTimer