Destination S3-V2: Bug Fix: File xfer uses part size for part, not file size

Author: johnny-schmidt

airbyte-cdk/bulk/toolkits/load-object-storage/src/main/kotlin/io/airbyte/cdk/load/write/object_storage/FilePartAccumulator.kt

@@ -16,6 +16,7 @@ import io.airbyte.cdk.load.message.DestinationFile
 import io.airbyte.cdk.load.message.MultiProducerChannel
 import io.airbyte.cdk.load.message.object_storage.LoadablePart
 import io.airbyte.cdk.load.write.FileBatchAccumulator
+import io.github.oshai.kotlinlogging.KotlinLogging
 import java.io.File
 import java.nio.file.Path
 
@@ -28,6 +29,8 @@ class FilePartAccumulator(
     private val stream: DestinationStream,
     private val outputQueue: MultiProducerChannel<BatchEnvelope<*>>,
 ) : FileBatchAccumulator {
+    val log = KotlinLogging.logger {}
+
     override suspend fun processFilePart(file: DestinationFile, index: Long) {
         val key =
             Path.of(pathFactory.getFinalDirectory(stream), "${file.fileMessage.fileRelativePath}")
@@ -44,8 +47,9 @@ class FilePartAccumulator(
 
         while (true) {
             val bytePart =
-                ByteArray(ObjectStorageUploadConfiguration.DEFAULT_FILE_SIZE_BYTES.toInt())
+                ByteArray(ObjectStorageUploadConfiguration.DEFAULT_PART_SIZE_BYTES.toInt())
             val read = fileInputStream.read(bytePart)
+            log.info { "Read $read bytes from file" }
 
             if (read == -1) {
                 val filePart: ByteArray? = null
@@ -62,6 +66,7 @@ class FilePartAccumulator(
                 handleFilePart(batch, stream.descriptor, index)
             }
         }
+        fileInputStream.close()
         localFile.delete()
     }
 

airbyte-cdk/bulk/toolkits/load-object-storage/src/main/kotlin/io/airbyte/cdk/load/write/object_storage/PartToObjectAccumulator.kt

@@ -48,7 +48,7 @@ class PartToObjectAccumulator<T : RemoteObject<*>>(
         val streamingUpload = upload.streamingUpload.await()
 
         log.info {
-            "Processing loadable part ${batch.part.partIndex} of ${batch.part.key} (empty=${batch.part.isEmpty}; final=${batch.part.isFinal})"
+            "Processing loadable part ${batch.part.partIndex} of ${batch.part.key} (size=${batch.part.bytes?.size}; final=${batch.part.isFinal})"
         }
 
         // Upload provided bytes and update indexes.

airbyte-cdk/bulk/toolkits/load-object-storage/src/test/kotlin/io/airbyte/cdk/load/write/object_storage/FilePartAccumulatorTest.kt

@@ -50,7 +50,7 @@ class FilePartAccumulatorTest {
     fun testFilePartAccumulatorExactlyPartSize() = runTest {
         val finalDirectory = "finalDirectory"
         every { pathFactory.getFinalDirectory(stream) } returns finalDirectory
-        val file = createFile(ObjectStorageUploadConfiguration.DEFAULT_FILE_SIZE_BYTES.toInt())
+        val file = createFile(ObjectStorageUploadConfiguration.DEFAULT_PART_SIZE_BYTES.toInt())
         val index = 21L
         val fileMessage = createFileMessage(file)
 
@@ -64,7 +64,7 @@ class FilePartAccumulatorTest {
         val finalDirectory = "finalDirectory"
         every { pathFactory.getFinalDirectory(stream) } returns finalDirectory
         val file =
-            createFile(ObjectStorageUploadConfiguration.DEFAULT_FILE_SIZE_BYTES.toInt() + 1000)
+            createFile(ObjectStorageUploadConfiguration.DEFAULT_PART_SIZE_BYTES.toInt() + 1000)
         val index = 21L
         val fileMessage = createFileMessage(file)
 

airbyte-integrations/connectors/destination-s3/build.gradle

@@ -6,7 +6,7 @@ plugins {
 airbyteBulkConnector {
     core = 'load'
     toolkits = ['load-s3', 'load-avro', 'load-aws']
-    cdk = '0.296'
+    cdk = 'local'
 }
 
 application {