-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Deafault to ByteArraySerializer when schema.registry.url is not set (#38
) * serialize to byte when no schema registry is present * update to distribute only tar * update changelog * ktlint fixes * use unique names for registered endpoints * ktlint
- Loading branch information
Showing
17 changed files
with
488 additions
and
84 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -40,7 +40,7 @@ jobs: | |
| - name: Build artifacts | ||
| uses: gradle/[email protected] | ||
| with: | ||
| arguments: assembleDist | ||
| arguments: distTar | ||
| - name: Upload artifacts | ||
| uses: ncipollo/release-action@v1 | ||
| with: | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
33 changes: 33 additions & 0 deletions
33
src/main/kotlin/io/provenance/abci/listener/ProducerSettings.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| package io.provenance.abci.listener | ||
|
|
||
| import com.typesafe.config.Config | ||
| import org.apache.kafka.clients.producer.KafkaProducer | ||
| import org.apache.kafka.clients.producer.Producer | ||
| import org.apache.kafka.clients.producer.ProducerConfig | ||
| import org.apache.kafka.common.serialization.Serializer | ||
|
|
||
| class ProducerSettings<K, V>( | ||
| private val config: Config, | ||
| private val keySerializer: Serializer<K>? = null, | ||
| private val valueSerializer: Serializer<V>? = null | ||
| ) { | ||
|
|
||
| fun createKafkaProducer(): Producer<K, V> { | ||
| val properties = config.getConfig("kafka-clients").toProperties() | ||
| require( | ||
| keySerializer != null || | ||
| properties.getProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG).isNotEmpty() | ||
| ) { | ||
| "Key serializer should be defined or declared in configuration" | ||
| } | ||
|
|
||
| require( | ||
| valueSerializer != null || | ||
| properties.getProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG).isNotEmpty() | ||
| ) { | ||
| "Value serializer should be defined or declared in configuration" | ||
| } | ||
|
|
||
| return KafkaProducer(properties) | ||
| } | ||
| } |
78 changes: 78 additions & 0 deletions
78
src/main/kotlin/io/provenance/abci/listener/ProtobufSerializer.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,78 @@ | ||
| package io.provenance.abci.listener | ||
|
|
||
| import com.google.protobuf.Message | ||
| import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig | ||
| import io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer | ||
| import org.apache.kafka.common.serialization.ByteArraySerializer | ||
| import org.apache.kafka.common.serialization.Serializer | ||
| import java.io.ByteArrayOutputStream | ||
|
|
||
| /** | ||
| * ProtobufSerializer is a custom Protobuf serializer for Kafka | ||
| * that chooses a serialization path based on whether the application | ||
| * is configured to use the Confluent Schema Registry or not. | ||
| * | ||
| * When the application is configured to use the Confluent Schema Registry | ||
| * the [KafkaProtobufSerializer] will be used. Otherwise, the [ByteArraySerializer] will be used. | ||
| * | ||
| * In order to enable the application to use the Confluent Schema Registry, | ||
| * you MUST set the following properties: | ||
| * | ||
| * `schema.registry.url={{ SR_URL }}` | ||
| * `value.serializer=io.provenance.abci.listener.ProtobufSerializer`. | ||
| * | ||
| * Example configuration: | ||
| * | ||
| * ``` | ||
| * # application.conf | ||
| * | ||
| * # application configuration properties | ||
| * ... | ||
| * | ||
| * kafka.producer { | ||
| * ... | ||
| * # Properties defined by org.apache.kafka.clients.producer.ProducerConfig. | ||
| * # can be defined in this configuration section. | ||
| * kafka-clients { | ||
| * bootstrap.servers = "localhost:9092" | ||
| * // other producer properties | ||
| * ... | ||
| * key.serializer = org.apache.kafka.common.serialization.StringSerializer | ||
| * value.serializer = io.provenance.abci.listener.ProtobufSerializer | ||
| * schema.registry.url = "http://127.0.0.1:8081" | ||
| * } | ||
| * } | ||
| * ``` | ||
| * For additional producer properties see [org.apache.kafka.clients.producer.ProducerConfig] | ||
| * | ||
| */ | ||
| class ProtobufSerializer<T : Message> : Serializer<T> { | ||
| private lateinit var serializer: Serializer<T> | ||
|
|
||
| override fun configure(configs: MutableMap<String, *>?, isKey: Boolean) { | ||
| val useSchemaRegistry = configs!!.containsKey(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG) | ||
| if (useSchemaRegistry) { | ||
| val serializer = KafkaProtobufSerializer<T>() | ||
| serializer.configure(configs, isKey) | ||
| this.serializer = serializer | ||
| } else { | ||
| val serializer = ByteArraySerializer() | ||
| @Suppress("UNCHECKED_CAST") | ||
| this.serializer = serializer as Serializer<T> | ||
| } | ||
| } | ||
|
|
||
| override fun serialize(topic: String?, data: T): ByteArray { | ||
| var bytes: ByteArray = byteArrayOf() | ||
| when (serializer) { | ||
| is KafkaProtobufSerializer -> bytes = serializer.serialize(topic, data) | ||
| is ByteArraySerializer -> { | ||
| val out = ByteArrayOutputStream() | ||
| data.writeTo(out) | ||
| bytes = out.toByteArray() | ||
| out.close() | ||
| } | ||
| } | ||
| return bytes | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.