Update comments and fix small issues

Author: milindl

lib/kafkajs/_common.js

@@ -188,7 +188,7 @@ const CompatibilityErrorMessages = Object.freeze({
   socketFactory: () =>
     "The socketFactory property is not supported.",
   logLevelName: (setLevel) =>
-    "The log level must be one of: " + logLevel.keys().join(", ") + ", was " + setLevel,
+    "The log level must be one of: " + Object.keys(logLevel).join(", ") + ", was " + setLevel,
   reauthenticationThreshold: () =>
     "Reauthentication threshold cannot be set, and reauthentication is automated when 80% of connections.max.reauth.ms is reached.",
   unsupportedKey: (key) =>

lib/kafkajs/_consumer.js

@@ -1069,7 +1069,7 @@ class Consumer {
    * Otherwise, they will be committed when commitOffsets is called without arguments.
    *
    * enable.auto.offset.store must be set to false to use this API.
-   * @param {import("../../types/kafkajs").TopicPartitionOffsetAndMetadata[]?} topicPartitions
+   * @param {import("../../types/kafkajs").TopicPartitionOffset[]?} topicPartitions
    */
   storeOffsets(topicPartitions) {
     if (this.#state !== ConsumerState.CONNECTED) {
@@ -1099,7 +1099,7 @@ class Consumer {
 
   /**
    * Commit offsets for the given topic partitions. If topic partitions are not specified, commits all offsets.
-   * @param {import("../../types/kafkajs").TopicPartitionOffsetAndMetadata[]?} topicPartitions
+   * @param {import("../../types/kafkajs").TopicPartitionOffset[]?} topicPartitions
    * @returns {Promise<void>} a promise that resolves when the offsets have been committed.
    */
   async commitOffsets(topicPartitions = null) {

lib/kafkajs/_producer.js

@@ -501,8 +501,19 @@ class Producer {
   }
 
   /**
-   *   send(record: ProducerRecord): Promise<RecordMetadata[]>
+   * Check if there is an ongoing transaction.
+   *
+   * NOTE: Since Producer itself represents a transaction, and there is no distinct
+   *       type for a transaction, this method exists on the producer.
+   * @returns {boolean} true if there is an ongoing transaction, false otherwise.
+   */
+  isActive() {
+    return this.#ongoingTransaction;
+  }
 
+  /**
+   * Sends a record of messages to a specific topic.
+   *
    * @param {import('../../types/kafkajs').ProducerRecord} sendOptions - The record to send. The keys `acks`, `timeout`, and `compression` are not used, and should not be set, rather, they should be set in the global config.
    * @returns {Promise<import("../../types/kafkajs").RecordMetadata[]>} Resolves with the record metadata for the messages.
    */
@@ -572,15 +583,15 @@ class Producer {
     const topicPartitionRecordMetadata = new Map();
     for (const recordMetadata of recordMetadataArr) {
       const key = `${recordMetadata.topicName},${recordMetadata.partition}`;
-      if (recordMetadata.baseOffset == null || !topicPartitionRecordMetadata.has(key)) {
+      if (recordMetadata.baseOffset === null || !topicPartitionRecordMetadata.has(key)) {
         topicPartitionRecordMetadata.set(key, recordMetadata);
         continue;
       }
 
       const currentRecordMetadata = topicPartitionRecordMetadata.get(key);
 
       // Don't overwrite a null baseOffset
-      if (currentRecordMetadata.baseOffset == null) {
+      if (currentRecordMetadata.baseOffset === null) {
         continue;
       }
 
@@ -598,7 +609,9 @@ class Producer {
   }
 
   /**
-    * sendBatch(batch: ProducerBatch): Promise<RecordMetadata[]>
+    * Sends a record of messages to various topics.
+    *
+    * NOTE: This method is identical to calling send() repeatedly and waiting on all the return values together.
     * @param {import('../../types/kafkajs').ProducerBatch} sendOptions - The record to send. The keys `acks`, `timeout`, and `compression` are not used, and should not be set, rather, they should be set in the global config.
     * @returns {Promise<import("../../types/kafkajs").RecordMetadata[]>} Resolves with the record metadata for the messages.
   */

test/promisified/producer/concurrentTransaction.spec.js

@@ -34,6 +34,7 @@ describe('Producer > Transactional producer', () => {
                 producer1 = newProducer();
                 await producer1.connect();
                 const transaction1 = await producer1.transaction();
+                expect(transaction1.isActive()).toBe(true);
                 await transaction1.send({ topic: topicName, messages: [message] });
 
                 // Producer 2 starts with the same transactional id to cause the concurrent transactions error