[service-utils] Expose more kafka configs, set max in-flight requests (…

…#6181)
thirdweb-dev · Feb 6, 2025 · dfd64d2 · dfd64d2
1 parent 4e703fd
commit dfd64d2
Show file tree

Hide file tree

Showing 3 changed files with 24 additions and 14 deletions.
diff --git a/.changeset/quiet-goats-know.md b/.changeset/quiet-goats-know.md
@@ -0,0 +1,5 @@
+---
+"@thirdweb-dev/service-utils": patch
+---
+
+[service-utils] Expose maxInFlightRequests in Kafka producer
diff --git a/packages/service-utils/src/node/kafka.ts b/packages/service-utils/src/node/kafka.ts
@@ -6,6 +6,20 @@ import { compress, decompress } from "lz4js";
 import KafkaJS from "kafkajs";
 const { CompressionCodecs } = KafkaJS;
 
+/**
+ * Reference: https://kafka.js.org/docs/producing#producing-messages
+ */
+export interface KafkaProducerSendOptions {
+  // Per-message settings.
+  acks?: number;
+  timeout?: number;
+
+  // Per-producer settings.
+  retries?: number;
+  allowAutoTopicCreation?: boolean;
+  maxInFlightRequests?: number;
+}
+
 /**
  * Creates a KafkaProducer which opens a persistent TCP connection.
  * This class is thread-safe so your service should re-use one instance.
@@ -98,18 +112,13 @@ export class KafkaProducer {
   async send(
     topic: string,
     messages: Record<string, unknown>[],
-    /**
-     * Reference: https://kafka.js.org/docs/producing#producing-messages
-     */
-    options?: {
-      acks?: number;
-      timeout?: number;
-      allowAutoTopicCreation?: boolean;
-    },
+    options?: KafkaProducerSendOptions,
   ): Promise<void> {
     if (!this.producer) {
       this.producer = this.kafka.producer({
         allowAutoTopicCreation: options?.allowAutoTopicCreation ?? false,
+        maxInFlightRequests: options?.maxInFlightRequests ?? 2000,
+        retry: { retries: options?.retries ?? 5 },
       });
       await this.producer.connect();
     }

diff --git a/packages/service-utils/src/node/usageV2.ts b/packages/service-utils/src/node/usageV2.ts
@@ -4,7 +4,7 @@ import {
   type UsageV2Source,
   getTopicName,
 } from "../core/usageV2.js";
-import { KafkaProducer } from "./kafka.js";
+import { KafkaProducer, type KafkaProducerSendOptions } from "./kafka.js";
 
 /**
  * Creates a UsageV2Producer which opens a persistent TCP connection.
@@ -63,11 +63,7 @@ export class UsageV2Producer {
     /**
      * Reference: https://kafka.js.org/docs/producing#producing-messages
      */
-    options?: {
-      acks?: number;
-      timeout?: number;
-      allowAutoTopicCreation?: boolean;
-    },
+    options?: KafkaProducerSendOptions,
   ): Promise<void> {
     const parsedEvents = events.map((event) => ({
       ...event,