Remove store from promisified API

Author: milindl

deps/librdkafka

@@ -125,11 +125,6 @@ class Consumer {
    */
   #messageCache = null;
 
-  /**
-   * Whether the user has enabled manual offset management (stores).
-   */
-  #userManagedStores = false;
-
   /**
    * Whether the user has enabled manual offset management (commits).
    */
@@ -536,20 +531,14 @@ class Consumer {
     rdKafkaConfig['offset_commit_cb'] = true;
     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'];
+    /* We handle offset storage within the promisified API by ourselves. Thus we don't allow the user to change this
+     * setting and set it to false. */
+    if (Object.hasOwn(this.#userConfig, 'enable.auto.offset.store')) {
+      throw new error.KafkaJSError(
+        "Changing 'enable.auto.offset.store' is unsupported while using the promisified API.",
+        { code: error.ErrorCodes.ERR__INVALID_ARG });
     }
+    rdKafkaConfig['enable.auto.offset.store'] = false;
 
     if (!Object.hasOwn(rdKafkaConfig, 'enable.auto.commit')) {
       this.#autoCommit = true; /* librdkafka default. */
@@ -663,13 +652,11 @@ class Consumer {
     payload._lastResolvedOffset = { offset, leaderEpoch };
 
     try {
-      if (!this.#userManagedStores) {
-        this.#internalClient._offsetsStoreSingle(
-          topic,
-          partition,
-          offset + 1,
-          leaderEpoch);
-      }
+      this.#internalClient._offsetsStoreSingle(
+        topic,
+        partition,
+        offset + 1,
+        leaderEpoch);
       this.#lastConsumedOffsets.set(key, offset + 1);
     } catch (e) {
       /* Not much we can do, except log the error. */
@@ -1076,9 +1063,7 @@ class Consumer {
     /* Store the offsets we need to store, or at least record them for cache invalidation reasons. */
     if (eachMessageProcessed) {
       try {
-        if (!this.#userManagedStores) {
-          this.#internalClient._offsetsStoreSingle(m.topic, m.partition, Number(m.offset) + 1, m.leaderEpoch);
-        }
+        this.#internalClient._offsetsStoreSingle(m.topic, m.partition, Number(m.offset) + 1, m.leaderEpoch);
         this.#lastConsumedOffsets.set(partitionKey(m), Number(m.offset) + 1);
       } catch (e) {
         /* Not much we can do, except log the error. */
@@ -1299,30 +1284,6 @@ class Consumer {
     // return m ?? null;
   }
 
-  /**
-   * Store offsets for the given topic partitions.
-   *
-   * Stored offsets will be commited automatically at a later point if autoCommit is enabled.
-   * 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").TopicPartitionOffset[]?} topicPartitions
-   */
-  storeOffsets(topicPartitions) {
-    if (this.#state !== ConsumerState.CONNECTED) {
-      throw new error.KafkaJSError('Store can only be called while connected.', { code: error.ErrorCodes.ERR__STATE });
-    }
-
-    if (!this.#userManagedStores) {
-      throw new error.KafkaJSError(
-        'Store can only be called when enable.auto.offset.store is explicitly set to false.', { code: error.ErrorCodes.ERR__INVALID_ARG });
-    }
-
-    const topicPartitionsRdKafka = topicPartitions.map(
-      topicPartitionOffsetMetadataToRdKafka);
-    this.#internalClient.offsetsStore(topicPartitionsRdKafka);
-  }
-
   async #commitOffsetsUntilNoStateErr(offsetsToCommit) {
     let err = { code: error.ErrorCodes.ERR_NO_ERROR };
     do {