JAVA-6224 Ensure Kotlin can encode ByteArrays correctly

bson-kotlin/src/main/kotlin/org/bson/codecs/kotlin/ArrayCodec.kt

@@ -47,8 +47,17 @@ internal data class ArrayCodec<R : Any, V>(private val kClass: KClass<R>, privat
                         else -> Pair(Object::class.java as Class<Any>, emptyList())
                     }
                 }
-            val codec =
-                if (nestedTypes.isEmpty()) codecRegistry.get(valueClass) else codecRegistry.get(valueClass, nestedTypes)
+            // ByteArrays are encoded as compact BSON Binary, not as a BSON Array of Int32.
+            // Resolved directly, as a registry may order ValueCodecProvider ahead of
+            // ArrayCodecProvider.
+            val codec: Codec<Any?> =
+                if (valueClass == ByteArray::class.java) {
+                    LenientByteArrayCodec as Codec<Any?>
+                } else if (nestedTypes.isEmpty()) {
+                    codecRegistry.get(valueClass)
+                } else {
+                    codecRegistry.get(valueClass, nestedTypes)
+                }
             return ArrayCodec(kClass, codec)
         }
     }

bson-kotlin/src/main/kotlin/org/bson/codecs/kotlin/ArrayCodecProvider.kt

@@ -24,8 +24,12 @@ import org.bson.codecs.configuration.CodecRegistry
 public class ArrayCodecProvider : CodecProvider {
     override fun <T : Any> get(clazz: Class<T>, registry: CodecRegistry): Codec<T>? = get(clazz, emptyList(), registry)
 
+    @Suppress("UNCHECKED_CAST")
     override fun <T : Any> get(clazz: Class<T>, typeArguments: List<Type>, registry: CodecRegistry): Codec<T>? =
-        if (clazz.isArray) {
+        if (clazz == ByteArray::class.java) {
+            // ByteArrays are encoded as compact BSON Binary, not as a BSON Array of Int32.
+            LenientByteArrayCodec as Codec<T>
+        } else if (clazz.isArray) {
             ArrayCodec.create(clazz.kotlin, typeArguments, registry)
         } else null
 }

bson-kotlin/src/main/kotlin/org/bson/codecs/kotlin/DataClassCodec.kt

@@ -251,8 +251,11 @@ internal data class DataClassCodec<T : Any>(
 
         @Suppress("UNCHECKED_CAST")
         private fun CodecRegistry.getCodec(kParameter: KParameter, clazz: Class<Any>, types: List<Type>): Codec<Any> {
+            // ByteArrays are encoded as compact BSON Binary, not as a BSON Array of Int32.
             val codec =
-                if (clazz.isArray) {
+                if (clazz == ByteArray::class.java) {
+                    LenientByteArrayCodec as Codec<Any>
+                } else if (clazz.isArray) {
                     ArrayCodec.create(clazz.kotlin, types, this)
                 } else if (types.isEmpty()) {
                     this.get(clazz)

bson-kotlin/src/main/kotlin/org/bson/codecs/kotlin/LenientByteArrayCodec.kt

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.bson.codecs.kotlin
+
+import java.io.ByteArrayOutputStream
+import org.bson.BsonInvalidOperationException
+import org.bson.BsonReader
+import org.bson.BsonType
+import org.bson.codecs.ByteArrayCodec
+import org.bson.codecs.DecoderContext
+
+/**
+ * A [ByteArrayCodec] that also decodes the legacy BSON Array representation of a `ByteArray`.
+ *
+ * Versions 5.2.0 - 5.9.x encoded `ByteArray` data class fields as a BSON Array of Int32, one element per byte, rather
+ * than as a compact BSON Binary. This codec still decodes those documents, so data written by those versions remains
+ * readable.
+ *
+ * Encoding is inherited unchanged from [ByteArrayCodec] and always produces BSON Binary. Re-saving a document therefore
+ * migrates it away from the legacy representation.
+ *
+ * Only values the legacy encoder could have produced are accepted: every element must be an Int32 within the signed
+ * byte range. Anything else throws [BsonInvalidOperationException] rather than silently decoding an unrelated BSON
+ * Array into bytes.
+ */
+internal object LenientByteArrayCodec : ByteArrayCodec() {
+
+    override fun decode(reader: BsonReader, decoderContext: DecoderContext): ByteArray =
+        if (reader.currentBsonType == BsonType.ARRAY) {
+            decodeLegacyArray(reader)
+        } else {
+            super.decode(reader, decoderContext)
+        }
+
+    private fun decodeLegacyArray(reader: BsonReader): ByteArray {
+        val bytes = ByteArrayOutputStream()
+        reader.readStartArray()
+        while (reader.readBsonType() != BsonType.END_OF_DOCUMENT) {
+            if (reader.currentBsonType != BsonType.INT32) {
+                throw invalidElement(reader.currentBsonType.toString())
+            }
+            val value = reader.readInt32()
+            if (value < Byte.MIN_VALUE || value > Byte.MAX_VALUE) {
+                throw invalidElement(value.toString())
+            }
+            bytes.write(value)
+        }
+        reader.readEndArray()
+        return bytes.toByteArray()
+    }
+
+    private fun invalidElement(found: String) =
+        BsonInvalidOperationException(
+            "Invalid element while decoding a byte array from a BSON Array: " +
+                "expected an INT32 in the range -128..127, but found $found.")
+}

bson-kotlin/src/test/kotlin/org/bson/codecs/kotlin/DataClassCodecTest.kt

@@ -16,11 +16,14 @@
 package org.bson.codecs.kotlin
 
 import kotlin.test.assertEquals
+import kotlin.test.assertTrue
 import org.bson.BsonDocument
 import org.bson.BsonDocumentReader
 import org.bson.BsonDocumentWriter
+import org.bson.BsonInvalidOperationException
 import org.bson.codecs.DecoderContext
 import org.bson.codecs.EncoderContext
+import org.bson.codecs.ValueCodecProvider
 import org.bson.codecs.configuration.CodecConfigurationException
 import org.bson.codecs.configuration.CodecRegistries.fromProviders
 import org.bson.codecs.kotlin.samples.Box
@@ -45,6 +48,7 @@ import org.bson.codecs.kotlin.samples.DataClassWithBsonExtraElements
 import org.bson.codecs.kotlin.samples.DataClassWithBsonId
 import org.bson.codecs.kotlin.samples.DataClassWithBsonIgnore
 import org.bson.codecs.kotlin.samples.DataClassWithBsonProperty
+import org.bson.codecs.kotlin.samples.DataClassWithByteArray
 import org.bson.codecs.kotlin.samples.DataClassWithCollections
 import org.bson.codecs.kotlin.samples.DataClassWithDataClassMapKey
 import org.bson.codecs.kotlin.samples.DataClassWithDefaults
@@ -59,6 +63,7 @@ import org.bson.codecs.kotlin.samples.DataClassWithMutableList
 import org.bson.codecs.kotlin.samples.DataClassWithMutableMap
 import org.bson.codecs.kotlin.samples.DataClassWithMutableSet
 import org.bson.codecs.kotlin.samples.DataClassWithNativeArrays
+import org.bson.codecs.kotlin.samples.DataClassWithNestedByteArrays
 import org.bson.codecs.kotlin.samples.DataClassWithNestedParameterized
 import org.bson.codecs.kotlin.samples.DataClassWithNestedParameterizedDataClass
 import org.bson.codecs.kotlin.samples.DataClassWithNullableGeneric
@@ -122,6 +127,11 @@ class DataClassCodecTest {
             | "arraySimple": ["a", "b", "c", "d"],
             | "nestedArrays":  [["e", "f"], [], ["g", "h"]],
             | "arrayOfMaps":  [{"A": ["aa"], "B": ["bb"]}, {}, {"C": ["cc", "ccc"]}],
+            | "byteArray": {"${'$'}binary": {"base64": "AQIDBA==", "subType": "00"}},
+            | "nestedByteArrays": [
+            |     {"${'$'}binary": {"base64": "AQI=", "subType": "00"}},
+            |     {"${'$'}binary": {"base64": "AwQF", "subType": "00"}}
+            | ],
             |}"""
                 .trimMargin()
 
@@ -130,7 +140,9 @@ class DataClassCodecTest {
                 arrayOf("a", "b", "c", "d"),
                 arrayOf(arrayOf("e", "f"), emptyArray(), arrayOf("g", "h")),
                 arrayOf(
-                    mapOf("A" to arrayOf("aa"), "B" to arrayOf("bb")), emptyMap(), mapOf("C" to arrayOf("cc", "ccc"))))
+                    mapOf("A" to arrayOf("aa"), "B" to arrayOf("bb")), emptyMap(), mapOf("C" to arrayOf("cc", "ccc"))),
+                byteArrayOf(1, 2, 3, 4),
+                arrayOf(byteArrayOf(1, 2), byteArrayOf(3, 4, 5)))
 
         assertRoundTrips(expected, dataClass)
     }
@@ -140,7 +152,7 @@ class DataClassCodecTest {
         val expected =
             """{
             |    "booleanArray": [true, false],
-            |    "byteArray": [1, 2],
+            |    "byteArray": {"${'$'}binary": {"base64": "AQI=", "subType": "00"}},
             |    "charArray": ["a", "b"],
             |    "doubleArray": [ 1.1, 2.2, 3.3],
             |    "floatArray": [1.0, 2.0, 3.0],
@@ -168,6 +180,111 @@ class DataClassCodecTest {
         assertRoundTrips(expected, dataClass)
     }
 
+    @Test
+    fun testDataClassWithByteArrayEncodesAsBinary() {
+        // A ByteArray field must encode as compact BSON Binary (subType 00),
+        // via ByteArrayCodec, not as a BSON Array of Int32 (one element per byte).
+        val expected = """{"byteArray": {"${'$'}binary": {"base64": "AQIDBA==", "subType": "00"}}}"""
+        assertRoundTrips(expected, DataClassWithByteArray(byteArrayOf(1, 2, 3, 4)))
+    }
+
+    @Test
+    fun testDataClassWithByteArrayDoesNotExpandDocumentSize() {
+        // BSON Array of ints encoding expands size ~8-10x, breaking the 16MB limit.
+        // Binary encoding keeps a 1MB payload close to 1MB.
+        val oneMegabyte = ByteArray(1_000_000)
+        val codec = DataClassCodec.create(DataClassWithByteArray::class, registry())!!
+        val document = BsonDocument()
+        codec.encode(
+            BsonDocumentWriter(document), DataClassWithByteArray(oneMegabyte), EncoderContext.builder().build())
+        val encodedSize = document.getBinary("byteArray").data.size
+        assertEquals(1_000_000, encodedSize)
+    }
+
+    @Test
+    fun testDataClassWithByteArrayDecodesLegacyBsonArray() {
+        // Versions 5.2.0 - 5.9.x encoded ByteArray as a BSON Array of Int32.
+        // Those documents must still decode.
+        assertDecodesTo(
+            BsonDocument.parse("""{"byteArray": [1, 2, 3, 4]}"""), DataClassWithByteArray(byteArrayOf(1, 2, 3, 4)))
+    }
+
+    @Test
+    fun testDataClassWithByteArrayDecodesLegacyBsonArrayWithFullByteRange() {
+        assertDecodesTo(
+            BsonDocument.parse("""{"byteArray": [-128, -1, 0, 127]}"""),
+            DataClassWithByteArray(byteArrayOf(-128, -1, 0, 127)))
+    }
+
+    @Test
+    fun testDataClassWithByteArrayDecodesEmptyLegacyBsonArray() {
+        assertDecodesTo(BsonDocument.parse("""{"byteArray": []}"""), DataClassWithByteArray(byteArrayOf()))
+    }
+
+    @Test
+    fun testDataClassWithArraysDecodesLegacyByteArrays() {
+        val data =
+            BsonDocument.parse(
+                """{
+                | "arraySimple": ["a", "b", "c", "d"],
+                | "nestedArrays":  [["e", "f"], [], ["g", "h"]],
+                | "arrayOfMaps":  [{"A": ["aa"], "B": ["bb"]}, {}, {"C": ["cc", "ccc"]}],
+                | "byteArray": [1, 2, 3, 4],
+                | "nestedByteArrays": [[1, 2], [3, 4, 5]],
+                |}"""
+                    .trimMargin())
+
+        val expected =
+            DataClassWithArrays(
+                arrayOf("a", "b", "c", "d"),
+                arrayOf(arrayOf("e", "f"), emptyArray(), arrayOf("g", "h")),
+                arrayOf(
+                    mapOf("A" to arrayOf("aa"), "B" to arrayOf("bb")), emptyMap(), mapOf("C" to arrayOf("cc", "ccc"))),
+                byteArrayOf(1, 2, 3, 4),
+                arrayOf(byteArrayOf(1, 2), byteArrayOf(3, 4, 5)))
+
+        assertDecodesTo(data, expected)
+    }
+
+    @Test
+    fun testNestedByteArraysDecodeLegacyBsonArraysWithValueCodecProviderFirst() {
+        // MongoClientSettings.DEFAULT_CODEC_REGISTRY orders ValueCodecProvider ahead of
+        // ArrayCodecProvider, so resolving the element codec via the registry would find
+        // ByteArrayCodec
+        // rather than LenientByteArrayCodec.
+        val registry = fromProviders(ValueCodecProvider(), ArrayCodecProvider(), DataClassCodecProvider())
+        val codec = DataClassCodec.create(DataClassWithNestedByteArrays::class, registry)!!
+        val decoded =
+            codec.decode(
+                BsonDocumentReader(BsonDocument.parse("""{"nestedByteArrays": [[1, 2], [3, 4]]}""")),
+                DecoderContext.builder().build())
+
+        assertEquals(DataClassWithNestedByteArrays(arrayOf(byteArrayOf(1, 2), byteArrayOf(3, 4))), decoded)
+    }
+
+    @Test
+    fun testDataClassWithByteArrayLegacyBsonArrayFailures() {
+        assertLegacyByteArrayDecodeFails("""{"byteArray": [1, 128]}""")
+        assertLegacyByteArrayDecodeFails("""{"byteArray": [1, -129]}""")
+        assertLegacyByteArrayDecodeFails("""{"byteArray": [1, 300]}""")
+        assertLegacyByteArrayDecodeFails("""{"byteArray": [1, "2"]}""")
+        assertLegacyByteArrayDecodeFails("""{"byteArray": [1, 2.0]}""")
+        assertLegacyByteArrayDecodeFails("""{"byteArray": [1, {"${'$'}numberLong": "2"}]}""")
+        assertLegacyByteArrayDecodeFails("""{"byteArray": [1, null]}""")
+    }
+
+    private fun assertLegacyByteArrayDecodeFails(json: String) {
+        val codec = DataClassCodec.create(DataClassWithByteArray::class, registry())!!
+        val exception =
+            assertThrows<CodecConfigurationException>(json) {
+                codec.decode(BsonDocumentReader(BsonDocument.parse(json)), DecoderContext.builder().build())
+            }
+        val cause = exception.cause
+        assertTrue(cause is BsonInvalidOperationException, json)
+        // Must fail on element validation, not on the absence of BSON Binary.
+        assertTrue(cause.message!!.contains("expected an INT32 in the range -128..127"), cause.message)
+    }
+
     @Test
     fun testDataClassWithDefaults() {
         val expectedDefault =

bson-kotlin/src/test/kotlin/org/bson/codecs/kotlin/samples/DataClasses.kt

@@ -52,7 +52,9 @@ data class DataClassWithCollections(
 data class DataClassWithArrays(
     val arraySimple: Array<String>,
     val nestedArrays: Array<Array<String>>,
-    val arrayOfMaps: Array<Map<String, Array<String>>>
+    val arrayOfMaps: Array<Map<String, Array<String>>>,
+    val byteArray: ByteArray,
+    val nestedByteArrays: Array<ByteArray>
 ) {
     override fun equals(other: Any?): Boolean {
         if (this === other) return true
@@ -70,13 +72,18 @@ data class DataClassWithArrays(
             map.keys.forEach { key -> if (!map[key].contentEquals(otherMap[key])) return false }
         }
 
+        if (!byteArray.contentEquals(other.byteArray)) return false
+        if (!nestedByteArrays.contentDeepEquals(other.nestedByteArrays)) return false
+
         return true
     }
 
     override fun hashCode(): Int {
         var result = arraySimple.contentHashCode()
         result = 31 * result + nestedArrays.contentDeepHashCode()
         result = 31 * result + arrayOfMaps.contentHashCode()
+        result = 31 * result + byteArray.contentHashCode()
+        result = 31 * result + nestedByteArrays.contentDeepHashCode()
         return result
     }
 }
@@ -134,6 +141,28 @@ data class DataClassWithNativeArrays(
     }
 }
 
+data class DataClassWithByteArray(val byteArray: ByteArray) {
+    override fun equals(other: Any?): Boolean {
+        if (this === other) return true
+        if (javaClass != other?.javaClass) return false
+        other as DataClassWithByteArray
+        return byteArray.contentEquals(other.byteArray)
+    }
+
+    override fun hashCode(): Int = byteArray.contentHashCode()
+}
+
+data class DataClassWithNestedByteArrays(val nestedByteArrays: Array<ByteArray>) {
+    override fun equals(other: Any?): Boolean {
+        if (this === other) return true
+        if (javaClass != other?.javaClass) return false
+        other as DataClassWithNestedByteArrays
+        return nestedByteArrays.contentDeepEquals(other.nestedByteArrays)
+    }
+
+    override fun hashCode(): Int = nestedByteArrays.contentDeepHashCode()
+}
+
 data class DataClassWithDefaults(
     val boolean: Boolean = false,
     val string: String = "String",