Repository: ImFlog/schema-registry-plugin
Pull request #167 (2024-02-01 23:07:45)

Adding subject metadata and ruleset capabilities
Adding support for subject metadata, rulesets and schema normalisation.

This is also supported in the maven plugin. 

What I did is a very rudimentary implementation, if you have any ideas. I am happy to have a look and modify the code. 

Referencing a file, is an easy way, else I am afraid that the Gradle extension will become to complicated. 

Also my Kotlin experience is limited. 

StephanePaulus: I did the code changes you asked for and added some quick tests. I will look into adding an integration test, example and readme for the changes. 

About the downloading of the schema, this data will already be returned when also downloading the optional metadata. 

    download {
        // Optional
        metadata = new MetadataExtension(true, "path/to/metadata/")

        // extension of the output file depends on the the schema type
        subject('avroSubject', '/absolutPath/src/main/avro')
        subject('protoSubject', 'src/main/proto')
        subject('jsonSubject', 'src/main/json')

        // You can use a regex to download multiple schemas at once
        subjectPattern('avro.*', 'src/main/avro')
    }


StephanePaulus: After looking into writing integration-test, the ruleSet and metadata are not being downloading when using `confluentinc/cp-schema-registry`, I guess this only works for confluent-cloud. 
StephanePaulus: Hey @ImFlog, do you think the changes I have made are sufficient? 

Patch:
diff --git a/README.md b/README.md
index c1b086c..cf39c40 100644
--- a/README.md
+++ b/README.md
@@ -9,10 +9,15 @@ for Gradle builds.
 
 ## Usage
 
-Follow the [gradle plugin portal instructions](https://plugins.gradle.org/plugin/com.github.imflog.kafka-schema-registry-gradle-plugin).
+The plugin relies on two libraries that are not available on common repositories, you will have to add them to
+your `buildscript`:
 
-<details>
-<summary>For plugin version < 1.12.0</summary>
+* one developed by confluent for which you need to add the `https://packages.confluent.io/maven/`
+* one developed by myself to add new capability to Avro for which you need to add the `https://jitpack.io`
+
+<div class="tabbed-code-block">
+  <details open>
+    <summary>Groovy</summary>
 
 ```groovy
 buildscript {
@@ -30,7 +35,30 @@ plugins {
     id "com.github.imflog.kafka-schema-registry-gradle-plugin" version "X.X.X"
 }
 ```
-</details>
+
+  </details>
+  <details>
+    <summary>Kotlin</summary>
+
+```kotlin
+buildscript {
+    repositories {
+        gradlePluginPortal()
+        maven("https://packages.confluent.io/maven/")
+        maven("https://jitpack.io")
+    }
+}
+plugins {
+    id("com.github.imflog.kafka-schema-registry-gradle-plugin") version "X.X.X"
+}
+```
+
+  </details>
+</div>
+
+Where "X.X.X" is the current version,
+see [gradle plugin portal](https://plugins.gradle.org/plugin/com.github.imflog.kafka-schema-registry-gradle-plugin) for
+details.
 
 ## Tasks
 
@@ -60,8 +88,9 @@ schemaRegistry {
   Could be removed if https://github.com/gradle/gradle/issues/1010 is fixed.
 * `outputDirectory` is the directory where action result will be stored as files (only register for now).
   This is an optional parameter.
-* `pretty` is whether the downloaded Avro or json schemas should be formatted ("pretty-printed") or minified. 
+* `pretty` is whether the downloaded Avro or json schemas should be formatted ("pretty-printed") or minified.
   This is an optional parameter.
+
 ### Download schemas
 
 Like the name of the task imply, this task is responsible for retrieving schemas from a schema registry.
@@ -106,6 +135,7 @@ Here is the list of all the signatures for the `subject` extension:
 You can configure the metadata extension in order to download the schemas metadata in json files.
 It will be saved in files named like the schema file but suffixed by `-metadata.json` in the outputPath you specify
 and defaults to the same output directory as your schemas.
+The metadata extension will also download optional `metadata` and `rulesets` if your schema-registry supports this.  
 
 NB:
 
@@ -204,6 +234,8 @@ schemaRegistry {
                 .addReference('avroSubjectLatestVersionExplicit', 'avroSubjectLatestVersionExplicitType', -1)
         subject('protoWithReferences', 'dependent/path.proto', "PROTOBUF").addReference('protoSubject', 'protoSubjectType', 1)
         subject('jsonWithReferences', 'dependent/path.json', "JSON").addReference('jsonSubject', 'jsonSubjectType', 1)
+                .setMetadata('/absolutPath/dependent/metadata.json').setRuleSet('/absolutPath/dependent/ruleset.json')
+                .setNormalized(true)
     }
 }
 ```
@@ -221,6 +253,13 @@ you can call the `addLocalReference("name", "/a/path")`,
 this will add a reference from a local file and inline it in the schema registry call.
 The addLocalReference calls can be chained.
 
+You can also provide `setNormalized(true)` to normalize the schema. Here is a link for more information: [Schema Normalization](https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#schema-normalization)
+
+For some schema registries like the one offered by `confluent-cloud`, you can add `metadata` and `rulesets` to a subject.
+Here are some links for more information: 
+* [Schema Registry Rules](https://docs.confluent.io/platform/7.5/schema-registry/fundamentals/data-contracts.html#rules)
+* [Schema Registry Metadata](https://docs.confluent.io/platform/7.5/schema-registry/fundamentals/data-contracts.html#metadata-properties)
+
 Notes:
 
 * A registered.csv file will be created with the following format `subject, path, id`
@@ -322,7 +361,7 @@ here: [org.apache.kafka.common.config.SslConfigs](https://github.com/confluentin
 
 ### Examples
 
-Detailed examples can be found in the [examples directory](examples).
+Detailed examples can be found in the [examples' directory](examples).
 
 ## Version compatibility
 
@@ -333,7 +372,7 @@ It was easier to introduce all the changes in one shot instead of supporting bot
 users:
 
 * plugin versions above 1.X.X support the confluent version > 5.5.X (Avro / Json / Protobuf)
-* older plugin versions should support anything below 5.4.X
+* plugin versions should support anything below 5.4.X
 
 ## Developing
 
@@ -372,7 +411,10 @@ buildscript {
         // The new repository to import, you may not want this in your final gradle configuration.
         mavenLocal()
         maven {
-            url "http://packages.confluent.io/maven/"
+            url "https://packages.confluent.io/maven/"
+        }
+        maven {
+          url = "https://jitpack.io"
         }
     }
     dependencies {
@@ -385,4 +427,4 @@ apply plugin: "com.github.imflog.kafka-schema-registry-gradle-plugin"
 
 ## Thanks to all the sponsors :pray:
 
-<a href="https://github.com/marktros"><img style="border-radius: 50%;" src="https://github.com/marktros.png" width="60px" alt="Mark Ethan Trostler" /></a>
\ No newline at end of file
+<a href="https://github.com/marktros"><img style="border-radius: 50%;" src="https://github.com/marktros.png" width="60px" alt="Mark Ethan Trostler" /></a>
diff --git a/build.gradle.kts b/build.gradle.kts
index a099cf3..907ee79 100644
--- a/build.gradle.kts
+++ b/build.gradle.kts
@@ -7,7 +7,6 @@ version = "1.13.1-SNAPSHOT"
 plugins {
     kotlin("jvm") version "1.9.22"
     id("com.gradle.plugin-publish") version "1.2.1"
-    id("com.github.johnrengelman.shadow") version "7.1.2"
 }
 
 repositories {
@@ -17,16 +16,13 @@ repositories {
 }
 
 // Dependencies versions
-val confluentVersion = "7.5.3"
+val confluentVersion = "7.6.0"
 val avroVersion = "1.11.2"
 dependencies {
-    shadow(gradleApi())
-    implementation(platform("io.confluent:kafka-schema-registry-parent:$confluentVersion"))
-    implementation("io.confluent", "kafka-schema-registry") {
+    implementation(gradleApi())
+    implementation("io.confluent", "kafka-schema-registry", confluentVersion) {
         exclude("org.slf4j", "slf4j-log4j12")
     }
-    implementation("io.confluent", "kafka-json-schema-provider")
-    implementation("io.confluent", "kafka-protobuf-provider")
     // Our custom avro version. See https://github.com/ImFlog/avro
     implementation("com.github.ImFlog.avro", "avro", avroVersion)
 }
@@ -40,6 +36,10 @@ tasks.withType<KotlinCompile>().configureEach {
     }
 }
 
+java {
+    withSourcesJar()
+}
+
 // Unit tests
 val junitVersion = "5.7.2"
 val mockkVersion = "1.11.0"
@@ -90,14 +90,6 @@ task<Test>("integrationTest") {
 }
 
 // Publish plugin
-tasks {
-    shadowJar {
-        archiveClassifier.set("")
-        from(sourceSets.main.get().output)
-        from(project.configurations.runtimeClasspath)
-    }
-}
-
 val registryPluginName = "com.github.imflog.kafka-schema-registry-gradle-plugin"
 @Suppress("UnstableApiUsage")
 gradlePlugin {
@@ -118,4 +110,4 @@ gradlePlugin {
             implementationClass = "com.github.imflog.schema.registry.SchemaRegistryPlugin"
         }
     }
-}
\ No newline at end of file
+}
diff --git a/examples/avro-kts/build.gradle.kts b/examples/avro-kts/build.gradle.kts
index 811aff9..670a2bd 100644
--- a/examples/avro-kts/build.gradle.kts
+++ b/examples/avro-kts/build.gradle.kts
@@ -17,6 +17,10 @@ schemaRegistry {
                 .addReference("company", "company", -1)
         subject("location-address", "schemas/avro/location-address.avsc", "AVRO")
         subject("location-latlong", "schemas/avro/location-latlong.avsc", "AVRO")
+        subject("my-record","schemas/avro/my-record.avsc", "AVRO")
+            .setMetadata("extra/metadata.json")
+            .setRuleSet("extra/ruleSet.json")
+            .setNormalized(true)
     }
 
     config {
diff --git a/examples/avro/build.gradle b/examples/avro/build.gradle
index e25c813..e040e29 100644
--- a/examples/avro/build.gradle
+++ b/examples/avro/build.gradle
@@ -20,6 +20,10 @@ schemaRegistry {
                 .addReference('company', 'company', -1)
         subject(locationAddressSubject)
         subject(locationCoordSubject)
+        subject("my-record","schemas/avro/my-record.avsc", "AVRO")
+                .setMetadata("extra/metadata.json")
+                .setRuleSet("extra/ruleSet.json")
+                .setNormalized(true)
     }
 
     config {
@@ -33,6 +37,7 @@ schemaRegistry {
         metadata = new MetadataExtension(true)
 //        subject('company', 'schemas/avro/downloaded') // is retrieved via reference from the user subject
         subject('user', 'schemas/avro/downloaded', true)
+        subject('my-record', 'schemas/avro/downloaded', true)
         subjectPattern('location.*', 'schemas/avro/downloaded/location')
     }
 
diff --git a/examples/build.gradle.kts b/examples/build.gradle.kts
index e602151..0c3ed15 100644
--- a/examples/build.gradle.kts
+++ b/examples/build.gradle.kts
@@ -1,7 +1,9 @@
 buildscript {
     repositories {
         gradlePluginPortal()
-//        mavenLocal()
+        mavenLocal()
+        maven("https://packages.confluent.io/maven/")
+        maven("https://jitpack.io")
     }
 
     dependencies {
diff --git a/examples/extra/metadata.json b/examples/extra/metadata.json
new file mode 100644
index 0000000..f48f5fb
--- /dev/null
+++ b/examples/extra/metadata.json
@@ -0,0 +1,9 @@
+{
+  "tags": {
+    "**.ssn": [ "PII" ]
+  },
+  "properties": {
+    "owner": "Bob Jones",
+    "email": "bob@acme.com"
+  }
+}
diff --git a/examples/extra/ruleSet.json b/examples/extra/ruleSet.json
new file mode 100644
index 0000000..9267229
--- /dev/null
+++ b/examples/extra/ruleSet.json
@@ -0,0 +1,19 @@
+{
+  "domainRules": [
+    {
+      "name": "encryptPII",
+      "kind": "TRANSFORM",
+      "type": "ENCRYPT",
+      "mode": "WRITEREAD",
+      "tags": [
+        "PII"
+      ],
+      "params": {
+        "encrypt.kek.name": "kafka-csfle",
+        "encrypt.kms.key.id": "projects/gcp-project/locations/europe-west6/keyRings/gcp-keyring/cryptoKeys/kafka-csfle",
+        "encrypt.kms.type": "gcp-kms"
+      },
+      "onFailure": "ERROR,NONE"
+    }
+  ]
+}
diff --git a/examples/json/build.gradle b/examples/json/build.gradle
index 83e1ef3..1d986c5 100644
--- a/examples/json/build.gradle
+++ b/examples/json/build.gradle
@@ -21,6 +21,10 @@ schemaRegistry {
             .addReference('company', 'company', -1)
         subject(locationAddressSubject)
         subject(locationCoordSubject)
+        subject('my-record','schemas/json/my-record.json','JSON')
+                .setMetadata('extra/metadata.json')
+                .setRuleSet('extra/ruleSet.json')
+                .setNormalized(true)
     }
 
     config {
diff --git a/examples/protobuf/build.gradle b/examples/protobuf/build.gradle
index 282aee1..983e682 100644
--- a/examples/protobuf/build.gradle
+++ b/examples/protobuf/build.gradle
@@ -20,6 +20,10 @@ schemaRegistry {
                 .addReference('company', 'company', 1)
         subject(locationAddressSubject)
         subject(locationCoordSubject)
+        subject('my-record','schemas/protobuf/com/example/my-record.proto','PROTOBUF')
+            .setMetadata('extra/metadata.json')
+            .setRuleSet('extra/ruleSet.json')
+            .setNormalized(true)
     }
 
     config {
diff --git a/examples/schemas/avro/my-record.avsc b/examples/schemas/avro/my-record.avsc
new file mode 100644
index 0000000..747e90c
--- /dev/null
+++ b/examples/schemas/avro/my-record.avsc
@@ -0,0 +1,10 @@
+{
+  "namespace": "com.github.imflog",
+  "type": "record",
+  "name": "MyRecord",
+  "fields":[{
+    "name":"ssn",
+    "type":"string",
+    "confluent:tags": [ "PII" ]
+  }]
+}
diff --git a/examples/schemas/json/my-record.json b/examples/schemas/json/my-record.json
new file mode 100644
index 0000000..e710cae
--- /dev/null
+++ b/examples/schemas/json/my-record.json
@@ -0,0 +1,12 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "$id": "MyRecord",
+  "type": "object",
+  "properties": {
+    "ssn": {
+      "type": "string",
+      "confluent:tags": [ "PII" ]
+    }
+  },
+  "additionalProperties": false
+}
diff --git a/examples/schemas/protobuf/com/example/my-record.proto b/examples/schemas/protobuf/com/example/my-record.proto
new file mode 100644
index 0000000..6e7e150
--- /dev/null
+++ b/examples/schemas/protobuf/com/example/my-record.proto
@@ -0,0 +1,10 @@
+syntax = "proto3";
+
+import "confluent/meta.proto";
+
+
+message MyRecord {
+  string ssn = 1 [
+    (confluent.field_meta).tags = "PII"
+  ];
+}
diff --git a/src/integration/kotlin/com/github/imflog/schema/registry/tasks/compatibility/CompatibilityTaskIT.kt b/src/integration/kotlin/com/github/imflog/schema/registry/tasks/compatibility/CompatibilityTaskIT.kt
index 80c4a91..4a1f9e3 100644
--- a/src/integration/kotlin/com/github/imflog/schema/registry/tasks/compatibility/CompatibilityTaskIT.kt
+++ b/src/integration/kotlin/com/github/imflog/schema/registry/tasks/compatibility/CompatibilityTaskIT.kt
@@ -1,6 +1,7 @@
 package com.github.imflog.schema.registry.tasks.compatibility
 
 import com.github.imflog.schema.registry.SchemaType
+import com.github.imflog.schema.registry.Subject
 import com.github.imflog.schema.registry.parser.SchemaParser
 import com.github.imflog.schema.registry.utils.KafkaTestContainersUtils
 import io.confluent.kafka.schemaregistry.ParsedSchema
@@ -231,20 +232,17 @@ class CompatibilityTaskIT : KafkaTestContainersUtils() {
         val addressFile = folderRule.newFile(addressPath)
         addressFile.writeText(addressSchema)
         val addressReferenceName = "Address"
-        val addressSubject = "$subjectName-address-mixed"
-        client.register(addressSubject, parser.parseSchemaFromFile(addressSubject, addressFile.path, listOf(), listOf()))
+        val addressSubject = Subject("$subjectName-address-mixed",addressFile.path, type.toString())
+        client.register(addressSubject.inputSubject, parser.parseSchemaFromFile(addressSubject))
 
         val playerPath = "$type/player.$extension"
-        val playerSubject = "$subjectName-player-mixed"
         val playerFile = folderRule.newFile(playerPath)
         playerFile.writeText(playerSchema)
+        val playerSubject = Subject("$subjectName-player-mixed",playerFile.path, type.toString())
         client.register(
-            playerSubject,
+            playerSubject.inputSubject,
             parser.parseSchemaFromFile(
-                playerSubject,
-                playerFile.path,
-                listOf(),
-                listOf()
+                playerSubject
             )
         )
 
@@ -262,9 +260,9 @@ class CompatibilityTaskIT : KafkaTestContainersUtils() {
             schemaRegistry {
                 url = '$schemaRegistryEndpoint'
                 compatibility {
-                    subject('$playerSubject', '${playerFile.absolutePath}', '${type.name}')
+                    subject('${playerSubject.inputSubject}', '${playerFile.absolutePath}', '${type.name}')
                         .addLocalReference('$userSubject', '$userPath')
-                        .addReference('$addressReferenceName', '$addressSubject', 1)
+                        .addReference('$addressReferenceName', '${addressSubject.inputSubject}', 1)
                 }
             }
         """
diff --git a/src/integration/kotlin/com/github/imflog/schema/registry/tasks/download/DownloadTaskIT.kt b/src/integration/kotlin/com/github/imflog/schema/registry/tasks/download/DownloadTaskIT.kt
index ac1bb68..ae55331 100644
--- a/src/integration/kotlin/com/github/imflog/schema/registry/tasks/download/DownloadTaskIT.kt
+++ b/src/integration/kotlin/com/github/imflog/schema/registry/tasks/download/DownloadTaskIT.kt
@@ -3,15 +3,13 @@ package com.github.imflog.schema.registry.tasks.download
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.databind.PropertyNamingStrategies
 import com.fasterxml.jackson.databind.SerializationFeature
-import com.github.imflog.schema.registry.SchemaType
 import com.github.imflog.schema.registry.toSchemaType
 import com.github.imflog.schema.registry.utils.KafkaTestContainersUtils
 import io.confluent.kafka.schemaregistry.ParsedSchema
 import io.confluent.kafka.schemaregistry.avro.AvroSchema
-import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaReference
+import io.confluent.kafka.schemaregistry.client.rest.entities.*
 import io.confluent.kafka.schemaregistry.json.JsonSchema
 import io.confluent.kafka.schemaregistry.protobuf.ProtobufSchema
-import org.apache.avro.Schema
 import org.assertj.core.api.Assertions
 import org.gradle.internal.impldep.org.junit.rules.TemporaryFolder
 import org.gradle.testkit.runner.BuildResult
@@ -107,7 +105,7 @@ class DownloadTaskIT : KafkaTestContainersUtils() {
 
         val type = AvroSchema.TYPE
         val schema =  AvroSchema(
-                """{
+            """{
                             "type": "record",
                             "name": "User",
                             "fields": [
@@ -354,10 +352,17 @@ class DownloadTaskIT : KafkaTestContainersUtils() {
                     "type": "record",
                     "name": "User",
                     "fields": [
-                        { "name": "name", "type": "string" }
+                        { "name": "name", "type": "string", "confluent:tags": [ "PII" ] }
                     ]
-                }"""
-            )
+                }""", emptyList(), emptyMap(),
+                Metadata(mapOf(Pair("**.ssn", setOf("PII"))), mapOf(Pair("owner", "Bob Jones"),
+                    Pair("email", "bob@acme.com")),null),
+                RuleSet(null, listOf(
+                    Rule("encryptPII",null,RuleKind.TRANSFORM,RuleMode.WRITEREAD,"ENCRYPT",
+                        setOf("PII"), mapOf(Pair("encrypt.kek.name", "kafka-csfle"),
+                            Pair("encrypt.kms.key.id", "projects/gcp-project/locations/europe-west6/keyRings/gcp-keyring/cryptoKeys/kafka-csfle"),
+                            Pair("encrypt.kms.type", "gcp-kms")),null,null,"ERROR,NONE",false))),
+                null,true),true
         )
 
         buildFile = folderRule.newFile("build.gradle")
@@ -479,6 +484,76 @@ class DownloadTaskIT : KafkaTestContainersUtils() {
         Assertions.assertThat(result?.task(":downloadSchemasTask")?.outcome).isEqualTo(TaskOutcome.SUCCESS)
     }
 
+    @Test
+    fun `Should download schema that has been normalized`() {
+        // Given
+        val subjectName = "test-proto"
+
+        client.register(
+            subjectName,
+            ProtobufSchema(
+                """
+                        syntax = "proto3";
+                        option java_outer_classname = "User";
+                        package foo.v1;
+                        option java_multiple_files = true;
+                        message TestMessage {
+                            string name = 1;
+                            string description = 2;
+                        }
+                    """, emptyList(), emptyMap(),null ,null),
+            true)
+
+        buildFile = folderRule.newFile("build.gradle")
+        buildFile.writeText(
+            """
+            plugins {
+                id 'java'
+                id 'com.github.imflog.kafka-schema-registry-gradle-plugin'
+            }
+
+            import com.github.imflog.schema.registry.tasks.download.MetadataExtension
+            schemaRegistry {
+                url = '$schemaRegistryEndpoint'
+                download {
+                    metadata = new MetadataExtension(true)
+                    subject('$subjectName', '${folderRule.root.absolutePath}/src/main/protobuf/test')
+                }
+            }
+        """
+        )
+
+        // When
+        val result: BuildResult? = GradleRunner.create()
+            .withGradleVersion("6.7.1")
+            .withProjectDir(folderRule.root)
+            .withArguments(DownloadTask.TASK_NAME)
+            .withPluginClasspath()
+            .withDebug(true)
+            .build()
+
+        // Then
+        val schemaFile = "$subjectName.proto"
+        val metadataFile = "$subjectName-metadata.json"
+        Assertions.assertThat(File(folderRule.root, "src/main/protobuf/test/$schemaFile")).exists()
+        Assertions.assertThat(File(folderRule.root, "src/main/protobuf/test/$metadataFile")).exists()
+        Assertions.assertThat(result?.task(":downloadSchemasTask")?.outcome).isEqualTo(TaskOutcome.SUCCESS)
+        val resultFile = File(folderRule.root, "src/main/protobuf/test/$schemaFile")
+
+        Assertions.assertThat(resultFile.readText().trim()).isEqualTo(
+        """
+        syntax = "proto3";
+        package foo.v1;
+        
+        option java_multiple_files = true;
+        option java_outer_classname = "User";
+        
+        message TestMessage {
+          string name = 1;
+          string description = 2;
+        }""".trimIndent().trim())
+    }
+
     private class SchemaArgumentProvider : ArgumentsProvider {
         override fun provideArguments(context: ExtensionContext): Stream<out Arguments> =
             Stream.of(
diff --git a/src/integration/kotlin/com/github/imflog/schema/registry/tasks/register/RegisterTaskIT.kt b/src/integration/kotlin/com/github/imflog/schema/registry/tasks/register/RegisterTaskIT.kt
index 69f2372..4137b93 100644
--- a/src/integration/kotlin/com/github/imflog/schema/registry/tasks/register/RegisterTaskIT.kt
+++ b/src/integration/kotlin/com/github/imflog/schema/registry/tasks/register/RegisterTaskIT.kt
@@ -95,6 +95,81 @@ class RegisterTaskIT : KafkaTestContainersUtils() {
             )
     }
 
+    @ParameterizedTest
+    @ArgumentsSource(SchemaWithMetadataAndRuleSetArgumentProvider::class)
+    fun `RegisterSchemasTask should register normalised schemas with metadata and rulesets`(
+        type: SchemaType,
+        userSchema: String,
+        playerSchema: String,
+        metadata: String,
+        ruleSet: String,
+    ) {
+        folderRule.create()
+        val typeFolder = folderRule.newFolder(type.name)
+        val resultFolder = folderRule.newFolder("${type.name}/results")
+        val subjectName = "parameterized-${type.name}"
+        val extension = type.extension
+
+        val userFile = typeFolder.resolve("user.$extension")
+        userFile.writeText(userSchema)
+        val userSubject = "$subjectName-user-local"
+
+        val metadataFile = typeFolder.resolve("metadata.json")
+        metadataFile.writeText(metadata)
+
+        val ruleSetFile = typeFolder.resolve("ruleSet.json")
+        ruleSetFile.writeText(ruleSet)
+
+        val playerFile = typeFolder.resolve("player.$extension")
+        val playerPath = playerFile.relativeTo(folderRule.root).path
+        playerFile.writeText(playerSchema)
+        val playerSubject = "$subjectName-player-local"
+
+        // Small trick, for protobuf the name to import is not User but user.proto
+        val referenceName = if (type == SchemaType.PROTOBUF) "user.proto" else "User"
+
+        buildFile = folderRule.newFile("build.gradle")
+        buildFile.writeText(
+            """
+            plugins {
+                id 'java'
+                id 'com.github.imflog.kafka-schema-registry-gradle-plugin'
+            }
+
+            schemaRegistry {
+                url = '$schemaRegistryEndpoint'
+                outputDirectory = '${resultFolder.absolutePath}'
+                register {
+                    subject('$userSubject', '${userFile.absolutePath}', '${type.name}')
+                        .setMetadata('${metadataFile.absolutePath}')
+                        .setRuleSet('${ruleSetFile.absolutePath}')
+                        .setNormalized(true)
+                    subject('$playerSubject', '$playerPath', '${type.name}')
+                        .addReference('$referenceName', '$userSubject', 1)
+                        .setNormalized(true)
+                }
+            }
+        """
+        )
+
+        val result: BuildResult? = GradleRunner.create()
+            .withGradleVersion("6.7.1")
+            .withProjectDir(folderRule.root)
+            .withArguments(RegisterSchemasTask.TASK_NAME)
+            .withPluginClasspath()
+            .withDebug(true)
+            .build()
+        Assertions.assertThat(result?.task(":registerSchemasTask")?.outcome).isEqualTo(TaskOutcome.SUCCESS)
+        Assertions.assertThat(resultFolder.resolve("registered.csv")).exists()
+        Assertions.assertThat(resultFolder.resolve("registered.csv").readText())
+            .matches(
+                """subject, path, id
+                |$userSubject, ${userFile.absolutePath}, \d+
+                |$playerSubject, $playerPath, \d+
+                |""".trimMargin()
+            )
+    }
+
     @ParameterizedTest
     @ArgumentsSource(LocalSchemaArgumentProvider::class)
     fun `Should register local references`(
@@ -277,6 +352,165 @@ class RegisterTaskIT : KafkaTestContainersUtils() {
             )
     }
 
+    private class SchemaWithMetadataAndRuleSetArgumentProvider : ArgumentsProvider {
+        override fun provideArguments(context: ExtensionContext): Stream<out Arguments> =
+            Stream.of(
+                Arguments.of(
+                    SchemaType.AVRO,
+                    """{
+                        "type": "record",
+                        "name": "User",
+                        "fields": [
+                            { "name": "name", "type": "string" }
+                        ]
+                    }""",
+                    """{
+                        "type": "record",
+                        "name": "Player",
+                        "fields": [
+                            { "name": "identifier", "type": "string", "confluent:tags": [ "PII" ] }, 
+                            { "name": "user", "type": "User" }
+                        ]
+                    }""",
+                    """{
+                      "domainRules": [
+                        {
+                          "name": "encryptPII",
+                          "kind": "TRANSFORM",
+                          "type": "ENCRYPT",
+                          "mode": "WRITEREAD",
+                          "tags": [
+                            "PII"
+                          ],
+                          "params": {
+                            "encrypt.kek.name": "kafka-csfle",
+                            "encrypt.kms.key.id": "projects/gcp-project/locations/europe-west6/keyRings/gcp-keyring/cryptoKeys/kafka-csfle",
+                            "encrypt.kms.type": "gcp-kms"
+                          },
+                          "onFailure": "ERROR,NONE"
+                        }
+                      ]
+                    }""",
+                    """{
+                      "tags": {
+                        "**.ssn": [ "PII" ]
+                      },
+                      "properties": {
+                        "owner": "Bob Jones",
+                        "email": "bob@acme.com"
+                      }
+                    }"""
+                ),
+                Arguments.of(
+                    SchemaType.JSON,
+                    """{
+                        "${"$"}schema": "http://json-schema.org/draft-07/schema#",
+                        "type": "object",
+                        "properties": {
+                            "name": {
+                                "type": "string",
+                                "confluent:tags": [ "PII" ]
+                            }
+                        },
+                        "additionalProperties": false
+                    }""",
+                    """{
+                        "${"$"}schema": "http://json-schema.org/draft-07/schema#",
+                        "type": "object",
+                        "properties": {
+                            "identifier": {"type": "string"},
+                            "user": {"${"$"}ref": "User"}
+                        },
+                        "additionalProperties": false
+                    }""",
+                    """{
+                      "domainRules": [
+                        {
+                          "name": "encryptPII",
+                          "kind": "TRANSFORM",
+                          "type": "ENCRYPT",
+                          "mode": "WRITEREAD",
+                          "tags": [
+                            "PII"
+                          ],
+                          "params": {
+                            "encrypt.kek.name": "kafka-csfle",
+                            "encrypt.kms.key.id": "projects/gcp-project/locations/europe-west6/keyRings/gcp-keyring/cryptoKeys/kafka-csfle",
+                            "encrypt.kms.type": "gcp-kms"
+                          },
+                          "onFailure": "ERROR,NONE"
+                        }
+                      ]
+                    }""",
+                    """{
+                      "tags": {
+                        "**.ssn": [ "PII" ]
+                      },
+                      "properties": {
+                        "owner": "Bob Jones",
+                        "email": "bob@acme.com"
+                      }
+                    }"""
+                ),
+                Arguments.of(
+                    SchemaType.PROTOBUF,
+                    """
+                    syntax = "proto3";
+                    import "confluent/meta.proto";
+
+                    package com.github.imflog;
+                    
+                    option java_package = "com.github.proto.imflog.v1";
+                    option java_multiple_files = true;
+                    option java_outer_classname = "UserProto";
+                    
+                    message User {
+                        string name = 1 [(.confluent.field_meta).tags = "PII"];
+                    }
+                    """,
+                    """
+                    syntax = "proto3";
+                    package com.github.imflog;
+                    
+                    import "user.proto";
+                    
+                    message Player {
+                        string identifier = 1;
+                        User user = 2;
+                    }
+                    """,
+                    """{
+                      "domainRules": [
+                        {
+                          "name": "encryptPII",
+                          "kind": "TRANSFORM",
+                          "type": "ENCRYPT",
+                          "mode": "WRITEREAD",
+                          "tags": [
+                            "PII"
+                          ],
+                          "params": {
+                            "encrypt.kek.name": "kafka-csfle",
+                            "encrypt.kms.key.id": "projects/gcp-project/locations/europe-west6/keyRings/gcp-keyring/cryptoKeys/kafka-csfle",
+                            "encrypt.kms.type": "gcp-kms"
+                          },
+                          "onFailure": "ERROR,NONE"
+                        }
+                      ]
+                    }""",
+                    """{
+                      "tags": {
+                        "**.ssn": [ "PII" ]
+                      },
+                      "properties": {
+                        "owner": "Bob Jones",
+                        "email": "bob@acme.com"
+                      }
+                    }"""
+                )
+            )
+    }
+
     private class LocalSchemaArgumentProvider : ArgumentsProvider {
         override fun provideArguments(context: ExtensionContext): Stream<out Arguments> =
             Stream.of(
diff --git a/src/main/kotlin/com/github/imflog/schema/registry/Subject.kt b/src/main/kotlin/com/github/imflog/schema/registry/Subject.kt
index 04d3c6f..60e43bc 100644
--- a/src/main/kotlin/com/github/imflog/schema/registry/Subject.kt
+++ b/src/main/kotlin/com/github/imflog/schema/registry/Subject.kt
@@ -1,6 +1,10 @@
 package com.github.imflog.schema.registry
 
+import com.google.gson.Gson
+import io.confluent.kafka.schemaregistry.client.rest.entities.Metadata
+import io.confluent.kafka.schemaregistry.client.rest.entities.RuleSet
 import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaReference
+import java.io.File
 
 data class Subject(
     val inputSubject: String,
@@ -9,6 +13,10 @@ data class Subject(
 ) {
     val references: MutableList<SchemaReference> = mutableListOf()
     val localReferences: MutableList<LocalReference> = mutableListOf()
+    var metadata: Metadata? = null
+    var ruleSet: RuleSet? = null
+    var normalize: Boolean = false
+
 
     fun addReference(name: String, subject: String, version: Int): Subject {
         references.add(SchemaReference(name, subject, version))
@@ -25,7 +33,22 @@ data class Subject(
         return this
     }
 
-    // Used for data class destructuring
-    operator fun component4() = references
-    operator fun component5() = localReferences
-}
\ No newline at end of file
+    fun setMetadata(path: String): Subject {
+        val metadataContent = File(path).readText(Charsets.UTF_8)
+        metadata =  Gson().fromJson(metadataContent, Metadata::class.java)
+        return this
+    }
+
+    fun setRuleSet(path: String): Subject {
+        val ruleSetContent = File(path).readText(Charsets.UTF_8)
+        ruleSet = Gson().fromJson(ruleSetContent, RuleSet::class.java)
+        return this
+    }
+
+    fun setNormalized(normalize: Boolean): Subject {
+        this.normalize = normalize
+        return this
+    }
+
+
+}
diff --git a/src/main/kotlin/com/github/imflog/schema/registry/parser/SchemaParser.kt b/src/main/kotlin/com/github/imflog/schema/registry/parser/SchemaParser.kt
index 051c78f..c589548 100644
--- a/src/main/kotlin/com/github/imflog/schema/registry/parser/SchemaParser.kt
+++ b/src/main/kotlin/com/github/imflog/schema/registry/parser/SchemaParser.kt
@@ -3,8 +3,11 @@ package com.github.imflog.schema.registry.parser
 import com.github.imflog.schema.registry.LocalReference
 import com.github.imflog.schema.registry.SchemaParsingException
 import com.github.imflog.schema.registry.SchemaType
+import com.github.imflog.schema.registry.Subject
 import io.confluent.kafka.schemaregistry.ParsedSchema
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient
+import io.confluent.kafka.schemaregistry.client.rest.entities.Metadata
+import io.confluent.kafka.schemaregistry.client.rest.entities.RuleSet
 import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaReference
 import java.io.File
 
@@ -29,19 +32,16 @@ abstract class SchemaParser(
 
     @Throws(SchemaParsingException::class, NotImplementedError::class)
     fun parseSchemaFromFile(
-        subject: String,
-        schemaPath: String,
-        remoteReferences: List<SchemaReference>,
-        localReferences: List<LocalReference>,
+        subject: Subject
     ): ParsedSchema {
-        val schemaContent = rootDir.resolve(schemaPath).readText()
-        val parsedLocalSchemaString = if (localReferences.isNotEmpty()) {
-            resolveLocalReferences(subject, schemaContent, localReferences)
+        val schemaContent = rootDir.resolve(subject.file).readText()
+        val parsedLocalSchemaString = if (subject.localReferences.isNotEmpty()) {
+            resolveLocalReferences(subject.inputSubject, schemaContent, subject.localReferences)
         } else schemaContent
 
         return client
-            .parseSchema(schemaType.registryType, parsedLocalSchemaString, remoteReferences)
-            .orElseThrow { SchemaParsingException(subject, schemaType) }
+            .parseSchema(schemaType.registryType, parsedLocalSchemaString, subject.references, subject.metadata,subject.ruleSet)
+            .orElseThrow { SchemaParsingException(subject.inputSubject, schemaType) }
     }
 
     abstract fun resolveLocalReferences(
diff --git a/src/main/kotlin/com/github/imflog/schema/registry/tasks/compatibility/CompatibilityTaskAction.kt b/src/main/kotlin/com/github/imflog/schema/registry/tasks/compatibility/CompatibilityTaskAction.kt
index ac65915..a6e26a3 100644
--- a/src/main/kotlin/com/github/imflog/schema/registry/tasks/compatibility/CompatibilityTaskAction.kt
+++ b/src/main/kotlin/com/github/imflog/schema/registry/tasks/compatibility/CompatibilityTaskAction.kt
@@ -21,21 +21,16 @@ class CompatibilityTaskAction(
 
     fun run(): Int {
         var errorCount = 0
-        for ((subject, path, type, remoteReferences, localReferences) in subjects) {
-            logger.debug("Loading schema for subject($subject) from $path.")
+        subjects.forEach { subject ->
+            logger.debug("Loading schema for subject(${subject.inputSubject}) from ${subject.file}.")
             val isCompatible = try {
                 val parsedSchema = SchemaParser
-                    .provide(type.toSchemaType(), client, rootDir)
-                    .parseSchemaFromFile(
-                        subject,
-                        path,
-                        remoteReferences,
-                        localReferences
-                    )
-                val isCompatible = client.testCompatibility(subject, parsedSchema)
+                    .provide(subject.type.toSchemaType(), client, rootDir)
+                    .parseSchemaFromFile(subject)
+                val isCompatible = client.testCompatibility(subject.inputSubject, parsedSchema)
                 if (!isCompatible) {
                     try {
-                        client.testCompatibilityVerbose(subject, parsedSchema).forEach {
+                        client.testCompatibilityVerbose(subject.inputSubject, parsedSchema).forEach {
                             logger.error("Returned errors : $it")
                         }
                     } catch (_: Exception) {
@@ -56,9 +51,9 @@ class CompatibilityTaskAction(
                 }
             }
             if (isCompatible) {
-                logger.infoIfNotQuiet("Schema $path is compatible with subject: $subject")
+                logger.infoIfNotQuiet("Schema ${subject.file} is compatible with subject: ${subject.inputSubject}")
             } else {
-                logger.error("Schema $path is not compatible with subject: $subject")
+                logger.error("Schema ${subject.file} is not compatible with subject: ${subject.inputSubject}")
                 errorCount++
             }
         }
diff --git a/src/main/kotlin/com/github/imflog/schema/registry/tasks/register/RegisterTaskAction.kt b/src/main/kotlin/com/github/imflog/schema/registry/tasks/register/RegisterTaskAction.kt
index c0583d9..3683d8f 100644
--- a/src/main/kotlin/com/github/imflog/schema/registry/tasks/register/RegisterTaskAction.kt
+++ b/src/main/kotlin/com/github/imflog/schema/registry/tasks/register/RegisterTaskAction.kt
@@ -7,6 +7,8 @@ import com.github.imflog.schema.registry.Subject
 import com.github.imflog.schema.registry.parser.SchemaParser
 import com.github.imflog.schema.registry.toSchemaType
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient
+import io.confluent.kafka.schemaregistry.client.rest.entities.Metadata
+import io.confluent.kafka.schemaregistry.client.rest.entities.RuleSet
 import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaReference
 import org.gradle.api.logging.Logging
 import java.io.File
@@ -27,10 +29,10 @@ class RegisterTaskAction(
     fun run(): Int {
         var errorCount = 0
         writeOutputFileHeader()
-        subjects.forEach { (subject, path, type, references: List<SchemaReference>, localReferences) ->
+        subjects.forEach { subject ->
             try {
-                val schemaId = registerSchema(subject, path, type.toSchemaType(), references, localReferences)
-                writeRegisteredSchemaOutput(subject, path, schemaId)
+                val schemaId = registerSchema(subject)
+                writeRegisteredSchemaOutput(subject.inputSubject,subject.file, schemaId)
             } catch (e: Exception) {
                 logger.error("Could not register schema for '$subject'", e)
                 errorCount++
@@ -40,18 +42,14 @@ class RegisterTaskAction(
     }
 
     private fun registerSchema(
-        subject: String,
-        path: String,
-        type: SchemaType,
-        references: List<SchemaReference>,
-        localReferences: List<LocalReference>
+        subject: Subject
     ): Int {
         val parsedSchema = SchemaParser
-            .provide(type, client, rootDir)
-            .parseSchemaFromFile(subject, path, references, localReferences)
-        logger.infoIfNotQuiet("Registering $subject (from $path)")
-        val schemaId = client.register(subject, parsedSchema)
-        logger.infoIfNotQuiet("$subject (from $path) has been registered with id $schemaId")
+            .provide(subject.type.toSchemaType(), client, rootDir)
+            .parseSchemaFromFile(subject)
+        logger.infoIfNotQuiet("Registering $subject (from $subject.file)")
+        val schemaId = client.register(subject.inputSubject, parsedSchema, subject.normalize)
+        logger.infoIfNotQuiet("$subject (from $subject.file) has been registered with id $schemaId")
         return schemaId
     }
 


Test patch:
diff --git a/src/test/kotlin/com/github/imflog/schema/registry/tasks/download/DownloadTaskActionTest.kt b/src/test/kotlin/com/github/imflog/schema/registry/tasks/download/DownloadTaskActionTest.kt
index 236e9ee..d3bbba8 100644
--- a/src/test/kotlin/com/github/imflog/schema/registry/tasks/download/DownloadTaskActionTest.kt
+++ b/src/test/kotlin/com/github/imflog/schema/registry/tasks/download/DownloadTaskActionTest.kt
@@ -468,7 +468,6 @@ class DownloadTaskActionTest {
         val testSubject = "test"
         val testLibSubject = "test_lib"
         val outputDir = "src/main/avro/external"
-        val metadataDir = "src/main/avro/metadata"
 
         val registryClient = MockSchemaRegistryClient(listOf(AvroSchemaProvider()))
 
diff --git a/src/test/kotlin/com/github/imflog/schema/registry/tasks/register/RegisterTaskActionTest.kt b/src/test/kotlin/com/github/imflog/schema/registry/tasks/register/RegisterTaskActionTest.kt
index 9d7f060..a406a2f 100644
--- a/src/test/kotlin/com/github/imflog/schema/registry/tasks/register/RegisterTaskActionTest.kt
+++ b/src/test/kotlin/com/github/imflog/schema/registry/tasks/register/RegisterTaskActionTest.kt
@@ -350,4 +350,104 @@ class RegisterTaskActionTest {
         // Then
         Assertions.assertThat(errorCount).isEqualTo(0)
     }
+
+    @Test
+    fun `Should register schema with ruleSet`() {
+        // Given
+        val registryClient =
+            MockSchemaRegistryClient(listOf(AvroSchemaProvider(), JsonSchemaProvider(), ProtobufSchemaProvider()))
+        val projectDirAbsolutePath = Paths.get("").toAbsolutePath().toString()
+        val schemaPath = "$projectDirAbsolutePath/src/test/resources/"
+        val subjects = listOf(
+            Subject("test", "${schemaPath}testSubjectWithTag.avsc", "AVRO")
+                .setRuleSet("${schemaPath}testRuleSet.json")
+        )
+
+        // When
+        val errorCount = RegisterTaskAction(
+            registryClient,
+            folderRule.root.toFile(),
+            subjects,
+            null
+        ).run()
+
+        // Then
+        Assertions.assertThat(errorCount).isEqualTo(0)
+        Assertions.assertThat(registryClient.getLatestSchemaMetadata("test").ruleSet).isNotNull
+    }
+
+    @Test
+    fun `Should register schema with metadata`() {
+        // Given
+        val registryClient =
+            MockSchemaRegistryClient(listOf(AvroSchemaProvider(), JsonSchemaProvider(), ProtobufSchemaProvider()))
+        val projectDirAbsolutePath = Paths.get("").toAbsolutePath().toString()
+        val schemaPath = "$projectDirAbsolutePath/src/test/resources/"
+        val subjects = listOf(
+            Subject("test", "${schemaPath}testSimpleSubject.avsc", "AVRO")
+                .setMetadata("${schemaPath}testMetadata.json")
+        )
+
+        // When
+        val errorCount = RegisterTaskAction(
+            registryClient,
+            folderRule.root.toFile(),
+            subjects,
+            null
+        ).run()
+
+        // Then
+        Assertions.assertThat(errorCount).isEqualTo(0)
+        Assertions.assertThat(registryClient.getLatestSchemaMetadata("test").metadata).isNotNull
+    }
+
+    @Test
+    fun `Should register normalized schema`() {
+        // Given
+        val registryClient =
+            MockSchemaRegistryClient(listOf(AvroSchemaProvider(), JsonSchemaProvider(), ProtobufSchemaProvider()))
+        val resultFolder = Files.createDirectories(folderRule.resolve("results/protobuf")).toFile()
+        File(folderRule.toFile(), "src/main/avro/external/test.proto")
+            .writeText(
+                """
+            syntax = "proto3";
+            
+            option java_package = "com.example.proto.v1";
+            package foo.v1;
+               
+            option java_outer_classname = "FooProto";
+            option java_multiple_files = true;
+            
+            message Foo {
+              string bar = 1;
+            }
+            """
+            )
+
+        val subjects = listOf(
+            Subject("test", "src/main/avro/external/test.proto", "PROTOBUF").setNormalized(true),
+        )
+
+        // when
+        RegisterTaskAction(
+            registryClient,
+            folderRule.toFile(),
+            subjects,
+            resultFolder.path
+        ).run()
+
+        // then
+        Assertions.assertThat(registryClient.getLatestSchemaMetadata("test").schema)
+            .isEqualTo("syntax = \"proto3\";\n" +
+                    "package foo.v1;\n" +
+                    "\n" +
+                    "option java_multiple_files = true;\n" +
+                    "option java_outer_classname = \"FooProto\";\n" +
+                    "option java_package = \"com.example.proto.v1\";\n" +
+                    "\n" +
+                    "message Foo {\n" +
+                    "  string bar = 1;\n" +
+                    "}\n"
+        )
+    }
 }
diff --git a/src/test/resources/testMetadata.json b/src/test/resources/testMetadata.json
new file mode 100644
index 0000000..f48f5fb
--- /dev/null
+++ b/src/test/resources/testMetadata.json
@@ -0,0 +1,9 @@
+{
+  "tags": {
+    "**.ssn": [ "PII" ]
+  },
+  "properties": {
+    "owner": "Bob Jones",
+    "email": "bob@acme.com"
+  }
+}
diff --git a/src/test/resources/testRuleSet.json b/src/test/resources/testRuleSet.json
new file mode 100644
index 0000000..9267229
--- /dev/null
+++ b/src/test/resources/testRuleSet.json
@@ -0,0 +1,19 @@
+{
+  "domainRules": [
+    {
+      "name": "encryptPII",
+      "kind": "TRANSFORM",
+      "type": "ENCRYPT",
+      "mode": "WRITEREAD",
+      "tags": [
+        "PII"
+      ],
+      "params": {
+        "encrypt.kek.name": "kafka-csfle",
+        "encrypt.kms.key.id": "projects/gcp-project/locations/europe-west6/keyRings/gcp-keyring/cryptoKeys/kafka-csfle",
+        "encrypt.kms.type": "gcp-kms"
+      },
+      "onFailure": "ERROR,NONE"
+    }
+  ]
+}
diff --git a/src/test/resources/testSimpleSubject.avsc b/src/test/resources/testSimpleSubject.avsc
new file mode 100644
index 0000000..ca6fb2d
--- /dev/null
+++ b/src/test/resources/testSimpleSubject.avsc
@@ -0,0 +1,11 @@
+{
+  "type": "record",
+  "name": "A",
+  "namespace": "com.mycompany",
+  "fields": [
+    {
+      "name": "foo",
+      "type": "string"
+    }
+  ]
+}
diff --git a/src/test/resources/testSubject.avsc b/src/test/resources/testSubject.avsc
index 0551b62..8d851a0 100644
--- a/src/test/resources/testSubject.avsc
+++ b/src/test/resources/testSubject.avsc
@@ -12,4 +12,4 @@
       "type": "B"
     }
   ]
-}
\ No newline at end of file
+}
diff --git a/src/test/resources/testSubjectWithTag.avsc b/src/test/resources/testSubjectWithTag.avsc
new file mode 100644
index 0000000..562f7a4
--- /dev/null
+++ b/src/test/resources/testSubjectWithTag.avsc
@@ -0,0 +1,12 @@
+{
+  "type": "record",
+  "name": "A",
+  "namespace": "com.mycompany",
+  "fields": [
+    {
+      "name": "ssn",
+      "type": "string",
+      "confluent:tags": [ "PII" ]
+    }
+  ]
+}