Status quo

Author: frifriSF59

airbyte-integrations/connectors/destination-s3-data-lake/src/main/kotlin/io/airbyte/integrations/destination/s3_data_lake/S3DataLakePartitioner.kt

@@ -8,6 +8,7 @@ import io.airbyte.cdk.load.command.Dedupe
 import io.airbyte.cdk.load.command.DestinationCatalog
 import io.airbyte.cdk.load.data.ObjectValue
 import io.airbyte.cdk.load.message.DestinationRecordAirbyteValue
+import io.airbyte.cdk.load.message.DestinationRecordRaw
 import io.airbyte.cdk.load.pipeline.InputPartitioner
 import jakarta.inject.Singleton
 import kotlin.math.abs
@@ -25,14 +26,16 @@ class S3DataLakePartitioner(catalog: DestinationCatalog) : InputPartitioner {
         }
     private val random = Random(System.currentTimeMillis())
 
-    override fun getPartition(record: DestinationRecordAirbyteValue, numParts: Int): Int {
+    override fun getPartition(record: DestinationRecordRaw, numParts: Int): Int {
+        val recordAirbyteValue = record.asDestinationRecordAirbyteValue()
+
         if (numParts == 1) {
             return 0
         }
 
-        streamToPrimaryKeyFieldNames[record.stream]?.let { primaryKey ->
+        streamToPrimaryKeyFieldNames[recordAirbyteValue.stream]?.let { primaryKey ->
             val primaryKeyValues =
-                primaryKey.map { it.map { key -> (record.data as ObjectValue).values[key] } }
+                primaryKey.map { it.map { key -> (recordAirbyteValue.data as ObjectValue).values[key] } }
             val hash = primaryKeyValues.hashCode()
             /** abs(MIN_VALUE) == MIN_VALUE, so we need to handle this case separately */
             if (hash == Int.MIN_VALUE) {