Queue between upload file parts and the upload completer. It will hold the fact of upload completion only, so in theory it can be Channel.UNLIMITED, but to be safe we'll limit it to 10,000 queued completions.

fileObjectLoaderPartQueue

@Singleton

@Named(value = "filePartQueue")

@Requires(condition = IsFileTransferCondition::class)

fun fileObjectLoaderPartQueue(@Named(value = "globalMemoryManager") globalMemoryManager: ReservationManager): ResourceReservingPartitionedQueue<PipelineEvent<ObjectKey, ObjectLoaderPartFormatter.FormattedPart>>

Queue between file part chunking and loading of file parts. It will hold the actual part bytes and needs to be sized based on the available reserved memory.

fileQueue

@Singleton

@Named(value = "fileQueue")

@Requires(condition = IsFileTransferCondition::class)

fun fileQueue(@Named(value = "numInputPartitions") numInputPartitions: Int): PartitionedQueue<PipelineEvent<StreamKey, DestinationRecordRaw>>

A queue for records with file references for file uploading.

objectLoaderClampedPartSizeBytes

@Singleton

@Named(value = "objectLoaderClampedPartSizeBytes")

@Requires(bean = ObjectLoader::class)

fun objectLoaderClampedPartSizeBytes(@Named(value = "objectLoaderPartQueue") queue: ResourceReservingPartitionedQueue<*>, @Named(value = "dataChannelMedium") dataChannelMedium: DataChannelMedium, @Named(value = "dataChannelSocketPaths") dataChannelSocketPaths: List<String>): Long

If we naively accept the part size and concurrency settings, we might end up with a connector that passes CI but can't run in a resource-limited production environment, because there isn't enough memory even for the workers to hold parts in flight.

objectLoaderLoadedPartQueue

@Singleton

@Named(value = "objectLoaderLoadedPartQueue")

@Requires(bean = ObjectLoader::class)

fun <T : RemoteObject<*>> objectLoaderLoadedPartQueue(): PartitionedQueue<PipelineEvent<ObjectKey, ObjectLoaderPartLoader.PartResult<T>>>

Queue between part 2 (upload parts) and part 3 (finish the upload). It will hold the fact of upload completion only, so in theory it can be Channel.UNLIMITED, but to be safe we'll limit it to 10,000 queued completions.

recordObjectLoaderPartQueue

@Singleton

@Named(value = "objectLoaderPartQueue")

@Requires(bean = ObjectLoader::class)

fun recordObjectLoaderPartQueue(@Named(value = "globalMemoryManager") globalMemoryManager: ReservationManager): ResourceReservingPartitionedQueue<PipelineEvent<ObjectKey, ObjectLoaderPartFormatter.FormattedPart>>

Queue between step 1 (format parts) and step 2 (load them): it will hold the actual part bytes and needs to be sized based on the available reserved memory.

recordQueue

@Singleton

@Named(value = "recordQueue")

fun recordQueue(@Named(value = "numInputPartitions") numInputPartitions: Int): PartitionedQueue<PipelineEvent<StreamKey, DestinationRecordRaw>>

A queue for records uploading.