chore(plugin-server): use existing INGESTION_BATCH_SIZE and KAFKA_CON… (

#17927) chore(plugin-server): use existing INGESTION_BATCH_SIZE and KAFKA_CONSUMPTION_SESSION_TIMEOUT_MS
PostHog · Oct 11, 2023 · ed40b77 · ed40b77
1 parent 0df3b2f
commit ed40b77
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/plugin-server/src/main/ingestion-queues/kafka-queue.ts b/plugin-server/src/main/ingestion-queues/kafka-queue.ts
@@ -240,11 +240,11 @@ export class IngestionConsumer {
             topic: this.topic,
             groupId: this.consumerGroupId,
             autoCommit: true,
-            sessionTimeout: 30000,
+            sessionTimeout: this.pluginsServer.KAFKA_CONSUMPTION_SESSION_TIMEOUT_MS,
             consumerMaxBytes: this.pluginsServer.KAFKA_CONSUMPTION_MAX_BYTES,
             consumerMaxBytesPerPartition: this.pluginsServer.KAFKA_CONSUMPTION_MAX_BYTES_PER_PARTITION,
             consumerMaxWaitMs: this.pluginsServer.KAFKA_CONSUMPTION_MAX_WAIT_MS,
-            fetchBatchSize: 500,
+            fetchBatchSize: this.pluginsServer.INGESTION_BATCH_SIZE,
             topicCreationTimeoutMs: this.pluginsServer.KAFKA_TOPIC_CREATION_TIMEOUT_MS,
             eachBatch: (payload) => this.eachBatchConsumer(payload),
         })