Skip to content

Destination S3V2: Fix: File xfer uses part size for part, not file size #52662

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Feb 2, 2025
Merged

Destination S3V2: Fix: File xfer uses part size for part, not file size #52662

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion ...t-storage/src/main/kotlin/io/airbyte/cdk/load/write/object_storage/FilePartAccumulator.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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}")
Expand All @@ -44,8 +47,9 @@ class FilePartAccumulator(

while (true) {
val bytePart =
ByteArray(ObjectStorageUploadConfiguration.DEFAULT_FILE_SIZE_BYTES.toInt())
ByteArray(ObjectStorageUploadConfiguration.DEFAULT_PART_SIZE_BYTES.toInt())
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should this class be reading from the ObjectStorageUploadConfiguration instead of referencing the constant directly?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes

val read = fileInputStream.read(bytePart)
log.info { "Read $read bytes from file" }

if (read == -1) {
val filePart: ByteArray? = null
Expand All @@ -62,6 +66,7 @@ class FilePartAccumulator(
handleFilePart(batch, stream.descriptor, index)
}
}
fileInputStream.close()
localFile.delete()
}

Expand Down
2 changes: 1 addition & 1 deletion ...orage/src/main/kotlin/io/airbyte/cdk/load/write/object_storage/PartToObjectAccumulator.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
4 changes: 2 additions & 2 deletions ...orage/src/test/kotlin/io/airbyte/cdk/load/write/object_storage/FilePartAccumulatorTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand All @@ -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)

Expand Down
2 changes: 1 addition & 1 deletion airbyte-integrations/connectors/destination-s3/build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ plugins {
airbyteBulkConnector {
core = 'load'
toolkits = ['load-s3', 'load-avro', 'load-aws']
cdk = '0.296'
cdk = 'local'
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's repin the cdk after ya?

}

application {
Expand Down
Loading