Feature/185 218 basic auth schema registry (#219)

* 220: Upgrade to ABRiS 4.2.0

* Rename SchemaRegistryConfigKeys -> AbrisConfigKeys

* Refactor AbrisConfigUtil to accept map as schema registry config

* Rename SchemaConfigKeys -> AbrisConfigKeys

* Add support for arbitrary schema registry options

* For now, only print safe properties of the schema registry config

* Fix test

* Update readme

* Fix readme, update attributes

* Revert special printing of config, private members cannot be printed

Author: kevinwallimann

README.md

@@ -141,9 +141,12 @@ The `ConfluentAvroStreamDecodingTransformer` is built on [ABRiS](https://github.
 | `transformer.{transformer-id}.key.schema.record.namespace` | Yes for key naming strategies `record.name` and `topic.record.name` | Namespace of the record. |
 | `transformer.{transformer-id}.keep.columns` | No | Comma-separated list of columns to keep (e.g. offset, partition) |
 | `transformer.{transformer-id}.disable.nullability.preservation` | No | Set to true to ignore fix [#137](https://github.com/AbsaOSS/hyperdrive/issues/137) and to keep the same behaviour as for versions prior to and including v3.2.2. Default value: `false` |
+| `transformer.{transformer-id}.schema.registry.basic.auth.user.info.file` | No | A path to a text file, that contains one line in the form `<username>:<password>`. It will be passed as `basic.auth.user.info` to the schema registry config |
 
 For detailed information on the subject name strategy, please take a look at the [Schema Registry Documentation](https://docs.confluent.io/current/schema-registry/).
 
+Any additional properties for the schema registry config can be added with the prefix `transformer.{transformer-id}.schema.registry.options.`
+
 ##### ConfluentAvroStreamEncodingTransformer
 The `ConfluentAvroStreamEncodingTransformer` is built on [ABRiS](https://github.com/AbsaOSS/ABRiS). More details about the configuration properties can be found there.
 **Caution**: The `ConfluentAvroStreamEncodingTransformer` requires the property `writer.kafka.topic` to be set.
@@ -158,6 +161,9 @@ The `ConfluentAvroStreamEncodingTransformer` is built on [ABRiS](https://github.
 | `transformer.{transformer-id}.key.schema.naming.strategy` | Yes if `produce.keys` is true | Subject name strategy for key |
 | `transformer.{transformer-id}.key.schema.record.name` | Yes for key naming strategies `record.name` and `topic.record.name` | Name of the record. |
 | `transformer.{transformer-id}.key.schema.record.namespace` | Yes for key naming strategies `record.name` and `topic.record.name` | Namespace of the record. |
+| `transformer.{transformer-id}.schema.registry.basic.auth.user.info.file` | No | A path to a text file, that contains one line in the form `<username>:<password>`. It will be passed as `basic.auth.user.info` to the schema registry config |
+
+Any additional properties for the schema registry config can be added with the prefix `transformer.{transformer-id}.schema.registry.options.`
 
 ##### ColumnSelectorStreamTransformer
 | Property Name | Required | Description |

ingestor-default/src/main/scala/za/co/absa/hyperdrive/ingestor/implementation/transformer/avro/confluent/ConfluentAvroDecodingTransformer.scala

@@ -16,7 +16,6 @@
 package za.co.absa.hyperdrive.ingestor.implementation.transformer.avro.confluent
 
 import java.util.UUID
-
 import org.apache.commons.configuration2.Configuration
 import org.apache.commons.lang3.RandomStringUtils
 import org.apache.logging.log4j.LogManager
@@ -28,9 +27,10 @@ import za.co.absa.abris.config.FromAvroConfig
 import za.co.absa.hyperdrive.ingestor.api.context.HyperdriveContext
 import za.co.absa.hyperdrive.ingestor.api.transformer.{StreamTransformer, StreamTransformerFactory}
 import za.co.absa.hyperdrive.ingestor.api.utils.ConfigUtils
+import za.co.absa.hyperdrive.ingestor.api.utils.ConfigUtils.getOrThrow
 import za.co.absa.hyperdrive.ingestor.implementation.HyperdriveContextKeys
 import za.co.absa.hyperdrive.ingestor.implementation.reader.kafka.KafkaStreamReader.KEY_TOPIC
-import za.co.absa.hyperdrive.ingestor.implementation.utils.{AbrisConfigUtil, SchemaRegistryConsumerConfigKeys}
+import za.co.absa.hyperdrive.ingestor.implementation.utils.{AbrisConfigKeys, AbrisConfigUtil, AbrisConsumerConfigKeys, SchemaRegistryConfigUtil}
 
 private[transformer] class ConfluentAvroDecodingTransformer(
   val valueAvroConfig: FromAvroConfig,
@@ -40,11 +40,7 @@ private[transformer] class ConfluentAvroDecodingTransformer(
 )
   extends StreamTransformer {
 
-  private val logger = LogManager.getLogger
-
   override def transform(dataFrame: DataFrame): DataFrame = {
-    logger.info(s"SchemaRegistry settings: $valueAvroConfig")
-
     keyAvroConfigOpt match {
       case Some(keyAvroConfig) => getKeyValueDataFrame(dataFrame, keyAvroConfig)
       case None => getValueDataFrame(dataFrame)
@@ -111,21 +107,21 @@ private[transformer] class ConfluentAvroDecodingTransformer(
 object ConfluentAvroDecodingTransformer extends StreamTransformerFactory with ConfluentAvroDecodingTransformerAttributes {
   private val keyColumnPrefixLength = 4
 
-  object SchemaConfigKeys extends SchemaRegistryConsumerConfigKeys {
+  object AbrisConfigKeys extends AbrisConsumerConfigKeys {
     override val topic: String = KEY_TOPIC
-    override val schemaRegistryUrl: String = KEY_SCHEMA_REGISTRY_URL
     override val schemaId: String = KEY_SCHEMA_REGISTRY_VALUE_SCHEMA_ID
     override val namingStrategy: String = KEY_SCHEMA_REGISTRY_VALUE_NAMING_STRATEGY
     override val recordName: String = KEY_SCHEMA_REGISTRY_VALUE_RECORD_NAME
     override val recordNamespace: String = KEY_SCHEMA_REGISTRY_VALUE_RECORD_NAMESPACE
   }
 
   override def apply(config: Configuration): StreamTransformer = {
-    val valueAvroConfig = AbrisConfigUtil.getValueConsumerSettings(config, SchemaConfigKeys)
+    val schemaRegistryConfig = SchemaRegistryConfigUtil.getSchemaRegistryConfig(config)
+    val valueAvroConfig = AbrisConfigUtil.getValueConsumerSettings(config, AbrisConfigKeys, schemaRegistryConfig)
 
     val consumeKeys = ConfigUtils.getOptionalBoolean(KEY_CONSUME_KEYS, config).getOrElse(false)
     val keyAvroConfigOpt = if (consumeKeys) {
-      Some(AbrisConfigUtil.getKeyConsumerSettings(config, SchemaConfigKeys))
+      Some(AbrisConfigUtil.getKeyConsumerSettings(config, AbrisConfigKeys, schemaRegistryConfig))
     } else {
       None
     }

ingestor-default/src/main/scala/za/co/absa/hyperdrive/ingestor/implementation/transformer/avro/confluent/ConfluentAvroDecodingTransformerAttributes.scala

@@ -17,8 +17,7 @@ package za.co.absa.hyperdrive.ingestor.implementation.transformer.avro.confluent
 
 import za.co.absa.hyperdrive.ingestor.api.{HasComponentAttributes, PropertyMetadata}
 
-trait ConfluentAvroDecodingTransformerAttributes extends HasComponentAttributes {
-  val KEY_SCHEMA_REGISTRY_URL = "schema.registry.url"
+trait ConfluentAvroDecodingTransformerAttributes extends HasComponentAttributes with SchemaRegistryAttributes {
 
   val KEY_SCHEMA_REGISTRY_VALUE_NAMING_STRATEGY = "value.schema.naming.strategy"
   val KEY_SCHEMA_REGISTRY_VALUE_SCHEMA_ID = "value.schema.id"
@@ -34,6 +33,7 @@ trait ConfluentAvroDecodingTransformerAttributes extends HasComponentAttributes
   val KEY_KEEP_COLUMNS = "keep.columns"
   val KEY_DISABLE_NULLABILITY_PRESERVATION = "disable.nullability.preservation"
 
+
   override def getName: String = "Confluent Avro Stream Decoder"
 
   override def getDescription: String = "Decoder for records in Avro format. The decoder connects to a Schema Registry instance to retrieve the schema information."
@@ -54,6 +54,9 @@ trait ConfluentAvroDecodingTransformerAttributes extends HasComponentAttributes
     KEY_SCHEMA_REGISTRY_KEY_RECORD_NAME -> PropertyMetadata("Key-Record name", Some("Key-Record name for naming strategies record.name or topic.record.name"), required = false),
     KEY_SCHEMA_REGISTRY_KEY_RECORD_NAMESPACE -> PropertyMetadata("Key-Record namespace", Some("Key-Record namespace for naming strategies record.name or topic.record.name"), required = false),
     KEY_KEEP_COLUMNS -> PropertyMetadata("Columns to keep", Some("Comma-separated list of columns to keep (e.g. offset, partition)"), required = false),
-    KEY_DISABLE_NULLABILITY_PRESERVATION -> PropertyMetadata("Disable nullability preservation", Some("Keep same behaviour as for versions prior to and including v3.2.2"), required = false)
+    KEY_DISABLE_NULLABILITY_PRESERVATION -> PropertyMetadata("Disable nullability preservation", Some("Keep same behaviour as for versions prior to and including v3.2.2"), required = false),
+    KEY_SCHEMA_REGISTRY_BASIC_AUTH_USER_INFO_FILE -> PropertyMetadata("Basic auth user info file", Some("Text file containing one line in the form <username>:<password> for basic auth in schema registry"), required = false)
   )
+
+  override def getExtraConfigurationPrefix: Option[String] = Some(KEY_SCHEMA_REGISTRY_EXTRA_CONFS_ROOT)
 }

ingestor-default/src/main/scala/za/co/absa/hyperdrive/ingestor/implementation/transformer/avro/confluent/ConfluentAvroEncodingTransformer.scala

@@ -27,7 +27,7 @@ import za.co.absa.hyperdrive.ingestor.api.transformer.{StreamTransformer, Stream
 import za.co.absa.hyperdrive.ingestor.api.utils.ConfigUtils
 import za.co.absa.hyperdrive.ingestor.implementation.HyperdriveContextKeys
 import za.co.absa.hyperdrive.ingestor.implementation.transformer.avro.confluent.ConfluentAvroEncodingTransformer.{getKeyAvroConfig, getValueAvroConfig}
-import za.co.absa.hyperdrive.ingestor.implementation.utils.{AbrisConfigUtil, SchemaRegistryProducerConfigKeys}
+import za.co.absa.hyperdrive.ingestor.implementation.utils.{AbrisConfigUtil, AbrisProducerConfigKeys, SchemaRegistryConfigUtil}
 import za.co.absa.hyperdrive.ingestor.implementation.writer.kafka.KafkaStreamWriter.KEY_TOPIC
 
 private[transformer] class ConfluentAvroEncodingTransformer(
@@ -75,9 +75,8 @@ private[transformer] class ConfluentAvroEncodingTransformer(
 
 object ConfluentAvroEncodingTransformer extends StreamTransformerFactory with ConfluentAvroEncodingTransformerAttributes {
 
-  object SchemaConfigKeys extends SchemaRegistryProducerConfigKeys {
+  object AbrisConfigKeys extends AbrisProducerConfigKeys {
     override val topic: String = KEY_TOPIC
-    override val schemaRegistryUrl: String = KEY_SCHEMA_REGISTRY_URL
     override val namingStrategy: String = KEY_SCHEMA_REGISTRY_VALUE_NAMING_STRATEGY
     override val recordName: String = KEY_SCHEMA_REGISTRY_VALUE_RECORD_NAME
     override val recordNamespace: String = KEY_SCHEMA_REGISTRY_VALUE_RECORD_NAMESPACE
@@ -92,11 +91,15 @@ object ConfluentAvroEncodingTransformer extends StreamTransformerFactory with Co
     KEY_TOPIC -> KEY_TOPIC
   )
 
-  def getKeyAvroConfig(config: Configuration, expression: Expression): ToAvroConfig =
-    AbrisConfigUtil.getKeyProducerSettings(config, SchemaConfigKeys, expression)
+  def getKeyAvroConfig(config: Configuration, expression: Expression): ToAvroConfig = {
+    val schemaRegistryConfig = SchemaRegistryConfigUtil.getSchemaRegistryConfig(config)
+    AbrisConfigUtil.getKeyProducerSettings(config, AbrisConfigKeys, expression, schemaRegistryConfig)
+  }
 
-  def getValueAvroConfig(config: Configuration, expression: Expression): ToAvroConfig =
-    AbrisConfigUtil.getValueProducerSettings(config, SchemaConfigKeys, expression)
+  def getValueAvroConfig(config: Configuration, expression: Expression): ToAvroConfig = {
+    val schemaRegistryConfig = SchemaRegistryConfigUtil.getSchemaRegistryConfig(config)
+    AbrisConfigUtil.getValueProducerSettings(config, AbrisConfigKeys, expression, schemaRegistryConfig)
+  }
 }
 
 

ingestor-default/src/main/scala/za/co/absa/hyperdrive/ingestor/implementation/transformer/avro/confluent/ConfluentAvroEncodingTransformerAttributes.scala

@@ -17,8 +17,7 @@ package za.co.absa.hyperdrive.ingestor.implementation.transformer.avro.confluent
 
 import za.co.absa.hyperdrive.ingestor.api.{HasComponentAttributes, PropertyMetadata}
 
-trait ConfluentAvroEncodingTransformerAttributes extends HasComponentAttributes {
-  val KEY_SCHEMA_REGISTRY_URL = "schema.registry.url"
+trait ConfluentAvroEncodingTransformerAttributes extends HasComponentAttributes with SchemaRegistryAttributes {
   val KEY_SCHEMA_REGISTRY_VALUE_NAMING_STRATEGY = "value.schema.naming.strategy"
   val KEY_SCHEMA_REGISTRY_VALUE_RECORD_NAME = "value.schema.record.name"
   val KEY_SCHEMA_REGISTRY_VALUE_RECORD_NAMESPACE = "value.schema.record.namespace"
@@ -45,6 +44,9 @@ trait ConfluentAvroEncodingTransformerAttributes extends HasComponentAttributes
     KEY_SCHEMA_REGISTRY_KEY_NAMING_STRATEGY -> PropertyMetadata("Key-Schema naming strategy",
       Some("Subject name strategy of Schema Registry. Must be one of \"topic.name\", \"record.name\" or \"topic.record.name\""), required = false),
     KEY_SCHEMA_REGISTRY_KEY_RECORD_NAME -> PropertyMetadata("Key-Record name", Some("Key-Record name for naming strategies record.name or topic.record.name"), required = false),
-    KEY_SCHEMA_REGISTRY_KEY_RECORD_NAMESPACE -> PropertyMetadata("Key-Record namespace", Some("Key-Record namespace for naming strategies record.name or topic.record.name"), required = false)
+    KEY_SCHEMA_REGISTRY_KEY_RECORD_NAMESPACE -> PropertyMetadata("Key-Record namespace", Some("Key-Record namespace for naming strategies record.name or topic.record.name"), required = false),
+    KEY_SCHEMA_REGISTRY_BASIC_AUTH_USER_INFO_FILE -> PropertyMetadata("Basic auth user info file", Some("Text file containing one line in the form <username>:<password> for basic auth in schema registry"), required = false)
   )
+
+  override def getExtraConfigurationPrefix: Option[String] = Some(KEY_SCHEMA_REGISTRY_EXTRA_CONFS_ROOT)
 }

ingestor-default/src/main/scala/za/co/absa/hyperdrive/ingestor/implementation/transformer/avro/confluent/SchemaRegistryAttributes.scala

@@ -0,0 +1,25 @@
+
+/*
+ * Copyright 2018 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.hyperdrive.ingestor.implementation.transformer.avro.confluent
+
+trait SchemaRegistryAttributes {
+  val KEY_SCHEMA_REGISTRY_URL = "schema.registry.url"
+  val KEY_SCHEMA_REGISTRY_BASIC_AUTH_USER_INFO_FILE = "schema.registry.basic.auth.user.info.file"
+  val KEY_SCHEMA_REGISTRY_EXTRA_CONFS_ROOT = "schema.registry.options"
+}
+
+object SchemaRegistryAttributes extends SchemaRegistryAttributes

…ion/utils/SchemaRegistryConfigKeys.scala …plementation/utils/AbrisConfigKeys.scalaingestor-default/src/main/scala/za/co/absa/hyperdrive/ingestor/implementation/utils/SchemaRegistryConfigKeys.scala renamed to ingestor-default/src/main/scala/za/co/absa/hyperdrive/ingestor/implementation/utils/AbrisConfigKeys.scala ingestor-default/src/main/scala/za/co/absa/hyperdrive/ingestor/implementation/utils/SchemaRegistryConfigKeys.scala renamed to ingestor-default/src/main/scala/za/co/absa/hyperdrive/ingestor/implementation/utils/AbrisConfigKeys.scala

@@ -15,16 +15,15 @@
 
 package za.co.absa.hyperdrive.ingestor.implementation.utils
 
-private[hyperdrive] trait SchemaRegistryConfigKeys {
+private[hyperdrive] trait AbrisConfigKeys {
   val topic: String
-  val schemaRegistryUrl: String
   val namingStrategy: String
   val recordName: String
   val recordNamespace: String
 }
 
-private[hyperdrive] trait SchemaRegistryConsumerConfigKeys extends SchemaRegistryConfigKeys {
+private[hyperdrive] trait AbrisConsumerConfigKeys extends AbrisConfigKeys {
   val schemaId: String
 }
 
-private[hyperdrive] trait SchemaRegistryProducerConfigKeys extends SchemaRegistryConfigKeys
+private[hyperdrive] trait AbrisProducerConfigKeys extends AbrisConfigKeys