Code examples for upcoming data contracts blog (#285)

* Example code for data-contracts blog.

* Ignoring gradle wrapper stuff.

* Ignoring gradle wrapper stuff.

* update setup script.

* Updates to README, adding console examples.

* Image updates on documentation.

* Shoutout Gilles.

* Formatting of readme elements, like notes.

* Disclaimer on setup script, how long it takes to execute.

Author: sandonjacobs

data-contracts/.gitignore

@@ -0,0 +1,27 @@
+### Gradle template
+**/.gradle/
+**/gradle/
+**/.gradlew.bat
+**/.gradlew
+
+**/build/
+
+# Ignore Gradle GUI config
+gradle-app.setting
+
+# Avoid ignoring Gradle wrapper jar file (.jar files are usually ignored)
+!**/gradle-wrapper.jar
+# Avoid ignore Gradle wrappper properties
+!**/gradle-wrapper.properties
+
+# Cache of project
+.gradletasknamecache
+
+# Eclipse Gradle plugin generated files
+# Eclipse Core
+.project
+# JDT-specific (Eclipse Java Development Tools)
+.classpath
+
+.idea/
+

data-contracts/README.md

@@ -0,0 +1,130 @@
+# Managing Data Contracts in Confluent Cloud
+
+Data contracts consist not only of the schemas to define the structure of events, but also rulesets allowing for more fine-grained validations,
+controls, and discovery. In this demo, we'll evolve a schema by adding migration rules.
+
+I'd like to credit my colleague Gilles Philippart's work as the basis for the schemas and rules in this demo. My goal here is to add examples with commonly used
+developer tools and practices to his work in this area, which can be found [here in GitHub](https://github.com/gphilipp/migration-rules-demo).
+
+## Running the Example
+
+<img src="./images/overview.png" width="1000" height="600">
+
+In the workflow above, we see these tools in action:
+* The [Confluent Terraform Provider](https://registry.terraform.io/providers/confluentinc/confluent/latest/docs) is used to define Confluent Cloud assets (Kafka cluster(s), Data Governance, Kafka Topics, and Schema Configurations).
+* Using the newly created Schema Registry, data engineers and architects define the schema of the events that comprise the organization's canonical data model - i.e. entities, events, and commands that are shared across applications. - along with other parts of the data contract. This includes data quality rules, metadata, and migration rules. A gradle plugin is utilized to register the schemas and related elements of the data contract with the Schema Registry.
+* Applications which producer and/or consume these event types can download the schemas from the Schema Registry. In our example, this is a JVM application built using Gradle. A gradle plugin is used to download the schemas, after which another gradle plugin is used to generate Java classes from those schemas - thus providing the application with compile-time type safety.
+
+### Prerequisites
+
+Clone the `confluentinc/demo-scene` GitHub repository (if you haven't already) and navigate to the `demo-scene` directory:
+
+```shell
+git clone [email protected]:confluentinc/demo-scene.git
+cd demo-scene
+```
+
+Here are the tools needed to run this tutorial:
+* [Confluent Cloud](http://confluent.cloud)
+* [Confluent CLI](https://docs.confluent.io/confluent-cli/current/install.html)
+* [Terraform](https://developer.hashicorp.com/terraform/install?product_intent=terraform)
+* [jq](https://jqlang.github.io/jq/)
+* [Gradle](https://gradle.org/install/) (version 8.5 or higher)
+* JDK 17
+* IDE of choice
+
+> When installing and configuring the Confluent CLI, include the Confluent Cloud credentials as environment variables for future use. For instance with bash or zsh, include these export statements:
+> 
+> ```shell
+> export CONFLUENT_CLOUD_API_KEY=<API KEY>
+> export CONFLUENT_CLOUD_API_SECRET<API SECRET>
+> ```
+>
+
+Terraform can use the value of any environment variable whose name begins with `TF_VAR_` as the value of a terraform variable of the same name. For more on this functionality, see the [terraform documentation](https://developer.hashicorp.com/terraform/cli/config/environment-variables#tf_var_name).
+
+Our example requires we set the value the `org_id` variable from Confluent Cloud. This denotes which organization will house the Confluent Cloud assets we are creating. So let's export the Confluent Cloud organization ID to a terraform environment variable.
+
+This command may open a browser window asking you to authenticate to Confluent Cloud. Once that's complete, the result of
+`confluent organization list` is queried by `jq` to extract the `id` of the current organization to which you are authenticated:
+
+```shell
+export TF_VAR_org_id=$(confluent organization list -o json | jq -c -r '.[] | select(.is_current)' | jq '.id')
+```
+
+### Create Assets in Confluent Cloud
+
+From the root of `data-contracts`, run the provided setup script - `setup.sh`. You may need to edit permissions to make this file executable.
+
+Because this script is provisioning multiple assets in Confluent Cloud and downloading the appropriate gradle-wrapper JAR files for each module,
+it may take a few minutes to complete.
+
+```shell
+chmod +x setup.sh
+./setup.sh
+Setup Confluent Cloud and all Gradle Builds for Demo
+-------------------------------------
+Initializing Confluent Cloud Environment...
+......
+......
+......
+......
+BUILD SUCCESSFUL in 2s
+6 actionable tasks: 6 executed
+Watched directory hierarchies: [/Users/sjacobs/code/confluentinc/demo-scene/data-contracts]
+Code Generation Complete
+-------------------------------------
+
+Setup Complete!
+```
+
+Let's have a look at what we've created in Confluent Cloud, we find a new Environment:
+
+![CC Environment](./images/environment.png)
+
+With a Kafka cluster:
+
+![Kafka Cluster](./images/cluster.png)
+
+And Data Contracts:
+
+![Data Contracts](./images/schemas.png)
+
+
+Locally, we also create a `properties` file containing the parameters needed for our Kafka clients to connect to Confluent Cloud. For an example of this 
+`properties` file, see [confluent.properties.orig](shared/src/main/resources/confluent.properties.orig).
+
+> **NOTE**: The file-based approach we're using here is NOT recommended for a production-quality application. Perhaps a secrets manager implementation would be better suited - which the major cloud providers all offer, or perhaps a tool like Hashicorp Vault. Such a tool would also have client libraries in a Maven repository for the JVM applications to access the secrets.
+> 
+
+### Run the Examples
+
+In `app-schema-v1`, the `ApplicationMain` object's `main` function starts a consumer in a new thread to subscribe to the `membership-avro` topic. It then begins
+producing randomly-generated events to `membership-avro` at a provided interval for a provided duration. By default, an event is produced every 1 second for 100 seconds. These events are created and consumed using version 1 of the membership schema. The console output of the consumer should look something like this:
+
+```shell
+[Thread-0] INFO io.confluent.devrel.datacontracts.shared.BaseConsumer - Received Membership d0e65c83-b1c5-451d-b08b-8d1ed6fca8d6, {"user_id": "d0e65c83-b1c5-451d-b08b-8d1ed6fca8d6", "start_date": "2023-01-14", "end_date": "2025-05-28"}
+[Thread-0] INFO io.confluent.devrel.datacontracts.shared.BaseConsumer - Received Membership 940cf6fa-eb12-46af-87e8-5a9bc33df119, {"user_id": "940cf6fa-eb12-46af-87e8-5a9bc33df119", "start_date": "2023-05-23", "end_date": "2025-07-02"}
+```
+
+The `app-schema-v2` module's `main` function starts a consumer subscribed to the `membership-avro` topic. But this time, events will be consumed using
+version 2 of the membership schema. Notice the `Map` of consumer overrides in the constructor of the `MembershipConsumer` in that module. As such, those
+same events which `app-schema-v1` produced using `major_version=1` of the schema are consumed using `major_version=2`:
+
+```shell
+[Thread-0] INFO io.confluent.devrel.datacontracts.shared.BaseConsumer - v2 - Received Membership b0e34c68-208c-4771-be19-79689fc9ad28, {"user_id": "b0e34c68-208c-4771-be19-79689fc9ad28", "validity_period": {"from": "2022-11-06", "to": "2025-09-03"}}
+[Thread-0] INFO io.confluent.devrel.datacontracts.shared.BaseConsumer - v2 - Received Membership 517f8c7e-4ae5-47ea-93a2-c1f00669d330, {"user_id": "517f8c7e-4ae5-47ea-93a2-c1f00669d330", "validity_period": {"from": "2023-01-29", "to": "2025-02-19"}}
+```
+
+This illustrates how the use of migration rules allows producers and consumers to independently make any code and configuration changes needed to accommodate schema changes.
+
+## Teardown
+
+When you're done with the demo, issue this command from the `cc-terraform` directory to destroy the Confluent Cloud environment
+we created:
+
+```shell
+terraform destroy -auto-approve
+```
+
+Check the Confluent Cloud console to ensure this environment no longer exists.

data-contracts/app-schema-v1/.gitignore

@@ -0,0 +1,4 @@
+/gradle/
+/.gradle/
+/gradlew
+/gradlew.bat

data-contracts/app-schema-v1/build.gradle.kts

@@ -0,0 +1,94 @@
+import java.io.FileInputStream
+import java.util.Properties
+
+buildscript {
+    repositories {
+        mavenCentral()
+        maven("https://packages.confluent.io/maven/")
+        maven("https://jitpack.io")
+        gradlePluginPortal()
+    }
+}
+
+plugins {
+    kotlin("jvm") version "2.0.21"
+    id("com.google.protobuf") version "0.9.4"
+    id("com.github.imflog.kafka-schema-registry-gradle-plugin") version "2.1.0"
+    id("com.bakdata.avro") version "1.2.1"
+}
+
+group = "io.confluent.devrel"
+
+repositories {
+    mavenLocal()
+    mavenCentral()
+    maven("https://packages.confluent.io/maven/")
+    maven("https://jitpack.io")
+}
+
+sourceSets {
+    main {
+        kotlin.srcDirs("src/main/kotlin", "build/generated-main-avro-java")
+    }
+}
+
+dependencies {
+    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.10.1")
+    implementation("io.confluent.devrel:data-contracts-shared:+")
+
+    implementation("org.apache.kafka:kafka-clients:${project.property("kafkaVersion")}")
+    implementation("io.confluent:kafka-avro-serializer:${project.property("confluentVersion")}")
+
+    implementation("io.github.serpro69:kotlin-faker:${project.property("fakerVersion")}")
+    implementation("io.github.serpro69:kotlin-faker-books:${project.property("fakerVersion")}")
+    implementation("io.github.serpro69:kotlin-faker-tech:${project.property("fakerVersion")}")
+
+    implementation("org.jetbrains.kotlinx:kotlinx-cli:0.3.6")
+    implementation("org.jetbrains.kotlinx:kotlinx-datetime:0.6.1")
+}
+
+kotlin {
+    jvmToolchain(17)
+}
+
+val schemaRegOutputDir = "${project.projectDir.absolutePath}/build/schema-registry-plugin"
+
+tasks.downloadSchemasTask {
+    doFirst {
+        mkdir(schemaRegOutputDir)
+    }
+}
+
+schemaRegistry {
+    val srProperties = Properties()
+    // At the moment, this is a file with which we are LOCALLY aware.
+    // In an ACTUAL CI/CD workflow, this would be externalized, perhaps provided from a base build image or other parameter.
+    srProperties.load(FileInputStream(File("${project.projectDir.absolutePath}/../shared/src/main/resources/confluent.properties")))
+
+    url = srProperties.getProperty("schema.registry.url")
+
+    val srCredTokens = srProperties.get("basic.auth.user.info").toString().split(":")
+    credentials {
+        username = srCredTokens[0]
+        password = srCredTokens[1]
+    }
+    outputDirectory = "${System.getProperty("user.home")}/tmp/schema-registry-plugin"
+    pretty = true
+
+    download {
+        // download the membership avro schema, version 1
+        subject("membership-avro-value", "${projectDir}/src/main/avro", 1)
+    }
+}
+
+tasks.clean {
+    doFirst {
+        delete(fileTree("${projectDir}/src/main/avro/").include("**/*.avsc"))
+    }
+}
+
+tasks.register("generateCode") {
+    group = "source generation"
+    description = "wrapper task for all source generation"
+    dependsOn("downloadSchemasTask", "generateAvroJava", "generateProto")
+}

data-contracts/app-schema-v1/gradle.properties

@@ -0,0 +1,6 @@
+confluentVersion=7.8.0
+fakerVersion=2.0.0-rc.3
+grpcVersion=1.15.1
+kafkaVersion=3.8.0
+protobufVersion=3.6.1
+slf4jVersion=2.0.11

data-contracts/app-schema-v1/settings.gradle.kts

@@ -0,0 +1 @@
+rootProject.name="app-schema-v1"

data-contracts/app-schema-v1/src/main/avro/.gitignore

@@ -0,0 +1 @@
+*.avsc

data-contracts/app-schema-v1/src/main/kotlin/io/confluent/devrel/dc/v1/ApplicationMain.kt

@@ -0,0 +1,72 @@
+package io.confluent.devrel.dc.v1
+
+import io.confluent.devrel.Membership
+import io.confluent.devrel.dc.v1.kafka.MembershipConsumer
+import io.confluent.devrel.dc.v1.kafka.MembershipProducer
+import kotlinx.cli.ArgParser
+import kotlinx.cli.ArgType
+import kotlinx.cli.default
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import kotlinx.datetime.Clock
+import java.time.LocalDate
+import java.util.*
+import kotlin.concurrent.thread
+import kotlin.random.Random
+import kotlin.time.DurationUnit
+import kotlin.time.toDuration
+
+class ApplicationMain {
+
+    companion object {
+
+
+        @JvmStatic
+        fun main(args: Array<String>) {
+            runBlocking {
+                println("Starting application main...")
+                println(args.joinToString(" "))
+                val parser = ArgParser("schema-v1")
+
+                val interval by parser.option(ArgType.Int,
+                    shortName = "i", fullName = "interval",
+                    description = "message send interval, seconds")
+                    .default(1)
+                val duration by parser.option(ArgType.Int,
+                    shortName = "d", fullName = "duration",
+                    description = "how long to run, seconds")
+                    .default(100)
+                parser.parse(args)
+
+                val messageInterval = interval.toDuration(DurationUnit.SECONDS)
+                val sendDuration = duration.toDuration(DurationUnit.SECONDS)
+
+                val producer = MembershipProducer()
+                val consumer = MembershipConsumer()
+
+                thread {
+                    consumer.start(listOf("membership-avro"))
+                }
+
+                coroutineScope {
+                    launch {
+                        val until = Clock.System.now().plus(sendDuration)
+                        while(Clock.System.now().compareTo(until) < 0) {
+                            val userId = UUID.randomUUID().toString()
+                            val membership = Membership.newBuilder()
+                                .setUserId(userId)
+                                .setStartDate(LocalDate.now().minusDays(Random.nextLong(100, 1000)))
+                                .setEndDate(LocalDate.now().plusWeeks(Random.nextLong(1, 52)))
+                                .build()
+                            producer.send("membership-avro", userId, membership)
+                            delay(messageInterval.inWholeSeconds)
+                        }
+                    }
+                }
+                producer.close()
+            }
+        }
+    }
+}

data-contracts/app-schema-v1/src/main/kotlin/io/confluent/devrel/dc/v1/kafka/MembershipConsumer.kt

@@ -0,0 +1,24 @@
+package io.confluent.devrel.dc.v1.kafka
+
+import io.confluent.devrel.Membership
+import io.confluent.devrel.datacontracts.shared.BaseConsumer
+import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig
+import org.apache.kafka.clients.consumer.ConsumerConfig
+
+class MembershipConsumer: BaseConsumer<String, Membership>(mapOf(
+    ConsumerConfig.GROUP_ID_CONFIG to "app-schema-v1",
+    ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to "earliest",
+    ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG to "org.apache.kafka.common.serialization.StringDeserializer",
+    ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG to "io.confluent.kafka.serializers.KafkaAvroDeserializer",
+    AbstractKafkaSchemaSerDeConfig.LATEST_COMPATIBILITY_STRICT to true,
+    AbstractKafkaSchemaSerDeConfig.USE_LATEST_VERSION to false,
+    AbstractKafkaSchemaSerDeConfig.USE_LATEST_WITH_METADATA to "major_version=1"
+)) {
+
+    override fun consumeRecord(
+        key: String,
+        value: Membership
+    ) {
+        logger.info("Received Membership ${key}, ${value}")
+    }
+}

data-contracts/app-schema-v1/src/main/kotlin/io/confluent/devrel/dc/v1/kafka/MembershipProducer.kt

@@ -0,0 +1,15 @@
+package io.confluent.devrel.dc.v1.kafka
+
+import io.confluent.devrel.Membership
+import io.confluent.devrel.datacontracts.shared.BaseProducer
+import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig
+import org.apache.kafka.clients.producer.ProducerConfig
+
+class MembershipProducer: BaseProducer<String, Membership>(mapOf(
+    ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG to "org.apache.kafka.common.serialization.StringSerializer",
+    ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG to "io.confluent.kafka.serializers.KafkaAvroSerializer",
+    ProducerConfig.CLIENT_ID_CONFIG to "membership-producer-app-v1",
+    AbstractKafkaSchemaSerDeConfig.LATEST_COMPATIBILITY_STRICT to true,
+    AbstractKafkaSchemaSerDeConfig.USE_LATEST_VERSION to false,
+    AbstractKafkaSchemaSerDeConfig.USE_LATEST_WITH_METADATA to "major_version=1"
+))