Repository: ImFlog/schema-registry-plugin
Pull request #24 (2019-06-08 23:23:36)

Add support for setting subject configuration
This patch adds support for configuring subjects in the registry, which currently involves setting the compatibility level to be enforced on subsequent schema registrations to a subject.

I needed this for an internal project.

ImFlog: I am wondering if we shouldn't add this compatibility setting in the register task too.
We would call it after the register to allow user to change it directly if specified.
WDYT ?
bspeakmon: I had the same thought and decided it wasn't the right thing to do.

Part of the idea of registering schemas is that the registration _should_ fail if the compatibility is invalid. If the same entity is both consumer and producer of a schema, that's fine; compatibility breaks only affect them. But if the compatibility is a contract between the schema producer and consumer as two different entities, then the producer should _not_ be able to change that at will, lest they break consumers without warning.

In our case, we have people iterating on a producer schema at will and many consumers of it. As long as the compatibility level is consistent, both sides are protected: change can happen at will. Registering a new .avsc as a subject does not give one the right to change the compatibility contract, however.
ImFlog: I thought about it too and you are right, this should not be in the register task.
What I have in mind is something like a `createSubjectTask` that create a subject and set the required compatibility. A task that would fail if the subject already exist.

I am not sure it impacts this code but I have question about what is the use case of such task ?
bspeakmon: I could think of one possible use case.

* Local development/test cases or DR process: existing code repository with schemas, brand new deployment with a schema registry. Run the build once to set everything up. Instead of writing a curl script in bash, you have a platform-independent tool for doing this. Existing subjects would be an error, so you can't accidentally clobber something important. 
ImFlog: Don't you think It would be better to replace this task by one that create a subject and then set compatibility that could only run once ? To avoid compatibility switch all the time ?
bspeakmon: I think not. At least in my use case, it's possible that I want to change compatibility later on. I may want to start at `FULL_TRANSITIVE` and later relax that to `BACKWARD_TRANSITIVE`, which is fully compatible with all existing subject versions. With a separate config task, I can do that while documenting my configuration intention separately from my schema registration action. 

My use case isn't the only one though. I can put together a task like you describe if you feel that the other use case (of an immutable, first-time setup) is important for you :)
ImFlog: No worries, I'll merge this and work on the one I talked about a bit later :)
I'll release this tommorow I think.
Thank you again !

Patch:
diff --git a/README.md b/README.md
index 230e492..1d05463 100644
--- a/README.md
+++ b/README.md
@@ -6,10 +6,12 @@ The aim of this plugin is to adapt the [Confluent schema registry maven plugin](
 See [gradle plugins portal](https://plugins.gradle.org/plugin/com.github.imflog.kafka-schema-registry-gradle-plugin)
 for instructions about how to add the plugin to your build configuration.
 
-When you do so, three tasks are added under registry group:
+When you do so, four tasks are added under registry group:
 * downloadSchemasTask
 * testSchemasTask
 * registerSchemasTask
+* configSubjectsTask
+
 What these tasks do and how to configure them is described in the following sections.
 ## Download schemas
 Like the name of the task imply, this task is responsible of retrieving schemas from a schema registry.
@@ -104,3 +106,31 @@ register{
     subject('userSubject', 'path/user.avsc', ['path/address.avsc', 'path/street.avsc'])
 }
 ```
+
+## Configure subjects
+
+This task sets the schema compatibility level for registered subjects.
+
+A DSL is available to specify which subjects to configure:
+```groovy
+schemaRegistry {
+    url = 'http://localhost:8081'
+    credentials {
+        username = 'basicauthentication-username'
+        password = 'basicauthentication-password'
+    } //optional
+    config {
+        subject('mySubject', 'FULL_TRANSITIVE')
+        subject('otherSubject', 'FORWARD')
+    }
+}
+```
+
+See the Confluent
+[Schema Registry documentation](https://docs.confluent.io/current/schema-registry/avro.html#compatibility-types)
+for more information on valid compatibility levels.
+
+You have to put the URL where the script can reach the Schema Registry.
+
+You have to list the (subject, compatibility-level) 
+
diff --git a/src/main/kotlin/com/github/imflog/schema/registry/SchemaRegistryPlugin.kt b/src/main/kotlin/com/github/imflog/schema/registry/SchemaRegistryPlugin.kt
index e0bdc34..2217b6f 100644
--- a/src/main/kotlin/com/github/imflog/schema/registry/SchemaRegistryPlugin.kt
+++ b/src/main/kotlin/com/github/imflog/schema/registry/SchemaRegistryPlugin.kt
@@ -3,6 +3,9 @@ package com.github.imflog.schema.registry
 import com.github.imflog.schema.registry.compatibility.CompatibilitySubjectExtension
 import com.github.imflog.schema.registry.compatibility.CompatibilityTask
 import com.github.imflog.schema.registry.compatibility.TEST_SCHEMAS_TASK
+import com.github.imflog.schema.registry.config.CONFIG_SUBJECTS_TASK
+import com.github.imflog.schema.registry.config.ConfigSubjectExtension
+import com.github.imflog.schema.registry.config.ConfigTask
 import com.github.imflog.schema.registry.download.DOWNLOAD_SCHEMAS_TASK
 import com.github.imflog.schema.registry.download.DownloadTask
 import com.github.imflog.schema.registry.register.REGISTER_SCHEMAS_TASK
@@ -35,6 +38,10 @@ class SchemaRegistryPlugin : Plugin<Project> {
                 "credentials",
                 SchemaRegistryBasicAuthExtension::class.java
             )
+            val configExtension = extensions.create(
+                "config",
+                ConfigSubjectExtension::class.java
+            )
 
             afterEvaluate {
                 tasks.create(
@@ -60,6 +67,14 @@ class SchemaRegistryPlugin : Plugin<Project> {
                     auth = authExtension
                     subjects = compatibilityExtension.subjects
                 }
+
+                tasks.create(
+                    CONFIG_SUBJECTS_TASK, ConfigTask::class.java
+                ).apply {
+                    url = globalExtension.url
+                    auth = authExtension
+                    subjects = configExtension.subjects
+                }
             }
         }
     }
diff --git a/src/main/kotlin/com/github/imflog/schema/registry/config/ConfigSubjectExtension.kt b/src/main/kotlin/com/github/imflog/schema/registry/config/ConfigSubjectExtension.kt
new file mode 100644
index 0000000..03d0df2
--- /dev/null
+++ b/src/main/kotlin/com/github/imflog/schema/registry/config/ConfigSubjectExtension.kt
@@ -0,0 +1,9 @@
+package com.github.imflog.schema.registry.config
+
+open class ConfigSubjectExtension {
+    val subjects: ArrayList<Pair<String, String>> = ArrayList()
+
+    fun subject(inputSubject: String, compatibility: String) {
+        subjects.add(Pair(inputSubject, compatibility))
+    }
+}
\ No newline at end of file
diff --git a/src/main/kotlin/com/github/imflog/schema/registry/config/ConfigTask.kt b/src/main/kotlin/com/github/imflog/schema/registry/config/ConfigTask.kt
new file mode 100644
index 0000000..aff9c18
--- /dev/null
+++ b/src/main/kotlin/com/github/imflog/schema/registry/config/ConfigTask.kt
@@ -0,0 +1,48 @@
+package com.github.imflog.schema.registry.config
+
+import com.github.imflog.schema.registry.RegistryClientWrapper
+import com.github.imflog.schema.registry.SchemaRegistryBasicAuthExtension
+import io.confluent.kafka.schemaregistry.avro.AvroCompatibilityLevel
+import org.gradle.api.DefaultTask
+import org.gradle.api.GradleScriptException
+import org.gradle.api.tasks.Input
+import org.gradle.api.tasks.TaskAction
+
+const val CONFIG_SUBJECTS_TASK = "configSubjectsTask"
+
+open class ConfigTask : DefaultTask() {
+    init {
+        group = "registry"
+        description = "Set subject compatibility in registry"
+    }
+
+    @Input
+    lateinit var url: String
+
+    @Input
+    lateinit var auth: SchemaRegistryBasicAuthExtension
+
+    @Input
+    lateinit var subjects: List<Pair<String, String>>
+
+    @TaskAction
+    fun configureSubjects() {
+        // validate that subject pair includes a valid AvroCompatibilityValue:
+        // can't use the enum directly due to https://youtrack.jetbrains.net/issue/KT-31244
+        subjects.forEach { (_, compatibility) ->
+            try {
+                AvroCompatibilityLevel.valueOf(compatibility)
+            } catch (ex: IllegalArgumentException) {
+                throw GradleScriptException("'$compatibility' is not a valid schema registry compatibility", ex)
+            }
+        }
+
+        val errorCount = ConfigTaskAction(
+            RegistryClientWrapper.client(url, auth),
+            subjects
+        ).run()
+        if (errorCount > 0) {
+            throw GradleScriptException("$errorCount subject configuration not set, see logs for details", Throwable())
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/main/kotlin/com/github/imflog/schema/registry/config/ConfigTaskAction.kt b/src/main/kotlin/com/github/imflog/schema/registry/config/ConfigTaskAction.kt
new file mode 100644
index 0000000..6280378
--- /dev/null
+++ b/src/main/kotlin/com/github/imflog/schema/registry/config/ConfigTaskAction.kt
@@ -0,0 +1,27 @@
+package com.github.imflog.schema.registry.config
+
+import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient
+import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException
+import org.gradle.api.logging.Logging
+
+class ConfigTaskAction(
+    private val client: SchemaRegistryClient,
+    private val subjectPairs: List<Pair<String, String>>
+) {
+
+    private val logger = Logging.getLogger(ConfigTaskAction::class.java)
+
+    fun run() : Int {
+        var errorCount = 0
+        for ((subject, config) in subjectPairs) {
+            logger.debug("$subject: setting config $config")
+            try {
+                client.updateCompatibility(subject, config)
+            } catch (ex: RestClientException) {
+                logger.error("Error during compatibility update for $subject", ex)
+                errorCount++
+            }
+        }
+        return errorCount
+    }
+}
\ No newline at end of file


Test patch:
diff --git a/src/test/kotlin/com/github/imflog/schema/registry/config/ConfigTaskActionTest.kt b/src/test/kotlin/com/github/imflog/schema/registry/config/ConfigTaskActionTest.kt
new file mode 100644
index 0000000..33e0c0f
--- /dev/null
+++ b/src/test/kotlin/com/github/imflog/schema/registry/config/ConfigTaskActionTest.kt
@@ -0,0 +1,31 @@
+package com.github.imflog.schema.registry.config
+
+import io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient
+import org.assertj.core.api.Assertions
+import org.junit.jupiter.api.Test
+
+class ConfigTaskActionTest {
+
+    @Test
+    fun `Should set subject config`() {
+        // given
+        val registryClient = MockSchemaRegistryClient()
+
+        val subjects = listOf(
+            Pair(
+                "test",
+                "FULL_TRANSITIVE"
+            )
+        )
+
+        // when
+        val errorCount = ConfigTaskAction(
+            registryClient,
+            subjects
+        ).run()
+
+        // then
+        Assertions.assertThat(errorCount).isEqualTo(0)
+    }
+
+}
\ No newline at end of file
diff --git a/src/test/kotlin/com/github/imflog/schema/registry/config/ConfigTaskTest.kt b/src/test/kotlin/com/github/imflog/schema/registry/config/ConfigTaskTest.kt
new file mode 100644
index 0000000..620677f
--- /dev/null
+++ b/src/test/kotlin/com/github/imflog/schema/registry/config/ConfigTaskTest.kt
@@ -0,0 +1,222 @@
+package com.github.imflog.schema.registry.config
+
+import com.github.imflog.schema.registry.REGISTRY_FAKE_AUTH_PORT
+import com.github.imflog.schema.registry.REGISTRY_FAKE_PORT
+import com.github.tomakehurst.wiremock.WireMockServer
+import com.github.tomakehurst.wiremock.client.WireMock
+import com.github.tomakehurst.wiremock.common.ConsoleNotifier
+import com.github.tomakehurst.wiremock.core.WireMockConfiguration
+import org.assertj.core.api.Assertions
+import org.gradle.internal.impldep.org.junit.rules.TemporaryFolder
+import org.gradle.testkit.runner.BuildResult
+import org.gradle.testkit.runner.GradleRunner
+import org.gradle.testkit.runner.TaskOutcome
+import org.junit.jupiter.api.AfterAll
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.BeforeAll
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import java.io.File
+
+class ConfigTaskTest {
+    lateinit var folderRule: TemporaryFolder
+
+    lateinit var buildFile: File
+
+    val username: String = "user"
+    val password: String = "pass"
+
+    companion object {
+        lateinit var wiremockServerItem: WireMockServer
+        lateinit var wiremockAuthServerItem: WireMockServer
+
+        @BeforeAll
+        @JvmStatic
+        fun initClass() {
+            wiremockServerItem = WireMockServer(
+                WireMockConfiguration
+                    .wireMockConfig()
+                    .port(REGISTRY_FAKE_PORT)
+                    .notifier(ConsoleNotifier(true))
+            )
+            wiremockServerItem.start()
+
+            wiremockAuthServerItem = WireMockServer(
+                WireMockConfiguration
+                    .wireMockConfig()
+                    .port(REGISTRY_FAKE_AUTH_PORT)
+                    .notifier(ConsoleNotifier(true))
+            )
+            wiremockAuthServerItem.start()
+        }
+
+        @AfterAll
+        @JvmStatic
+        fun tearDown() {
+            wiremockServerItem.stop()
+            wiremockAuthServerItem.stop()
+        }
+    }
+
+    @BeforeEach
+    fun init() {
+        folderRule = TemporaryFolder()
+        // Stub without authentication configuration
+        wiremockServerItem.stubFor(
+            WireMock.put(
+                WireMock
+                    .urlMatching("/config/.*")
+            )
+                .willReturn(
+                    WireMock.aResponse()
+                        .withStatus(200)
+                        .withBody("""{ "compatibility": "FULL_TRANSITIVE" }""")
+                )
+        )
+
+        // Stub with authentication configuration
+        wiremockAuthServerItem.stubFor(
+            WireMock.put(
+                WireMock
+                    .urlMatching("/config/.*")
+            )
+                .withBasicAuth(username, password)
+                .willReturn(
+                    WireMock.aResponse()
+                        .withStatus(200)
+                        .withBody("""{ "compatibility": "FULL_TRANSITIVE" }""")
+                )
+        )
+    }
+
+    @AfterEach
+    internal fun tearDown() {
+        folderRule.delete()
+    }
+
+    @Test
+    fun `ConfigTask should set subject compatibility`() {
+        folderRule.create()
+        buildFile = folderRule.newFile("build.gradle")
+        buildFile.writeText(
+            """
+            plugins {
+                id 'java'
+                id 'com.github.imflog.kafka-schema-registry-gradle-plugin'
+            }
+
+            schemaRegistry {
+                url = 'http://localhost:$REGISTRY_FAKE_AUTH_PORT/'
+                credentials {
+                    username = '$username'
+                    password = '$password'
+                }
+                config {
+                    subject('testSubject1', 'FULL_TRANSITIVE')
+                }
+            }
+            """.trimIndent()
+        )
+
+        val result: BuildResult? = GradleRunner.create()
+            .withGradleVersion("4.9")
+            .withProjectDir(folderRule.root)
+            .withArguments(CONFIG_SUBJECTS_TASK)
+            .withPluginClasspath()
+            .withDebug(true)
+            .build()
+        Assertions.assertThat(result?.task(":configSubjectsTask")?.outcome).isEqualTo(TaskOutcome.SUCCESS)
+
+    }
+
+    @Test
+    fun `ConfigTask should set subject compatibility without credentials`() {
+        folderRule.create()
+        buildFile = folderRule.newFile("build.gradle")
+        buildFile.writeText(
+            """
+            plugins {
+                id 'java'
+                id 'com.github.imflog.kafka-schema-registry-gradle-plugin'
+            }
+
+            schemaRegistry {
+                url = 'http://localhost:$REGISTRY_FAKE_PORT/'
+                config {
+                    subject('testSubject1', 'FULL_TRANSITIVE')
+                }
+            }
+            """.trimIndent()
+        )
+
+        val result: BuildResult? = GradleRunner.create()
+            .withGradleVersion("4.9")
+            .withProjectDir(folderRule.root)
+            .withArguments(CONFIG_SUBJECTS_TASK)
+            .withPluginClasspath()
+            .withDebug(true)
+            .build()
+        Assertions.assertThat(result?.task(":configSubjectsTask")?.outcome).isEqualTo(TaskOutcome.SUCCESS)
+    }
+
+    @Test
+    fun `ConfigTask should fail to set subject compatibility without credentials when required`() {
+        folderRule.create()
+        buildFile = folderRule.newFile("build.gradle")
+        buildFile.writeText(
+            """
+            plugins {
+                id 'java'
+                id 'com.github.imflog.kafka-schema-registry-gradle-plugin'
+            }
+
+            schemaRegistry {
+                url = 'http://localhost:$REGISTRY_FAKE_AUTH_PORT/'
+                config {
+                    subject('testSubject1', 'FULL_TRANSITIVE')
+                }
+            }
+            """.trimIndent()
+        )
+
+        val result: BuildResult? = GradleRunner.create()
+            .withGradleVersion("4.9")
+            .withProjectDir(folderRule.root)
+            .withArguments(CONFIG_SUBJECTS_TASK)
+            .withPluginClasspath()
+            .withDebug(true)
+            .buildAndFail()
+        Assertions.assertThat(result?.task(":configSubjectsTask")?.outcome).isEqualTo(TaskOutcome.FAILED)
+    }
+
+    @Test
+    fun `ConfigTask should detect and reject invalid compatibility settings`() {
+        folderRule.create()
+        buildFile = folderRule.newFile("build.gradle")
+        buildFile.writeText(
+            """
+            plugins {
+                id 'java'
+                id 'com.github.imflog.kafka-schema-registry-gradle-plugin'
+            }
+
+            schemaRegistry {
+                url = 'http://localhost:$REGISTRY_FAKE_PORT/'
+                config {
+                    subject('testSubject1', 'FULL_TRANSITIVE')
+                    subject('testSubject2', 'FUL_TRANSITIVE') // intentionally broken
+                }
+            }
+            """.trimIndent()
+        )
+
+        val result: BuildResult? = GradleRunner.create()
+            .withGradleVersion("4.9")
+            .withProjectDir(folderRule.root)
+            .withArguments(CONFIG_SUBJECTS_TASK)
+            .withPluginClasspath()
+            .withDebug(true)
+            .buildAndFail()
+        Assertions.assertThat(result?.task(":configSubjectsTask")?.outcome).isEqualTo(TaskOutcome.FAILED)
+    }
+}
\ No newline at end of file