-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #17 from OSGP/feature/FDP-2357-kod-status-en-meetb…
…erichten FDP-2357: AvroSerializer can handle all Avro messages
- Loading branch information
Showing
7 changed files
with
163 additions
and
41 deletions.
There are no files selected for viewing
43 changes: 43 additions & 0 deletions
43
kafka-avro/src/main/kotlin/com/gxf/utilities/kafka/avro/AvroEncoder.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,43 @@ | ||
| package com.gxf.utilities.kafka.avro | ||
|
|
||
| import org.apache.avro.message.BinaryMessageEncoder | ||
| import org.apache.avro.specific.SpecificData | ||
| import org.apache.avro.specific.SpecificRecordBase | ||
| import org.slf4j.LoggerFactory | ||
| import java.io.IOException | ||
| import java.io.OutputStream | ||
| import kotlin.reflect.KClass | ||
|
|
||
| object AvroEncoder { | ||
| val encoders: HashMap<KClass<out SpecificRecordBase>, BinaryMessageEncoder<SpecificRecordBase>> = HashMap() | ||
|
|
||
| private val logger = LoggerFactory.getLogger(AvroEncoder::class.java) | ||
|
|
||
| @Throws(IOException::class) | ||
| fun encode(message: SpecificRecordBase): ByteArray { | ||
| val encoder = getEncoder(message) | ||
| val byteBuffer = encoder.encode(message) | ||
| val bytes = ByteArray(byteBuffer.remaining()) | ||
| byteBuffer[bytes] | ||
| return bytes | ||
| } | ||
|
|
||
| @Throws(IOException::class) | ||
| fun encode(message: SpecificRecordBase, stream: OutputStream) { | ||
| val encoder = getEncoder(message) | ||
| encoder.encode(message, stream) | ||
| } | ||
|
|
||
| private fun getEncoder(message: SpecificRecordBase): BinaryMessageEncoder<SpecificRecordBase> { | ||
| val existingEncoder = encoders[message::class] | ||
|
|
||
| if(existingEncoder != null) { | ||
| return existingEncoder | ||
| } | ||
|
|
||
| logger.info("New encoder created for Avro schema {}", message::class) | ||
| val newEncoder = BinaryMessageEncoder<SpecificRecordBase>(SpecificData(), message.schema) | ||
| encoders[message::class] = newEncoder | ||
| return newEncoder | ||
| } | ||
| } |
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
78 changes: 78 additions & 0 deletions
78
kafka-avro/src/test/kotlin/com/gxf/utilities/kafka/avro/AvroEncoderTest.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 com.gxf.utilities.kafka.avro | ||
|
|
||
| import org.apache.avro.Schema | ||
| import org.apache.avro.specific.SpecificRecordBase | ||
| import org.assertj.core.api.Assertions.assertThat | ||
| import org.junit.jupiter.api.Test | ||
| import java.io.ByteArrayOutputStream | ||
|
|
||
| class AvroEncoderTest { | ||
| @Test | ||
| fun testEncodersCache() { | ||
| val message1 = AvroSchema1("field no 1", "field no 2") | ||
| val message2 = AvroSchema2("message in a bottle") | ||
| val message3 = AvroSchema2("another message for you") | ||
| val message4 = AvroSchema2("encode to stream!") | ||
|
|
||
| AvroEncoder.encode(message1) | ||
| AvroEncoder.encode(message2) | ||
| AvroEncoder.encode(message3) | ||
| AvroEncoder.encode(message4, ByteArrayOutputStream()) | ||
|
|
||
| assertThat(AvroEncoder.encoders).containsKeys(AvroSchema1::class) | ||
| assertThat(AvroEncoder.encoders).containsKeys(AvroSchema2::class) | ||
| assertThat(AvroEncoder.encoders.size).isEqualTo(2) | ||
| } | ||
| } | ||
|
|
||
| class AvroSchema1(private var field1: String, private var field2: String): SpecificRecordBase() { | ||
| override fun getSchema(): Schema = Schema.Parser() | ||
| .parse("{\"type\":\"record\",\"name\":\"AvroSchema1\",\"namespace\":\"com.alliander.gxf.utilities.kafka.avro\",\"fields\":[{\"name\":\"field1\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}}, {\"name\":\"field2\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}}]}") | ||
|
|
||
| override fun put(field: Int, value: Any?) { | ||
| when(field) { | ||
| 0 -> { | ||
| if(value != null) { | ||
| field1 = value.toString() | ||
| } | ||
| } | ||
| 1 -> { | ||
| if(value != null) { | ||
| field2 = value.toString() | ||
| } | ||
| } | ||
| else -> throw IndexOutOfBoundsException() | ||
| } | ||
| } | ||
|
|
||
| override fun get(field: Int): Any { | ||
| return when(field) { | ||
| 0 -> field1 | ||
| 1 -> field2 | ||
| else -> throw IndexOutOfBoundsException() | ||
| } | ||
| } | ||
| } | ||
|
|
||
| class AvroSchema2(private var message: String): SpecificRecordBase() { | ||
| override fun getSchema(): Schema = Schema.Parser() | ||
| .parse("{\"type\":\"record\",\"name\":\"AvroSchema2\",\"namespace\":\"com.alliander.gxf.utilities.kafka.avro\",\"fields\":[{\"name\":\"message\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}}]}") | ||
|
|
||
| override fun put(field: Int, value: Any?) { | ||
| when(field) { | ||
| 0 -> { | ||
| if(value != null) { | ||
| message = value.toString() | ||
| } | ||
| } | ||
| else -> throw IndexOutOfBoundsException() | ||
| } | ||
| } | ||
|
|
||
| override fun get(field: Int): Any { | ||
| return when(field) { | ||
| 0 -> message | ||
| else -> throw IndexOutOfBoundsException() | ||
| } | ||
| } | ||
| } |
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