Repository: avro-kotlin/avro4k (kotlin)

Issue:
Instances of ByteBuffer serialize to empty array
Hi!
I have just encountered an issue when serialising a data class with a `ByteArray` field using avro4k together with kafka-avro-serializer.

The data class I tried to serialize looks like this:
```kotlin
@Serializable
data class File(val content: ByteArray)
```
Here's how I'm serializing it:
```kotlin
Avro.default.toRecord(serializer, data).let { record ->
  return kafkaSerializer.serialize(topic, record)
}
```

The problem is that in the serialized record the array in the `File#content` field is always empty.

I traced back the problem to the way the `ByteBuffer` instances are created in [`ByteArrayEncoder`](https://github.com/sksamuel/avro4k/blob/064a9580e35c1e8ef6ba10ed934e899679e7221a/avro4k-core/src/main/kotlin/com/sksamuel/avro4k/encoder/ByteArrayEncoder.kt#L32).
The created buffers always have the current position set to the index of the last character i.e. `position == limit`.
When serializer attempts to read from such buffers, they appear empty as `ByteBuffer#hasRemaining` will always return `false`.
This in turn results in empty arrays in serialised records.

I think this is a bug as byte buffers created by this library should be ready to be read by consumers of the records.

Pull request description:
This change will rewind all instances of `ByteBuffers` created by `ByteArrayEncoder` before returning the to the clients.

Fixes #55

Patch:
diff --git a/avro4k-core/src/main/kotlin/com/sksamuel/avro4k/encoder/ByteArrayEncoder.kt b/avro4k-core/src/main/kotlin/com/sksamuel/avro4k/encoder/ByteArrayEncoder.kt
index 7337e7ce..80163cda 100644
--- a/avro4k-core/src/main/kotlin/com/sksamuel/avro4k/encoder/ByteArrayEncoder.kt
+++ b/avro4k-core/src/main/kotlin/com/sksamuel/avro4k/encoder/ByteArrayEncoder.kt
@@ -29,7 +29,8 @@ class ByteArrayEncoder(private val schema: Schema,
                .array()
             callback(GenericData.get().createFixed(null, padded, schema))
          }
-         Schema.Type.BYTES -> callback(ByteBuffer.allocate(bytes.size).put(bytes.toByteArray()))
+         //Wrapping the resulting byte array directly as this does not duplicate the byte array
+         Schema.Type.BYTES -> callback(ByteBuffer.wrap(bytes.toByteArray()))
          else -> throw SerializationException("Cannot encode byte array when schema is ${schema.type}")
       }
 


Test patch:
diff --git a/avro4k-core/src/test/kotlin/com/sksamuel/avro4k/encoder/ByteArrayEncoderTest.kt b/avro4k-core/src/test/kotlin/com/sksamuel/avro4k/encoder/ByteArrayEncoderTest.kt
index 96812f1c..e0e2bc1e 100644
--- a/avro4k-core/src/test/kotlin/com/sksamuel/avro4k/encoder/ByteArrayEncoderTest.kt
+++ b/avro4k-core/src/test/kotlin/com/sksamuel/avro4k/encoder/ByteArrayEncoderTest.kt
@@ -24,19 +24,19 @@ class ByteArrayEncoderTest : FunSpec({
    test("encode ByteArray to ByteBuffer to ") {
       val schema = Avro.default.schema(ByteArrayTest.serializer())
       Avro.default.toRecord(ByteArrayTest.serializer(), ByteArrayTest(byteArrayOf(1, 4, 9))) shouldBe
-         ListRecord(schema, ByteBuffer.allocate(3).put(1).put(4).put(9))
+         ListRecord(schema, ByteBuffer.wrap(byteArrayOf(1,4,9)))
    }
 
    test("encode List<Byte> to ByteBuffer") {
       val schema = Avro.default.schema(ListByteTest.serializer())
       Avro.default.toRecord(ListByteTest.serializer(), ListByteTest(listOf(1, 4, 9))) shouldBe
-         ListRecord(schema, ByteBuffer.allocate(3).put(1).put(4).put(9))
+         ListRecord(schema, ByteBuffer.wrap(byteArrayOf(1,4,9)))
    }
 
    test("encode Array<Byte> to ByteBuffer") {
       val schema = Avro.default.schema(ArrayByteTest.serializer())
       Avro.default.toRecord(ArrayByteTest.serializer(), ArrayByteTest(arrayOf(1, 4, 9))) shouldBe
-         ListRecord(schema, ByteBuffer.allocate(3).put(1).put(4).put(9))
+         ListRecord(schema, ByteBuffer.wrap(byteArrayOf(1,4,9)))
    }
 
    test("encode ByteArray as FIXED when schema is Type.Fixed") {
diff --git a/avro4k-core/src/test/kotlin/com/sksamuel/avro4k/io/BasicIoTest.kt b/avro4k-core/src/test/kotlin/com/sksamuel/avro4k/io/BasicIoTest.kt
index 1fc5845c..52759db8 100644
--- a/avro4k-core/src/test/kotlin/com/sksamuel/avro4k/io/BasicIoTest.kt
+++ b/avro4k-core/src/test/kotlin/com/sksamuel/avro4k/io/BasicIoTest.kt
@@ -1,5 +1,6 @@
 package com.sksamuel.avro4k.io
 
+import com.sksamuel.avro4k.Avro
 import io.kotlintest.shouldBe
 import io.kotlintest.specs.FunSpec
 import kotlinx.serialization.Serializable
@@ -26,6 +27,9 @@ class BasicIoTest : FunSpec() {
       @Serializable
       data class LongTest(val z: Long)
 
+      @Suppress("ArrayInDataClass")
+      @Serializable
+      data class ByteArrayTest(val z : ByteArray)
       test("read write out booleans") {
          writeRead(BooleanTest(true), BooleanTest.serializer())
          writeRead(BooleanTest(false), BooleanTest.serializer()) {
@@ -61,5 +65,12 @@ class BasicIoTest : FunSpec() {
       test("read write out floats") {
          writeRead(FloatTest(3.4F), FloatTest.serializer())
       }
+      test("read write out byte arrays") {
+         val expected = ByteArrayTest("ABC".toByteArray())
+         writeRead(expected,ByteArrayTest.serializer()){
+            val deserialized = Avro.default.fromRecord(ByteArrayTest.serializer(),it)
+            expected.z shouldBe deserialized.z
+         }
+      }
    }
 }
\ No newline at end of file