Add error types to producer, and fix some errors

Author: milindl

MIGRATION.md

@@ -4,8 +4,10 @@
 
 ### Common
 
-1. Error Handling: Some possible subtypes of `KafkaJSError` have been removed,
-   and additional information has been added into `KafkaJSError`. 
+* Configuration changes
+
+* Error Handling: Some possible subtypes of `KafkaJSError` have been removed,
+   and additional information has been added into `KafkaJSError`.
    Internally, fields have been added denoting if the error is fatal, retriable, or abortable (the latter two only relevant for a
    transactional producer).
    Some error-specific fields have also been removed. An exhaustive list is at the bottom of this section.
@@ -17,7 +19,7 @@
                checks based on `error.code` or `error.type`.
 
    **Example:**:
-   ```js
+   ```javascript
    try {
       await producer.send(/* args */);
    } catch (error) {
@@ -59,6 +61,46 @@
 
 ### Producer
 
+* `sendBatch` is currently unsupported - but will be supported. TODO. However, the actual batching semantics are handled by librdkafka.
+* Changes to `send`:
+  1. `acks`, `compression` and `timeout` are not set on a per-send basis. Rather, they must be configured in the configuration.
+    Before:
+    ```javascript
+    const kafka = new Kafka({/* ... */});
+    const producer = kafka.producer();
+    await producer.connect();
+
+    await producer.send({
+      topic: 'test',
+      messages: [ /* ... */ ],
+      acks: 1,
+      compression: CompressionTypes.GZIP,
+      timeout: 30000,
+    });
+    ```
+
+    After:
+    ```javascript
+    const kafka = new Kafka({/* ... */});
+    const producer = kafka.producer({
+      rdKafka: {
+        topicConfig: {
+          "acks": "1",
+          "compression.codec": "gzip",
+          "message.timeout.ms": "30000",
+        },
+      }
+    });
+    await producer.connect();
+
+    await producer.send({
+      topic: 'test',
+      messages: [ /* ... */ ],
+    });
+    ```
+
+ * Error-handling for a failed `send` is stricter. While sending multiple messages, even if one of the messages fails, the method throws an error.
+
 ### Consumer
 
-## node-rdkafka
+## node-rdkafka

lib/kafkajs/_common.js

@@ -1,5 +1,4 @@
 const error = require("./_error");
-const LibrdKafkaError = require('../error');
 
 /**
  * @function kafkaJSToRdKafkaConfig()
@@ -47,7 +46,13 @@ async function kafkaJSToRdKafkaConfig(config) {
   return { globalConfig, topicConfig };
 }
 
+/**
+ * Converts a topicPartitionOffset from KafkaJS to a format that can be used by node-rdkafka.
+ * @param {import("../../types/kafkajs").TopicPartitionOffset} tpo
+ * @returns {{topic: string, partition: number, offset: number}}
+ */
 function topicPartitionOffsetToRdKafka(tpo) {
+  // TODO: do we need some checks for negative offsets and stuff? Or 'named' offsets?
   return {
     topic: tpo.topic,
     partition: tpo.partition,
@@ -57,8 +62,8 @@ function topicPartitionOffsetToRdKafka(tpo) {
 
 /**
  * Convert a librdkafka error from node-rdkafka into a KafkaJSError.
- * @param {LibrdKafkaError} librdKafkaError to convert from.
- * @returns KafkaJSError
+ * @param {import("../error")} librdKafkaError to convert from.
+ * @returns {error.KafkaJSError} the converted error.
  */
 function createKafkaJsErrorFromLibRdKafkaError(librdKafkaError) {
   const properties = {
@@ -72,34 +77,57 @@ function createKafkaJsErrorFromLibRdKafkaError(librdKafkaError) {
   let err = null;
 
   if (properties.code === error.ErrorCodes.ERR_OFFSET_OUT_OF_RANGE) {
-    err = new error.KafkaJSOffsetOutOfRange(e, properties);
+    err = new error.KafkaJSOffsetOutOfRange(librdKafkaError, properties);
   } else if (properties.code === error.ErrorCodes.ERR_REQUEST_TIMED_OUT) {
-    err = new error.KafkaJSRequestTimeoutError(e, properties);
+    err = new error.KafkaJSRequestTimeoutError(librdKafkaError, properties);
   } else if (properties.code === error.ErrorCodes.ERR__PARTIAL) {
-    err = new error.KafkaJSPartialMessageError(e, properties);
+    err = new error.KafkaJSPartialMessageError(librdKafkaError, properties);
   } else if (properties.code === error.ErrorCodes.ERR__AUTHENTICATION) {
-    err = new error.KafkaJSSASLAuthenticationError(e, properties);
+    err = new error.KafkaJSSASLAuthenticationError(librdKafkaError, properties);
   } else if (properties.code === error.ErrorCodes.ERR_GROUP_COORDINATOR_NOT_AVAILABLE) {
-    err = new error.KafkaJSGroupCoordinatorNotAvailableError(e, properties);
+    err = new error.KafkaJSGroupCoordinatorNotAvailableError(librdKafkaError, properties);
   } else if (properties.code === error.ErrorCodes.ERR__NOT_IMPLEMENTED) {
-    err = new error.KafkaJSNotImplemented(e, properties);
+    err = new error.KafkaJSNotImplemented(librdKafkaError, properties);
   } else if (properties.code === error.ErrorCodes.ERR__TIMED_OUT) {
-    err = new error.KafkaJSTimedOut(e, properties);
+    err = new error.KafkaJSTimedOut(librdKafkaError, properties);
   } else if (properties.code === error.ErrorCodes.ERR__ALL_BROKERS_DOWN) {
-    err = new error.KafkaJSNoBrokerAvailableError(e, properties);
+    err = new error.KafkaJSNoBrokerAvailableError(librdKafkaError, properties);
   } else if (properties.code === error.ErrorCodes.ERR__TRANSPORT) {
-    err = new error.KafkaJSConnectionError(e, properties);
+    err = new error.KafkaJSConnectionError(librdKafkaError, properties);
   } else if (properties.code > 0) { /* Indicates a non-local error */
-    err = new error.KafkaJSProtocolError(e, properties);
+    err = new error.KafkaJSProtocolError(librdKafkaError, properties);
   } else {
-    err = new error.KafkaJSError(e, properties);
+    err = new error.KafkaJSError(librdKafkaError, properties);
   }
 
+  console.log("Converted err = " + JSON.stringify(err, null, 2) + " librdkafka erro = " + JSON.stringify(librdKafkaError, null, 2));
   return err;
 }
 
+/**
+ * Converts KafkaJS headers to a format that can be used by node-rdkafka.
+ * @param {import("../../types/kafkajs").IHeaders|null} kafkaJSHeaders
+ * @returns {import("../../").MessageHeader[]|null} the converted headers.
+ */
+function convertToRdKafkaHeaders(kafkaJSHeaders) {
+  if (!kafkaJSHeaders) return null;
+
+  const headers = [];
+  for (const [key, value] of Object.entries(kafkaJSHeaders)) {
+    if (value.constructor === Array) {
+      for (const v of value) {
+        headers.push({ key, value: v });
+      }
+    } else {
+      headers.push({ key, value });
+    }
+  }
+  return headers;
+}
+
 module.exports = {
   kafkaJSToRdKafkaConfig,
   topicPartitionOffsetToRdKafka,
   createKafkaJsErrorFromLibRdKafkaError,
+  convertToRdKafkaHeaders,
 };

lib/kafkajs/_error.js

@@ -30,7 +30,7 @@ class KafkaJSError extends Error {
 
         const errTypes = Object
             .keys(LibrdKafkaError.codes)
-            .filter(k => LibrdKafkaError.codes[k] === kjsErr.code);
+            .filter(k => LibrdKafkaError.codes[k] === this.code);
 
         if (errTypes.length !== 1) {
             this.type = LibrdKafkaError.codes.ERR_UNKNOWN;
@@ -184,4 +184,4 @@ module.exports = {
     isRebalancing,
     isKafkaJSError,
     ErrorCodes: LibrdKafkaError.codes,
-};
+};

lib/kafkajs/_kafka.js

@@ -5,17 +5,29 @@ const error = require('./_error');
 class Kafka {
   #commonClientConfig = {};
 
+  /**
+   *
+   * @param {import("../../types/kafkajs").KafkaConfig} config
+   */
   constructor(config) {
     this.#commonClientConfig = config;
   }
 
+  /**
+   * Merge the producer/consumer specific configuration with the common configuration.
+   * @param {import("../../types/kafkajs").ProducerConfig|import("../../types/kafkajs").ConsumerConfig} config
+   * @returns
+   */
   #mergeConfiguration(config) {
     let baseConfig = Object.assign({}, this.#commonClientConfig);
     config = Object.assign({}, config);
 
+    // TODO: there's some confusion around this, as we currently allow
+    // rdKafka to be a function, but here, we don't seem to treat it as such.
+    // Correct this, so that only objects are allowed for `rdKafka`.
     let rdKafka = baseConfig.rdKafka;
     Object.assign(baseConfig, config);
-    if (rdKafka && config.rdKafka) {
+    if (typeof rdKafka === 'object' && typeof config.rdKafka === 'object') {
       baseConfig.rdKafka = {
         ...rdKafka,
         ...config.rdKafka,
@@ -24,10 +36,20 @@ class Kafka {
     return baseConfig;
   }
 
+  /**
+   * Creates a new producer.
+   * @param {import("../../types/kafkajs").ProducerConfig} config
+   * @returns {Producer}
+   */
   producer(config) {
     return new Producer(this.#mergeConfiguration(config));
   }
 
+  /**
+   * Creates a new consumer.
+   * @param {import("../../types/kafkajs").Consumer} config
+   * @returns {Consumer}
+   */
   consumer(config) {
     return new Consumer(this.#mergeConfiguration(config));
   }