Remove kafkaJsCompatibilityMode variable and assume true

Author: milindl

lib/kafkajs/_consumer.js

@@ -213,18 +213,17 @@ class Consumer {
    */
   #running = false;
 
-  /**
-   * Whether the consumer is in KafkaJS compatibility mode.
-   * @type {boolean}
-   */
-  #kafkaJSCompatibilityMode = false;
-
   /**
    * The message cache for KafkaJS compatibility mode.
    * @type {MessageCache|null}
    */
   #messageCache = null;
 
+  /**
+   * Whether the user has enabled manual offset management (stores).
+   */
+  #userManagedStores = false;
+
   /**
    * @constructor
    * @param {import("../../types/kafkajs").ConsumerConfig} kJSConfig
@@ -343,12 +342,13 @@ class Consumer {
         try {
           if (err.code === LibrdKafkaError.codes.ERR__ASSIGN_PARTITIONS) {
 
-            if (this.#checkPendingSeeks && this.#kafkaJSCompatibilityMode)
+            const checkPendingSeeks = this.#pendingSeeks.size !== 0;
+            if (checkPendingSeeks)
               assignment = this.#assignAsPerSeekedOffsets(assignment);
 
             this.#internalClient.assign(assignment);
 
-            if (this.#checkPendingSeeks) {
+            if (checkPendingSeeks) {
               const offsetsToCommit = assignment
                 .filter((topicPartition) => topicPartition.offset !== undefined)
                 .map((topicPartition) => ({
@@ -491,11 +491,6 @@ class Consumer {
       throw new error.KafkaJSError(CompatibilityErrorMessages.runOptionsAutoCommitThreshold(), { code: error.ErrorCodes.ERR__NOT_IMPLEMENTED });
     }
 
-    /* Offset storage management is manual in kafkaJS compability mode if auto-commit is turned on (ie by default). */
-    if (rdKafkaConfig['enable.auto.commit']) {
-      rdKafkaConfig['enable.auto.offset.store'] = false;
-    }
-
     /* Set the logger */
     if (Object.hasOwn(kjsConfig, 'logger')) {
       this.#logger = kjsConfig.logger;
@@ -517,8 +512,6 @@ class Consumer {
      * log level, as librdkafka will control the granularity. */
     if (!compatibleConfig || Object.keys(compatibleConfig).length === 0) {
       this.#logger.setLogLevel(logLevel.DEBUG);
-    } else {
-      this.#kafkaJSCompatibilityMode = true;
     }
 
     /* Even if we are in compability mode, setting a 'debug' in the main config must override the logger's level. */
@@ -542,6 +535,21 @@ class Consumer {
     }
     rdKafkaConfig['rebalance_cb'] = this.#rebalanceCallback.bind(this);
 
+    /* Offset management is different from case to case.
+     * Case 1: User has changed value of enable.auto.offset.store. In this case, we respect that.
+     * Case 2: automatic committing is on. In this case, we turn off auto.offset.store and store offsets manually.
+     *         this is necessary for cache invalidation and management, as we want to put things into the store
+     *         after eachMessage is called, and not on consume itself.
+     * Case 3: automatic committing is off. In this case, we turn off auto.offset.store too. Since the user might
+     *         call an empty commit() and expect things to work properly (ie. the right offsets be stored).
+     * All this works out a singular, simple condition.
+     */
+    if (!Object.hasOwn(this.#userConfig, 'enable.auto.offset.store')) {
+      rdKafkaConfig['enable.auto.offset.store'] = false;
+    } else {
+      this.#userManagedStores = !rdKafkaConfig['enable.auto.offset.store'];
+    }
+
     return rdKafkaConfig;
   }
 
@@ -699,10 +707,7 @@ class Consumer {
     }
 
     const rdKafkaConfig = this.#config();
-
-    if (this.#kafkaJSCompatibilityMode) {
-      this.#messageCache = new MessageCache(Math.floor(rdKafkaConfig['max.poll.interval.ms'] * 0.8));
-    }
+    this.#messageCache = new MessageCache(Math.floor(rdKafkaConfig['max.poll.interval.ms'] * 0.8));
 
     this.#state = ConsumerState.CONNECTING;
     this.#internalClient = new RdKafka.KafkaConsumer(rdKafkaConfig);
@@ -804,10 +809,6 @@ class Consumer {
       throw new error.KafkaJSError(CompatibilityErrorMessages.runOptionsPartitionsConsumedConcurrently(), { code: error.ErrorCodes.ERR__NOT_IMPLEMENTED });
     }
 
-    if (!this.#kafkaJSCompatibilityMode) {
-      throw new error.KafkaJSError('run() can only be used in KafkaJS compatibility mode.', { code: error.ErrorCodes.ERR__NOT_IMPLEMENTED });
-    }
-
     if (this.#running) {
       throw new error.KafkaJSError('Consumer is already running.', { code: error.ErrorCodes.ERR__STATE });
     }
@@ -829,7 +830,7 @@ class Consumer {
       /* Invalidate the message cache if needed. */
       if (this.#messageCache.isStale()) {
         await this.#clearCacheAndResetPositions(true);
-        this.#lock.release();
+        await this.#lock.release();
         continue;
       }
 
@@ -841,7 +842,7 @@ class Consumer {
       });
 
       if (!m) {
-        this.#lock.release();
+        await this.#lock.release();
         continue;
       }
 
@@ -876,7 +877,7 @@ class Consumer {
          * This is especially true since the pattern of pause() followed by throwing an error
          * is encouraged. To meet the API contract, we seek one offset backward at this point (which
          * means seeking to the message offset). */
-        this.seek({
+        await this.seek({
           topic: m.topic,
           partition: m.partition,
           offset: m.offset,
@@ -886,7 +887,7 @@ class Consumer {
       /* Store the offsets we need to store, or at least record them for cache invalidation reasons. */
       if (eachMessageProcessed) {
         try {
-          if (this.#internalConfig['enable.auto.commit']) {
+          if (!this.#userManagedStores) {
             this.#internalClient.offsetsStore([{ topic: m.topic, partition: m.partition, offset: Number(m.offset) + 1 }]);
           }
           this.#lastConsumedOffsets.set(`${m.topic}|${m.partition}`, Number(m.offset) + 1);
@@ -916,14 +917,12 @@ class Consumer {
 
   /**
    * Consumes a single message from the consumer within the given timeout.
+   * THIS METHOD IS NOT IMPLEMENTED.
    * @note This method cannot be used with run(). Either that, or this must be used.
    *
    * @param {any} args
    * @param {number} args.timeout - the timeout in milliseconds, defaults to 1000.
    * @returns {import("../..").Message|null} a message, or null if the timeout was reached.
-   *
-   * @note This API is currently in an experimental stage and subject to change.
-   *       This should not be used in KafkaJS compatibility mode (ie with kafkaJS blocks in the config).
    */
   async consume({ timeout } = { timeout: 1000 }) {
     if (this.#state !== ConsumerState.CONNECTED) {
@@ -934,10 +933,6 @@ class Consumer {
       throw new error.KafkaJSError('consume() and run() cannot be used together.', { code: error.ErrorCodes.ERR__CONFLICT });
     }
 
-    if (this.#kafkaJSCompatibilityMode) {
-      throw new error.KafkaJSError('consume() cannot be used in KafkaJS compatibility mode.', { code: error.ErrorCodes.ERR__NOT_IMPLEMENTED });
-    }
-
     this.#internalClient.setDefaultConsumeTimeout(timeout);
     let m = null;
 
@@ -948,7 +943,8 @@ class Consumer {
       this.#internalClient.setDefaultConsumeTimeout(undefined);
     }
 
-    return m ?? null;
+    throw new error.KafkaJSError('consume() is not implemented.' + m, { code: error.ErrorCodes.ERR__NOT_IMPLEMENTED });
+    // return m ?? null;
   }
 
   async #commitOffsetsUntilNoStateErr(offsetsToCommit) {
@@ -1043,8 +1039,7 @@ class Consumer {
       /* We need a complete reset of the cache if we're seeking to a different offset even for one partition.
        * At a later point, this may be improved at the cost of added complexity of maintaining message generation,
        * or else purging the cache of just those partitions which are seeked. */
-      if (this.#kafkaJSCompatibilityMode)
-        await this.#clearCacheAndResetPositions(true);
+      await this.#clearCacheAndResetPositions(true);
 
       /* It's assumed that topicPartition is already assigned, and thus can be seeked to and committed to.
        * Errors are logged to detect bugs in the internal code. */
@@ -1064,7 +1059,7 @@ class Consumer {
     }
 
     /* Offsets are committed on seek only when in compatibility mode. */
-    if (offsetsToCommit.length !== 0 && this.#internalConfig['enable.auto.commit'] && this.#kafkaJSCompatibilityMode) {
+    if (offsetsToCommit.length !== 0 && this.#internalConfig['enable.auto.commit']) {
       await this.#commitOffsetsUntilNoStateErr(offsetsToCommit);
     }
 
@@ -1078,7 +1073,7 @@ class Consumer {
    * If at any time, the consumer is assigned the partition, the seek will be performed.
    * Depending on the value of the librdkafka property 'enable.auto.commit', the consumer will commit the offset seeked to.
    * @param {import("../../types/kafkajs").TopicPartitionOffset} topicPartitionOffset
-   * @returns {Promise<void>|null} a promise that resolves when the seek has been performed (only when not in compatibility mode), or null (when in compatibility mode)
+   * @returns {Promise<void>|null} a promise that resolves when the seek has been performed.
    */
   seek(topicPartitionOffset) {
     if (this.#state !== ConsumerState.CONNECTED) {
@@ -1098,12 +1093,6 @@ class Consumer {
 
     this.#checkPendingSeeks = true;
     this.#pendingSeeks.set(`${rdKafkaTopicPartitionOffset.topic}|${rdKafkaTopicPartitionOffset.partition}`, rdKafkaTopicPartitionOffset.offset);
-
-    /* Immediately realize the seek if we're not in compatibility mode. And clear pending seeks.
-     * We don't need them for rebalance. */
-    if (!this.#kafkaJSCompatibilityMode) {
-      return this.#seekInternal().then(() => this.#pendingSeeks.clear());
-    }
   }
 
   async describeGroup() {
@@ -1160,8 +1149,7 @@ class Consumer {
       return;
     }
     this.#internalClient.pause(topics);
-    if (this.#kafkaJSCompatibilityMode)
-      this.#messageCache.stale = true;
+    this.#messageCache.stale = true;
 
     topics.map(JSON.stringify).forEach(topicPartition => this.#pausedPartitions.add(topicPartition));