#000: Review comment fix - Generation of metrics

Sanketika-Obsrv · Jan 3, 2025 · 9767d5e · 9767d5e
1 parent 88e7e41
commit 9767d5e
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 5 deletions.
diff --git a/...e/hudi-connector/src/main/scala/org/sunbird/obsrv/function/RowDataConverterFunction.scala b/...e/hudi-connector/src/main/scala/org/sunbird/obsrv/function/RowDataConverterFunction.scala
@@ -46,15 +46,11 @@ class RowDataConverterFunction(config: HudiConnectorConfig, datasetId: String) e
 
   override def map(event: MMap[String, AnyRef]): RowData = {
     try {
-      println("======= map method..")
-      println(s"Event size: ${event.size}")
-
       if (event.nonEmpty) {
-        inputEventCount.inc(event.size)  // Increment by the event size
+        inputEventCount.inc()
       }
       val rowData = convertToRowData(event)
       println(s"Current inputEventCount: ${inputEventCount.getCount}")
-
       rowData
     } catch {
       case ex: Exception =>

diff --git a/...e/hudi-connector/src/main/scala/org/sunbird/obsrv/streaming/HudiConnectorStreamTask.scala b/...e/hudi-connector/src/main/scala/org/sunbird/obsrv/streaming/HudiConnectorStreamTask.scala
@@ -9,6 +9,8 @@ import org.apache.flink.configuration.Configuration
 import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend
 import org.apache.flink.streaming.api.datastream.{DataStream, DataStreamSink}
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
+import org.apache.flink.streaming.api.functions.ProcessFunction
+import org.apache.flink.table.data.RowData
 import org.apache.hudi.common.config.TimestampKeyGeneratorConfig
 import org.apache.hudi.configuration.{FlinkOptions, OptionsResolver}
 import org.apache.hudi.sink.utils.Pipelines