remove init()

Author: arcoraven

packages/service-utils/src/node/kafka.ts

@@ -1,10 +1,5 @@
 import { checkServerIdentity } from "node:tls";
-import {
-  CompressionTypes,
-  Kafka,
-  type Producer,
-  type ProducerConfig,
-} from "kafkajs";
+import { CompressionTypes, Kafka, type Producer } from "kafkajs";
 import { compress, decompress } from "lz4js";
 
 // CompressionCodecs is not exported properly in kafkajs. Source: https://github.com/tulios/kafkajs/issues/1391
@@ -17,11 +12,10 @@ const { CompressionCodecs } = KafkaJS;
  *
  * Example:
  * ```ts
- * usageV2 = new KafkaProducer(..)
- * await usageV2.init()
- * await usageV2.sendEvents(events)
+ * kafka = new KafkaProducer(...)
+ * await kafka.send(topic, events)
  * // Non-blocking:
- * // void usageV2.sendEvents(events).catch(console.error)
+ * // void kafka.send(topic, events).catch((e) => console.error(e))
  * ```
  */
 export class KafkaProducer {
@@ -94,26 +88,9 @@ export class KafkaProducer {
     }
   }
 
-  /**
-   * Connect the producer.
-   * This must be called before calling `sendEvents()`.
-   */
-  async init(configOverrides?: ProducerConfig) {
-    this.producer = this.kafka.producer({
-      allowAutoTopicCreation: false,
-      ...configOverrides,
-    });
-    await this.producer.connect();
-  }
-
   /**
    * Send messages to a Kafka topic.
    * This method may throw. To call this non-blocking:
-   *
-   * ```ts
-   * kafka = new KafkaProducer(...)
-   * void kafka.send(topic, messages).catch(console.error)
-   *
    * @param topic
    * @param messages
    * @param configOverrides
@@ -124,13 +101,17 @@ export class KafkaProducer {
     /**
      * Reference: https://kafka.js.org/docs/producing#producing-messages
      */
-    configOverrides?: {
+    options?: {
       acks?: number;
       timeout?: number;
+      allowAutoTopicCreation?: boolean;
     },
   ): Promise<void> {
     if (!this.producer) {
-      throw new Error("Producer not initialized. Call `init()` first.");
+      this.producer = this.kafka.producer({
+        allowAutoTopicCreation: options?.allowAutoTopicCreation ?? false,
+      });
+      await this.producer.connect();
     }
 
     await this.producer.send({
@@ -139,9 +120,8 @@ export class KafkaProducer {
         value: JSON.stringify(m),
       })),
       compression: this.compression,
-      acks: -1, // All brokers must acknowledge
-      timeout: 10_000, // 10 seconds
-      ...configOverrides,
+      acks: options?.acks ?? -1, // Default: All brokers must acknowledge
+      timeout: options?.timeout ?? 10_000, // Default: 10 seconds
     });
   }
 

packages/service-utils/src/node/usageV2.ts

@@ -1,5 +1,4 @@
 import { randomUUID } from "node:crypto";
-import type { ProducerConfig } from "kafkajs";
 import {
   type UsageV2Event,
   type UsageV2Source,
@@ -14,10 +13,9 @@ import { KafkaProducer } from "./kafka.js";
  * Example:
  * ```ts
  * usageV2 = new UsageV2Producer(..)
- * await usageV2.init()
  * await usageV2.sendEvents(events)
  * // Non-blocking:
- * // void usageV2.sendEvents(events).catch(console.error)
+ * // void usageV2.sendEvents(events).catch((e) => console.error(e))
  * ```
  */
 export class UsageV2Producer {
@@ -55,32 +53,20 @@ export class UsageV2Producer {
     this.topic = getTopicName(config.source);
   }
 
-  /**
-   * Connect the producer.
-   * This must be called before calling `sendEvents()`.
-   */
-  async init(configOverrides?: ProducerConfig) {
-    return this.kafkaProducer.init(configOverrides);
-  }
-
   /**
    * Send usageV2 events.
    * This method may throw. To call this non-blocking:
-   *
-   * ```ts
-   * usageV2 = new UsageV2Producer(...)
-   * void usageV2.sendEvents(events).catch(console.error)
-   *
-   * @param events - The events to send.
+   * @param events
    */
   async sendEvents(
     events: UsageV2Event[],
     /**
      * Reference: https://kafka.js.org/docs/producing#producing-messages
      */
-    configOverrides?: {
+    options?: {
       acks?: number;
       timeout?: number;
+      allowAutoTopicCreation?: boolean;
     },
   ): Promise<void> {
     const parsedEvents = events.map((event) => ({
@@ -92,7 +78,7 @@ export class UsageV2Producer {
         ? event.team_id.slice(5)
         : event.team_id,
     }));
-    await this.kafkaProducer.send(this.topic, parsedEvents, configOverrides);
+    await this.kafkaProducer.send(this.topic, parsedEvents, options);
   }
 
   /**