This repository has been archived by the owner on Apr 12, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 9
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* WIP: Producer * Validation sample * Fix avro array and review UI * Improve UI * Add stringProducer * Fix string consumer * Move objectMapper and avroParser to DI * Remove generic throwables * Fix tests * Improve ProducerViewModelTest * Add tests StringProducerTest * Add AvroProducerTest * Improve JsonToAvroConverterTest * Small improvement to hints * Improve ProducerViewModelTest * Add happy path to ListTopicViewModelTest
- Loading branch information
1 parent
cb0c929
commit a8a035b
Showing
18 changed files
with
1,098 additions
and
15 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
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 |
|---|---|---|
| @@ -1,24 +1,35 @@ | ||
| package insulator.di | ||
|
|
||
| import com.fasterxml.jackson.databind.ObjectMapper | ||
| import insulator.lib.configuration.ConfigurationRepo | ||
| import insulator.lib.configuration.model.Cluster | ||
| import insulator.lib.jsonhelper.JsonFormatter | ||
| import insulator.lib.jsonhelper.JsonToAvroConverter | ||
| import insulator.lib.kafka.AdminApi | ||
| import insulator.lib.kafka.AvroProducer | ||
| import insulator.lib.kafka.Consumer | ||
| import insulator.lib.kafka.SchemaRegistry | ||
| import insulator.lib.kafka.StringProducer | ||
| import kotlinx.serialization.json.Json | ||
| import org.apache.avro.Schema | ||
| import org.koin.core.qualifier.named | ||
| import org.koin.dsl.module | ||
|
|
||
| val libModule = module { | ||
|
|
||
| // Configurations | ||
| // Configurations and helpers | ||
| single { Json {} } | ||
| single { ConfigurationRepo(get()) } | ||
| single { JsonFormatter(get()) } | ||
| single { Schema.Parser() } | ||
| single { ObjectMapper() } | ||
| single { JsonToAvroConverter(get()) } | ||
|
|
||
| scope<Cluster> { | ||
| factory { AdminApi(get(), get()) } | ||
| factory { Consumer(get()) } | ||
| factory { AvroProducer(get(named("avroProducer")), get(), get()) } | ||
| factory { StringProducer(get()) } | ||
| factory { SchemaRegistry(get()) } | ||
| } | ||
| } |
135 changes: 135 additions & 0 deletions
135
src/main/kotlin/insulator/lib/jsonhelper/JsonToAvroConverter.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,135 @@ | ||
| package insulator.lib.jsonhelper | ||
|
|
||
| import arrow.core.Either | ||
| import arrow.core.left | ||
| import arrow.core.right | ||
| import com.fasterxml.jackson.core.JsonParseException | ||
| import com.fasterxml.jackson.core.JsonProcessingException | ||
| import com.fasterxml.jackson.databind.ObjectMapper | ||
| import org.apache.avro.AvroRuntimeException | ||
| import org.apache.avro.Conversions | ||
| import org.apache.avro.Schema | ||
| import org.apache.avro.Schema.Parser | ||
| import org.apache.avro.SchemaParseException | ||
| import org.apache.avro.generic.GenericData | ||
| import org.apache.avro.generic.GenericRecord | ||
| import org.apache.avro.generic.GenericRecordBuilder | ||
| import java.nio.ByteBuffer | ||
| import javax.xml.bind.DatatypeConverter | ||
|
|
||
| class JsonToAvroConverter(private val objectMapper: ObjectMapper) { | ||
|
|
||
| fun convert(jsonString: String, schemaString: String): Either<Throwable, GenericRecord> { | ||
| return try { | ||
| val jsonMap = objectMapper.readValue(jsonString, Map::class.java) | ||
| val schema = Parser().parse(schemaString) | ||
| val parsed = parseRecord(schema, jsonMap) | ||
| if (GenericData().validate(schema, parsed)) { | ||
| parsed.right() | ||
| } else JsonToAvroException("Generated record is invalid, check the schema").left() | ||
| } catch (jsonException: SchemaParseException) { | ||
| InvalidSchemaException().left() | ||
| } catch (jsonException: JsonParseException) { | ||
| InvalidJsonException().left() | ||
| } catch (jsonException: JsonProcessingException) { | ||
| InvalidJsonException().left() | ||
| } catch (jsonToAvroException: JsonToAvroException) { | ||
| jsonToAvroException.left() | ||
| } catch (avroRuntime: AvroRuntimeException) { | ||
| JsonToAvroException(avroRuntime.message).left() | ||
| } | ||
| } | ||
|
|
||
| private fun parseRecord(schema: Schema, jsonMap: Map<*, *>?): GenericRecord { | ||
| if (schema.type != Schema.Type.RECORD || jsonMap == null) | ||
| throw JsonToAvroException("Expecting record ${schema.name}") | ||
| val recordBuilder = GenericRecordBuilder(schema) | ||
| schema.fields.forEach { fieldSchema -> | ||
| val fieldName = fieldSchema.name() | ||
| if (fieldName !in jsonMap) throw JsonToAvroException("Expecting \"${fieldName}\" with type ${printType(fieldSchema.schema())}", fieldName) | ||
| val jsonValue = jsonMap[fieldName] | ||
|
|
||
| recordBuilder.set(fieldSchema, parseField(fieldSchema.schema(), jsonValue)) | ||
| } | ||
| return recordBuilder.build() | ||
| } | ||
|
|
||
| private fun printType(schema: Schema): String { | ||
| return when (schema.type) { | ||
| Schema.Type.NULL -> "null" | ||
| Schema.Type.RECORD -> "record \"${schema.name}\"" | ||
| Schema.Type.BYTES -> "bytes (eg \"0x00\")" | ||
| Schema.Type.ENUM -> "enum [${schema.enumSymbols.joinToString(", ")}]" | ||
| Schema.Type.UNION -> "union [${schema.types.joinToString(", ") { it.name }}]" | ||
| Schema.Type.ARRAY -> "array of \"${schema.elementType.name}\"" | ||
| else -> schema.type.name.toLowerCase() | ||
| } | ||
| } | ||
|
|
||
| private fun parseField(fieldSchema: Schema, jsonValue: Any?): Any? { | ||
| return when (fieldSchema.type) { | ||
| Schema.Type.NULL -> if (jsonValue == null) null else throw JsonToAvroException("$jsonValue should be ${printType(fieldSchema)}") | ||
| Schema.Type.RECORD -> parseRecord(fieldSchema, jsonValue as? Map<*, *>) | ||
| Schema.Type.FLOAT -> parseFloat(fieldSchema, jsonValue) | ||
| Schema.Type.BYTES -> parseBytes(fieldSchema, jsonValue) | ||
| Schema.Type.ENUM -> parseEnum(fieldSchema, jsonValue) | ||
| Schema.Type.UNION -> parseUnion(fieldSchema, jsonValue) | ||
| Schema.Type.ARRAY -> parseArray(fieldSchema, jsonValue) | ||
| Schema.Type.LONG -> parseLong(jsonValue) | ||
| else -> jsonValue | ||
| } | ||
| } | ||
|
|
||
| private fun parseLong(jsonValue: Any?) = | ||
| when (jsonValue) { | ||
| is Long -> jsonValue | ||
| is Int -> jsonValue.toLong() | ||
| else -> throw JsonToAvroException("Expecting long but got ${jsonValue?.javaClass?.simpleName}") | ||
| } | ||
|
|
||
| private fun parseArray(fieldSchema: Schema, jsonValue: Any?): Any { | ||
| if (jsonValue !is ArrayList<*>) throw JsonToAvroException("Expecting ${printType(fieldSchema)} but got $jsonValue") | ||
| return jsonValue.map { parseField(fieldSchema.elementType, it) }.toList() | ||
| } | ||
|
|
||
| private fun parseUnion(fieldSchema: Schema, jsonValue: Any?): Any? { | ||
| fieldSchema.types.forEach { | ||
| val parsed = kotlin.runCatching { parseField(it, jsonValue) } | ||
| if (parsed.isSuccess) return parsed.getOrNull() | ||
| } | ||
| throw JsonToAvroException("Expecting \"${fieldSchema.fields.first().name()}\" with type Union [${fieldSchema.types.joinToString(", ") { it.name }}]") | ||
| } | ||
|
|
||
| private fun parseEnum(fieldSchema: Schema, jsonValue: Any?): GenericData.EnumSymbol { | ||
| val symbols = fieldSchema.enumSymbols | ||
| return if (jsonValue == null || jsonValue.toString() !in symbols) | ||
| throw JsonToAvroException("Expecting ${printType(fieldSchema)} but got $jsonValue") | ||
| else GenericData.EnumSymbol(fieldSchema, jsonValue) | ||
| } | ||
|
|
||
| private fun parseBytes(fieldSchema: Schema, jsonValue: Any?): ByteBuffer? { | ||
| if (jsonValue == null) throw JsonToAvroException("Expecting ${printType(fieldSchema)} but got \"${jsonValue}\"") | ||
| if (jsonValue is Double && fieldSchema.logicalType.name == "decimal") | ||
| return Conversions.DecimalConversion().runCatching { | ||
| toBytes(jsonValue.toBigDecimal(), fieldSchema, fieldSchema.logicalType) | ||
| }.fold({ it }, { throw JsonToAvroException("Invalid $jsonValue ${it.message}") }) | ||
| return when (jsonValue) { | ||
| null -> null | ||
| is String -> | ||
| if (!jsonValue.toLowerCase().startsWith("0x")) throw JsonToAvroException("Invalid $jsonValue, BYTES value need to start with 0x") | ||
| else ByteBuffer.wrap(DatatypeConverter.parseHexBinary(jsonValue.substring(2))) | ||
| else -> throw JsonToAvroException("Expecting binary but got $jsonValue") | ||
| } | ||
| } | ||
|
|
||
| private fun parseFloat(fieldSchema: Schema, jsonValue: Any?) = | ||
| when (jsonValue) { | ||
| is Double -> jsonValue.toFloat() | ||
| is Float -> jsonValue | ||
| else -> throw JsonToAvroException("Expecting ${printType(fieldSchema)} but got $jsonValue") | ||
| } | ||
| } | ||
|
|
||
| class JsonToAvroException(message: String?, val nextField: String? = null) : Throwable(message) | ||
| class InvalidJsonException(message: String? = null) : Throwable(message) | ||
| class InvalidSchemaException(message: String? = null) : Throwable(message) |
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,53 @@ | ||
| package insulator.lib.kafka | ||
|
|
||
| import arrow.core.Either | ||
| import arrow.core.flatMap | ||
| import arrow.core.left | ||
| import arrow.core.right | ||
| import insulator.lib.jsonhelper.JsonToAvroConverter | ||
| import org.apache.avro.generic.GenericRecord | ||
| import org.apache.kafka.clients.producer.ProducerRecord | ||
| import org.apache.kafka.clients.producer.Producer as KafkaProducer | ||
|
|
||
| interface Producer { | ||
| fun validate(value: String, topic: String): Either<Throwable, Unit> | ||
| fun send(topic: String, key: String, value: String): Either<Throwable, Unit> | ||
| } | ||
|
|
||
| class AvroProducer( | ||
| private val avroProducer: KafkaProducer<String, GenericRecord>, | ||
| private val schemaRegistry: SchemaRegistry, | ||
| private val jsonAvroConverter: JsonToAvroConverter | ||
| ) : Producer { | ||
|
|
||
| private val schemaCache = HashMap<String, Either<Throwable, String>>() | ||
|
|
||
| override fun validate(value: String, topic: String) = | ||
| internalValidate(value, topic).flatMap { Unit.right() } | ||
|
|
||
| override fun send(topic: String, key: String, value: String) = | ||
| internalValidate(value, topic) | ||
| .map { ProducerRecord(topic, key, it) } | ||
| .flatMap { avroProducer.runCatching { send(it) }.fold({ Unit.right() }, { it.left() }) } | ||
|
|
||
| private fun internalValidate(value: String, topic: String) = | ||
| getCachedSchema(topic).flatMap { jsonAvroConverter.convert(value, it) } | ||
|
|
||
| private fun getCachedSchema(topic: String) = | ||
| schemaCache.getOrPut( | ||
| topic, | ||
| { | ||
| schemaRegistry.getSubject("$topic-value") | ||
| .map { it.schemas.maxByOrNull { s -> s.version }?.schema } | ||
| .flatMap { it?.right() ?: Throwable("Schema not found").left() } | ||
| } | ||
| ) | ||
| } | ||
|
|
||
| class StringProducer(private val stringProducer: KafkaProducer<String, String>) : Producer { | ||
| override fun validate(value: String, topic: String) = Unit.right() | ||
| override fun send(topic: String, key: String, value: String): Either<Throwable, Unit> { | ||
| val record = ProducerRecord(topic, key, value) | ||
| return stringProducer.runCatching { send(record) }.fold({ Unit.right() }, { it.left() }) | ||
| } | ||
| } |
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
Oops, something went wrong.