#649 Fix PR suggestions - thanks @coderabbitai.

Author: yruslan

README.md

@@ -945,9 +945,9 @@ pramen.sources = [
     # [Optional] Set name for the Kafka key column
     #key.column.name = "kafka_key"
     
-    # The Kafka key serializer. Can be "none", "binary", "string", "avro".
-    # When "avro", "key.naming.strategy" should be deined at the "schema.registry" section.
-    # Default is "binary", but if "key.naming.strategy" is defined, "avro" is selected automatically.
+    # The Kafka key serializer when key.naming.strategy is NOT defined. Can be "none", "binary", "string".
+    # When key.naming.strategy IS defined in schema.registry, Avro deserialization is used automatically.
+    # Default is "binary".
     #key.column.serializer = "none"
 
     kafka {

pramen/core/src/main/scala/za/co/absa/pramen/core/model/QueryBuilder.scala

@@ -53,9 +53,9 @@ object QueryBuilder {
       case _                           =>
         val parent = if (parentPath.isEmpty) "" else s" at $parentPath"
         if (prefix.isEmpty)
-          throw new IllegalArgumentException(s"No options are specified for the query. Usually, it is one of: '$SQL_KEY', '$PATH_KEY', '$TABLE_KEY', '$DB_TABLE_KEY'$parent.")
+          throw new IllegalArgumentException(s"No options are specified for the query. Usually, it is one of: '$SQL_KEY', '$PATH_KEY', '$TABLE_KEY', '$DB_TABLE_KEY', '$TOPIC_KEY'$parent.")
         else
-          throw new IllegalArgumentException(s"No options are specified for the '$prefix' query. Usually, it is one of: '$p$SQL_KEY', '$p$PATH_KEY', '$p$TABLE_KEY', '$p$DB_TABLE_KEY'$parent.")
+          throw new IllegalArgumentException(s"No options are specified for the '$prefix' query. Usually, it is one of: '$p$SQL_KEY', '$p$PATH_KEY', '$p$TABLE_KEY', '$p$DB_TABLE_KEY', '$p$TOPIC_KEY'$parent.")
     }
   }
 }

pramen/core/src/test/scala/za/co/absa/pramen/core/model/QueryBuilderSuite.scala

@@ -55,7 +55,7 @@ class QueryBuilderSuite extends AnyWordSpec {
         QueryBuilder.fromConfig(conf, "", "")
       }
 
-      assert(exception.getMessage == "No options are specified for the query. Usually, it is one of: 'sql', 'path', 'table', 'db.table'.")
+      assert(exception.getMessage == "No options are specified for the query. Usually, it is one of: 'sql', 'path', 'table', 'db.table', 'topic'.")
     }
 
     "throw an exception when the prefix is empty" in {
@@ -65,7 +65,7 @@ class QueryBuilderSuite extends AnyWordSpec {
         QueryBuilder.fromConfig(conf, "input", "")
       }
 
-      assert(exception.getMessage == "No options are specified for the 'input' query. Usually, it is one of: 'input.sql', 'input.path', 'input.table', 'input.db.table'.")
+      assert(exception.getMessage == "No options are specified for the 'input' query. Usually, it is one of: 'input.sql', 'input.path', 'input.table', 'input.db.table', 'input.topic'.")
     }
 
     "throw an exception when the prefix is empty and parent is specified" in {
@@ -75,7 +75,7 @@ class QueryBuilderSuite extends AnyWordSpec {
         QueryBuilder.fromConfig(conf, "input", "my.parent")
       }
 
-      assert(exception.getMessage == "No options are specified for the 'input' query. Usually, it is one of: 'input.sql', 'input.path', 'input.table', 'input.db.table' at my.parent.")
+      assert(exception.getMessage == "No options are specified for the 'input' query. Usually, it is one of: 'input.sql', 'input.path', 'input.table', 'input.db.table', 'input.topic' at my.parent.")
     }
   }
 }

pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/SourceTableParserSuite.scala

@@ -119,7 +119,7 @@ class SourceTableParserSuite extends AnyWordSpec {
         SourceTableParser.fromConfig(conf, "source.tables")
       }
 
-      assert(ex.getMessage.contains("No options are specified for the 'input' query. Usually, it is one of: 'input.sql', 'input.path', 'input.table', 'input.db.table' at source.tables[0]."))
+      assert(ex.getMessage.contains("No options are specified for the 'input' query. Usually, it is one of: 'input.sql', 'input.path', 'input.table', 'input.db.table', 'input.topic' at source.tables[0]."))
     }
 
     "throw an exception in case of duplicate entries" in {

pramen/extras/src/main/scala/za/co/absa/pramen/extras/source/KafkaAvroSource.scala

@@ -53,9 +53,9 @@ import java.time.LocalDate
   *    # [Optional] Set name for the Kafka key column
   *    key.column.name = "kafka_key"
   *
-  *    # The Kafka key serializer. Can be "none", "binary", "string", "avro".
-  *    # When "avro", "key.naming.strategy" should be deined at the "schema.registry" section.
-  *    # Default is "binary", but if "key.naming.strategy" is defined, "avro" is selected automatically.
+  *    # The Kafka key serializer when 'key.naming.strategy' is NOT defined. Can be "none", "binary", "string".
+  *    # When 'key.naming.strategy' IS defined in 'schema.registry', Avro deserialization is used automatically.
+  *    # Default is "binary".
   *    #key.column.serializer = "none"
   *
   *    kafka {
@@ -217,7 +217,7 @@ class KafkaAvroSource(sourceConfig: Config,
         col("timestampType").as("timestamp_type")
       ))
 
-    val hasKey = keyColumnSerializer != "none"
+    val hasKey = kafkaAvroConfig.keyNamingStrategy.isDefined || keyColumnSerializer != "none"
 
     val df2 = kafkaAvroConfig.keyNamingStrategy match {
       case Some(keyNamingStrategy) =>
@@ -231,16 +231,16 @@ class KafkaAvroSource(sourceConfig: Config,
           case "binary" => df1.withColumn(tempKafkaKeyColumnName, col("key"))
           case "string" => df1.withColumn(tempKafkaKeyColumnName, col("key").cast(StringType))
           case "avro" => throw new IllegalArgumentException("For the 'avro' serializer of Kafka topic key, 'schema.registry.key.naming.strategy' needs to be set.")
-          case x => throw new IllegalArgumentException("Unknown Kafka key serializer. Can be one of: none, binary, long, string, avro.")
+          case x => throw new IllegalArgumentException(s"Unknown Kafka key serializer '$x'. Can be one of: none, binary, string, avro.")
         }
     }
 
     val payloadFields = df2.select("data.*").schema.fieldNames.toSet
     if (payloadFields.contains(kafkaColumnName)) {
-      log.warn(s"Payload field '$kafkaColumnName' conflicts with Kafka metadata struct name and will be replaced.")
+      log.warn(s"Payload field '$kafkaColumnName' conflicts with reserved Kafka metadata struct name and will be replaced.")
     }
     if (payloadFields.contains(keyColumnName)) {
-      log.warn(s"Payload field '$keyColumnName' conflicts with Kafka key column name and will be replaced.")
+      log.warn(s"Payload field '$keyColumnName' conflicts with reserved Kafka key column name and will be removed.")
     }
 
     // Put data fields to the root level of the schema, and if data struct already has kafka_key and kafka fields,