Merge pull request #2 from Sanketika-Obsrv/main

Connector SDK - 1.1.0-Beta

Author: manjudr

connector-sdk-core/src/main/scala/org/sunbird/obsrv/connector/service/ConnectorRegistry.scala

@@ -24,7 +24,7 @@ object ConnectorRegistry {
   def getConnectorInstance(connectorInstanceId: String)(implicit postgresConnectionConfig: PostgresConnectionConfig): Option[ConnectorInstance] = {
     val postgresConnect = new PostgresConnect(postgresConnectionConfig)
     try {
-      val rs = postgresConnect.executeQuery(s"SELECT ci.*, d.dataset_config, cr.type as connector_type FROM connector_instances as ci JOIN connector_registry cr ON ci.connector_id = cr.id JOIN datasets d ON ci.dataset_id = d.id WHERE ci.id = '$connectorInstanceId' AND d.status = 'Live' AND cr.status='Live' AND ci.status = 'Live'")
+      val rs = postgresConnect.executeQuery(s"SELECT ci.*, d.entry_topic, cr.type as connector_type FROM connector_instances as ci JOIN connector_registry cr ON ci.connector_id = cr.id JOIN datasets d ON ci.dataset_id = d.id WHERE ci.id = '$connectorInstanceId' AND d.status = 'Live' AND cr.status='Live' AND ci.status = 'Live'")
       if (rs.next()) {
         Some(parseConnectorInstance(rs))
       } else {

connector-sdk-flink/src/main/scala/org/sunbird/obsrv/connector/source/SourceConnector.scala

@@ -14,6 +14,8 @@ import org.sunbird.obsrv.connector.util.EncryptionUtil
 import org.sunbird.obsrv.job.exception.ObsrvException
 import org.sunbird.obsrv.job.util._
 
+import java.io.File
+
 import scala.collection.mutable
 
 object SourceConnector {
@@ -23,7 +25,16 @@ object SourceConnector {
   private def getConfig(args: Array[String]): Config = {
     val configFilePathOpt = Option(ParameterTool.fromArgs(args).get("config.file.path"))
     val configFilePath = configFilePathOpt.getOrElse("config.json")
-    ConfigFactory.load(configFilePath).withFallback(ConfigFactory.load("connector.conf")).withFallback(ConfigFactory.systemEnvironment())
+    val configFile = new File(configFilePath)
+    val config: Config = if (configFile.exists()) {
+      println("Loading configuration file from path: " + configFilePath + "...")
+      ConfigFactory.parseFile(configFile).resolve()
+    } else {
+      println("Loading configuration file connector.conf inside the jar...")
+      ConfigFactory.load("connector.conf").withFallback(ConfigFactory.systemEnvironment())
+    }
+    config
+    // ConfigFactory.load(configFilePath).withFallback(ConfigFactory.load("connector.conf")).withFallback(ConfigFactory.systemEnvironment())
   }
 
   def process(args: Array[String], connectorSource: IConnectorSource)