Add Method to Extract Undecorated Topic from Datastream Metadata for Accurate SLA Latency Reporting (#984)

* Add functionality to send undecorated topic to support any type of topic names for reporting latency metrics

* Not required anymore

Author: harshcum

datastream-server-api/src/main/java/com/linkedin/datastream/server/api/transport/DatastreamRecordMetadata.java

@@ -123,6 +123,17 @@ public String getTopic() {
     return _topic;
   }
 
+ /**
+  * Gets the topic name without any decorations or modifications.
+  * This base implementation simply returns the topic name as is.
+  * Subclasses may override this method to provide custom topic name undecorating logic.
+  * @return The undecorated topic name
+  */
+ public String getUndecoratedTopic() {
+   // It is kept for backward compatibility.
+   return _topic;
+ }
+
   /**
    * An index identifying the exact {@link com.linkedin.datastream.common.BrooklinEnvelope} event produced,
    * from those obtainable through {@link com.linkedin.datastream.server.DatastreamProducerRecord#getEvents()}

datastream-server/src/main/java/com/linkedin/datastream/server/EventProducer.java

@@ -337,7 +337,7 @@ private void performSlaRelatedLogging(DatastreamRecordMetadata metadata, long ev
     if (_numEventsOutsideAltSlaLogEnabled) {
       try {
         if (sourceToDestinationLatencyMs > _availabilityThresholdAlternateSlaMs) {
-          TopicPartition topicPartition = new TopicPartition(metadata.getTopic(), metadata.getSourcePartition());
+          TopicPartition topicPartition = new TopicPartition(metadata.getUndecoratedTopic(), metadata.getSourcePartition());
           int numEvents = _trackEventsOutsideAltSlaMap.getOrDefault(topicPartition, 0);
           _trackEventsOutsideAltSlaMap.put(topicPartition, numEvents + 1);
         }
@@ -506,7 +506,8 @@ private void onSendCallback(DatastreamRecordMetadata metadata, Exception excepti
         // Report metrics
         checkpoint(metadata.getPartition(), metadata.getCheckpoint());
         // Reporting separate metrics for throughput violating topics.
-        if (_throughputViolatingTopicsProvider.apply(_datastreamTask).contains(metadata.getTopic())) {
+
+        if (_throughputViolatingTopicsProvider.apply(_datastreamTask).contains(metadata.getUndecoratedTopic())) {
           reportMetricsForThroughputViolatingTopics(metadata, eventSourceTimestamp, eventSendTimestamp);
         } else {
           reportMetrics(metadata, eventSourceTimestamp, eventSendTimestamp);

gradle/maven.gradle

@@ -1,5 +1,5 @@
 allprojects {
-  version = "5.5.5"
+  version = "5.5.6"
 }
 
 subprojects {