Repository: AbsaOSS/pramen (language: python)

Issue:
JDBC source fails on a strange error when information date sql type is not correct
## Describe the bug
A clear and concise description of what the bug is.

## Code and/or configuration snippet that caused the issue

```hocon
{
  name = "jdbc3"
  factory.class = "za.co.absa.pramen.core.source.JdbcSource"
  jdbc {
    driver = "driver2"
    connection.string = "url2"
    user = "user2"
    password = "password2"
  }

  has.information.date.column = false
  information.date.type = "wrong"
}
```

Stack trace
```
Class 'za.co.absa.pramen.core.source.JdbcSource' is not an instance of 'za.co.absa.pramen.api.ExternalChannelFactory[za.co.absa.pramen.api.Source]'
java.lang.IllegalArgumentException: Class 'za.co.absa.pramen.core.source.JdbcSource' is not an instance of 'za.co.absa.pramen.api.ExternalChannelFactory[za.co.absa.pramen.api.Source]'
	at za.co.absa.pramen.core.utils.ClassLoaderUtils$.loadSingletonClassOfType(ClassLoaderUtils.scala:37)
	at za.co.absa.pramen.core.ExternalChannelFactoryReflect$.fromConfig(ExternalChannelFactoryReflect.scala:49)
	at za.co.absa.pramen.core.ExternalChannelFactoryReflect$.fromConfigByName(ExternalChannelFactoryReflect.scala:105)
	at za.co.absa.pramen.core.source.JdbcSourceSuite.$anonfun$new$4(JdbcSourceSuite.scala:137)
```

## Expected behavior
1. `information.date.type` should be ignored when `has.information.date.column = false`
2. If `has.information.date.column = true` the error should reflect the actual error.

## Context
- Pramen/pramen-py version: 1.10.0
- Spark version: 3.4.4
- Scala/Python version: 2.21



Discussion / hints:

Fix patch:
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/ExternalChannelFactoryReflect.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/ExternalChannelFactoryReflect.scala
index 551f4271a..eb07e99d9 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/ExternalChannelFactoryReflect.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/ExternalChannelFactoryReflect.scala
@@ -41,13 +41,18 @@ object ExternalChannelFactoryReflect {
       throw new IllegalArgumentException(s"A class should be specified for the $channelType at '$parentPath'.")
     }
     val clazz = conf.getString(FACTORY_CLASS_KEY)
-    try {
-      val factory = ClassLoaderUtils.loadSingletonClassOfType[ExternalChannelFactoryV2[T]](clazz)
-      factory.apply(conf, workflowConfig, parentPath, spark)
+    val factory = try {
+      ClassLoaderUtils.loadSingletonClassOfType[ExternalChannelFactoryV2[T]](clazz)
     } catch {
       case _: Throwable =>
-        val factory = ClassLoaderUtils.loadSingletonClassOfType[ExternalChannelFactory[T]](clazz)
-        factory.apply(conf, parentPath, spark)
+        ClassLoaderUtils.loadSingletonClassOfType[ExternalChannelFactory[T]](clazz)
+    }
+
+    factory match {
+      case fv2: ExternalChannelFactoryV2[T] =>
+        fv2.apply(conf, workflowConfig, parentPath, spark)
+      case fv1: ExternalChannelFactory[T] =>
+        fv1.apply(conf, parentPath, spark)
     }
   }
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/model/TableReaderJdbcConfig.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/model/TableReaderJdbcConfig.scala
index 3ecffd2b0..894ab8a69 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/model/TableReaderJdbcConfig.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/model/TableReaderJdbcConfig.scala
@@ -75,7 +75,11 @@ object TableReaderJdbcConfig {
 
     val infoDateTypeStr = ConfigUtils.getOptionString(conf, INFORMATION_DATE_TYPE).getOrElse("date")
 
-    val infoDateType = SqlColumnType.fromStringStrict(infoDateTypeStr, parent)
+    val infoDateType = if (hasInformationDate) {
+      SqlColumnType.fromStringStrict(infoDateTypeStr, parent)
+    } else {
+      SqlColumnType.DATE
+    }
 
     val saveTimestampsAsDates = ConfigUtils.getOptionBoolean(conf, JDBC_TIMESTAMPS_AS_DATES).getOrElse(false)
 


Test patch:
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/ExternalChannelFactorySuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/ExternalChannelFactorySuite.scala
index 33ad9aa56..dde67667c 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/ExternalChannelFactorySuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/ExternalChannelFactorySuite.scala
@@ -70,6 +70,42 @@ class ExternalChannelFactorySuite extends AnyWordSpec with SparkTestBase {
 
       assert(ex.getMessage.contains("A class should be specified for the dummy"))
     }
+
+    "re-throw an exception from the V1 factory call" in {
+      val workflowConf = ConfigFactory.parseString(
+        """test = "test"
+          |""".stripMargin)
+
+
+      val conf = ConfigFactory.parseString(
+        """factory.class = "za.co.absa.pramen.core.mocks.ExternalChannelV2Mock"
+          |fail = true
+          |""".stripMargin)
+
+      val ex = intercept[RuntimeException] {
+        ExternalChannelFactoryReflect.fromConfig[ExternalChannelV2Mock](conf, workflowConf, "", "dummy")
+      }
+
+      assert(ex.getMessage == "Test exception")
+    }
+
+    "re-throw an exception from the V2 factory call" in {
+      val workflowConf = ConfigFactory.parseString(
+        """test = "test"
+          |""".stripMargin)
+
+
+      val conf = ConfigFactory.parseString(
+        """factory.class = "za.co.absa.pramen.core.mocks.ExternalChannelMock"
+          |fail = true
+          |""".stripMargin)
+
+      val ex = intercept[RuntimeException] {
+        ExternalChannelFactoryReflect.fromConfig[ExternalChannelMock](conf, workflowConf, "", "dummy")
+      }
+
+      assert(ex.getMessage == "Test exception")
+    }
   }
 
   "fromConfigByName" should {
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/ExternalChannelMock.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/ExternalChannelMock.scala
index 3d63a58f7..9b383a811 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/ExternalChannelMock.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/ExternalChannelMock.scala
@@ -25,10 +25,12 @@ class ExternalChannelMock(conf: Config, val value1: String, val value2: String)
 }
 
 object ExternalChannelMock extends ExternalChannelFactory[ExternalChannelMock] {
+  val CONFIG_FAIL_KEY = "fail"
   val CONFIG_KEY1 = "key1"
   val CONFIG_KEY2 = "key2"
 
   override def apply(conf: Config, parentPath: String, spark: SparkSession): ExternalChannelMock = {
+    if (conf.hasPath(CONFIG_FAIL_KEY)) throw new RuntimeException("Test exception")
     val value1 = conf.getString(CONFIG_KEY1)
     val value2 = conf.getString(CONFIG_KEY2)
     new ExternalChannelMock(conf, value1, value2)
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/ExternalChannelV2Mock.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/ExternalChannelV2Mock.scala
index 43e4d8f3b..f95e442bf 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/ExternalChannelV2Mock.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/ExternalChannelV2Mock.scala
@@ -25,12 +25,15 @@ class ExternalChannelV2Mock(conf: Config, val workflowConfig: Config, val value1
 }
 
 object ExternalChannelV2Mock extends ExternalChannelFactoryV2[ExternalChannelV2Mock] {
+  val CONFIG_FAIL_KEY = "fail"
   val CONFIG_KEY1 = "key1"
   val CONFIG_KEY2 = "key2"
 
   override def apply(conf: Config, workflowConfig: Config, parentPath: String, spark: SparkSession): ExternalChannelV2Mock = {
+    if (conf.hasPath(CONFIG_FAIL_KEY)) throw new RuntimeException("Test exception")
     val value1 = conf.getString(CONFIG_KEY1)
     val value2 = conf.getString(CONFIG_KEY2)
+
     new ExternalChannelV2Mock(conf, workflowConfig, value1, value2)
   }
 }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/source/JdbcSourceSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/source/JdbcSourceSuite.scala
index 955c0e768..06d7dba09 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/source/JdbcSourceSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/source/JdbcSourceSuite.scala
@@ -85,6 +85,35 @@ class JdbcSourceSuite extends AnyWordSpec with BeforeAndAfterAll with SparkTestB
        |      information.date.type = "date"
        |      information.date.app.format = "yyyy-MM-DD"
        |      information.date.sql.format = "YYYY-mm-DD"
+       |    },
+       |    {
+       |      name = "jdbc3"
+       |      factory.class = "za.co.absa.pramen.core.source.JdbcSource"
+       |      jdbc {
+       |        driver = "driver3"
+       |        connection.string = "url3"
+       |        user = "user3"
+       |        password = "password3"
+       |      }
+       |
+       |      has.information.date.column = false
+       |      information.date.type = "wrong"
+       |    },
+       |    {
+       |      name = "jdbc4"
+       |      factory.class = "za.co.absa.pramen.core.source.JdbcSource"
+       |      jdbc {
+       |        driver = "driver4"
+       |        connection.string = "url4"
+       |        user = "user4"
+       |        password = "password4"
+       |      }
+       |
+       |      has.information.date.column = true
+       |      information.date.type = "wrong"
+       |      information.date.column = "INFO_DATE"
+       |      information.date.app.format = "yyyy-MM-DD"
+       |      information.date.sql.format = "YYYY-mm-DD"
        |    }
        |  ]
        | }
@@ -116,6 +145,24 @@ class JdbcSourceSuite extends AnyWordSpec with BeforeAndAfterAll with SparkTestB
       assert(fetchedConfig.hasPath(DISABLE_COUNT_QUERY))
     }
 
+    "be able to get a source if the source is snapshot and info date column type is wrong" in {
+      val src = ExternalChannelFactoryReflect.fromConfigByName[Source](conf, None, "pramen.sources", "Jdbc3", "source").asInstanceOf[JdbcSource]
+      val (fetchedConfig, index) = ExternalChannelFactoryReflect.getConfigByName(conf, None, "pramen.sources", "jdbc2", "source")
+
+      assert(src.jdbcReaderConfig.jdbcConfig.driver == "driver3")
+      assert(!src.jdbcReaderConfig.hasInfoDate)
+      assert(index == 1)
+      assert(fetchedConfig.hasPath(DISABLE_COUNT_QUERY))
+    }
+
+    "throw a proper exception when info date column type is wrong" in {
+      val ex = intercept[IllegalArgumentException] {
+        ExternalChannelFactoryReflect.fromConfigByName[Source](conf, None, "pramen.sources", "Jdbc4", "source").asInstanceOf[JdbcSource]
+      }
+
+      assert(ex.getMessage.contains("nknown information type 'wrong'"))
+    }
+
     "be able to get a source by its name from the manager" in {
       val src = SourceManager.getSourceByName("Jdbc2", conf, None).asInstanceOf[JdbcSource]
 
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/reader/TableReaderJdbcSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/reader/TableReaderJdbcSuite.scala
index fd5b0a282..381a092a9 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/reader/TableReaderJdbcSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/reader/TableReaderJdbcSuite.scala
@@ -59,8 +59,6 @@ class TableReaderJdbcSuite extends AnyWordSpec with BeforeAndAfterAll with Spark
          |  has.information.date.column = false
          |
          |  information.date.column = "INFO_DATE"
-         |  information.date.type = "number"
-         |  information.date.format = "yyyy-MM-DD"
          |
          |  offset.column {
          |    name = "ts"
@@ -95,7 +93,7 @@ class TableReaderJdbcSuite extends AnyWordSpec with BeforeAndAfterAll with Spark
          |  has.information.date.column = true
          |
          |  information.date.column = "INFO_DATE"
-         |  information.date.type = "date"
+         |  information.date.type = "number"
          |}""".stripMargin)
 
     "be able to be constructed properly from config" in {
@@ -108,9 +106,6 @@ class TableReaderJdbcSuite extends AnyWordSpec with BeforeAndAfterAll with Spark
       assert(jdbc.jdbcConfig.primaryUrl.get == url)
       assert(jdbc.jdbcConfig.user.contains(user))
       assert(jdbc.jdbcConfig.password.contains(password))
-      assert(jdbc.infoDateColumn == "INFO_DATE")
-      assert(jdbc.infoDateType == SqlColumnType.NUMBER)
-      assert(jdbc.infoDateFormat == "yyyy-MM-DD")
       assert(jdbc.offsetInfoOpt.nonEmpty)
       assert(jdbc.offsetInfoOpt.get.offsetColumn == "ts")
       assert(jdbc.offsetInfoOpt.get.offsetType.dataTypeString == "datetime")
@@ -150,7 +145,7 @@ class TableReaderJdbcSuite extends AnyWordSpec with BeforeAndAfterAll with Spark
       assert(jdbc.jdbcConfig.user.contains(user))
       assert(jdbc.jdbcConfig.password.contains(password))
       assert(jdbc.infoDateColumn == "INFO_DATE")
-      assert(jdbc.infoDateType == SqlColumnType.DATE)
+      assert(jdbc.infoDateType == SqlColumnType.NUMBER)
       assert(jdbc.infoDateFormat == "yyyy-MM-dd")
       assert(jdbc.offsetInfoOpt.isEmpty)
       assert(jdbc.hasInfoDate)
@@ -202,6 +197,27 @@ class TableReaderJdbcSuite extends AnyWordSpec with BeforeAndAfterAll with Spark
       assert(!jdbc.hasInfoDate)
     }
 
+    "ensure jdbc minimal snapshot configuration works even if wrong format for the info date is specified" in {
+      val testConfig = ConfigFactory.parseString(
+        s"""reader {
+           |  jdbc {
+           |    driver = "$driver"
+           |    connection.string = "$url"
+           |    user = "$user"
+           |    password = "$password"
+           |  }
+           |
+           |  has.information.date.column = false
+           |  information.date.column = "INFO_DATE"
+           |  information.date.type = "wrong"
+           |}""".stripMargin)
+      val reader = TableReaderJdbc(testConfig.getConfig("reader"), testConfig.getConfig("reader"), "reader")
+
+      val jdbc = reader.getJdbcConfig
+
+      assert(!jdbc.hasInfoDate)
+    }
+
     "ensure jdbc minimal event configuration works" in {
       val testConfig = ConfigFactory.parseString(
         s"""reader {