only use queue

airbytehq · Dec 17, 2024 · a43a9c9 · a43a9c9
1 parent 5adcfb3
commit a43a9c9
Show file tree

Hide file tree

Showing 2 changed files with 1 addition and 5 deletions.
diff --git a/...k/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/task/implementor/ProcessBatchTask.kt b/...k/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/task/implementor/ProcessBatchTask.kt
@@ -29,7 +29,6 @@ class DefaultProcessBatchTask(
             val streamLoader = syncManager.getOrAwaitStreamLoader(batchEnvelope.streamDescriptor)
             val nextBatch = streamLoader.processBatch(batchEnvelope.batch)
             val nextWrapped = batchEnvelope.withBatch(nextBatch)
-            log.error { "processing batch ${batchEnvelope}" }
             taskLauncher.handleNewBatch(nextWrapped.streamDescriptor, nextWrapped)
         }
     }

diff --git a/...t-storage/src/main/kotlin/io/airbyte/cdk/load/write/object_storage/FilePartAccumulator.kt b/...t-storage/src/main/kotlin/io/airbyte/cdk/load/write/object_storage/FilePartAccumulator.kt
@@ -60,10 +60,7 @@ class FilePartAccumulator(
                                        index: Long,) {
 
         val wrapped = BatchEnvelope(batch, Range.singleton(index), streamDescriptor)
-        taskLauncher.handleNewBatch(streamDescriptor, wrapped)
-        if (batch.requiresProcessing) {
-            outputQueue.publish(wrapped)
-        }
+        outputQueue.publish(wrapped)
 
     }
 }