cleanup: rename promises to ackPromises to make it more clear thats w…

…hat they are
PostHog · Mar 22, 2024 · 3ccfb85 · 3ccfb85
1 parent 08e6605
commit 3ccfb85
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 5 deletions.
diff --git a/plugin-server/src/main/ingestion-queues/batch-processing/each-batch-ingestion.ts b/plugin-server/src/main/ingestion-queues/batch-processing/each-batch-ingestion.ts
@@ -41,7 +41,7 @@ type IngestionSplitBatch = {
 type IngestResult = {
     // Promises that the batch handler should await on before committing offsets,
     // contains the Kafka producer ACKs, to avoid blocking after every message.
-    promises?: Array<Promise<void>>
+    ackPromises?: Array<Promise<void>>
 }
 
 async function handleProcessingError(
@@ -166,7 +166,7 @@ export async function eachBatchParallelIngestion(
                             return await runner.runEventPipeline(pluginEvent)
                         })) as IngestResult
 
-                        result.promises?.forEach((promise) =>
+                        result.ackPromises?.forEach((promise) =>
                             processingPromises.push(
                                 promise.catch(async (error) => {
                                     await handleProcessingError(error, message, pluginEvent, queue)

diff --git a/plugin-server/src/worker/ingestion/event-pipeline/runner.ts b/plugin-server/src/worker/ingestion/event-pipeline/runner.ts
@@ -25,7 +25,7 @@ import { processPersonsStep } from './processPersonsStep'
 export type EventPipelineResult = {
     // Promises that the batch handler should await on before committing offsets,
     // contains the Kafka producer ACKs, to avoid blocking after every message.
-    promises?: Array<Promise<void>>
+    ackPromises?: Array<Promise<void>>
     // Only used in tests
     // TODO: update to test for side-effects of running the pipeline rather than
     // this return type.
@@ -135,9 +135,9 @@ export class EventPipelineRunner {
         return this.registerLastStep('createEventStep', [rawClickhouseEvent, person], [eventAck])
     }
 
-    registerLastStep(stepName: string, args: any[], promises?: Array<Promise<void>>): EventPipelineResult {
+    registerLastStep(stepName: string, args: any[], ackPromises?: Array<Promise<void>>): EventPipelineResult {
         pipelineLastStepCounter.labels(stepName).inc()
-        return { promises: promises, lastStep: stepName, args }
+        return { ackPromises, lastStep: stepName, args }
     }
 
     protected runStep<Step extends (...args: any[]) => any>(