Make subscriptions replaceable with flag

Author: milindl

lib/kafkajs/_consumer.js

@@ -73,6 +73,12 @@ class Consumer {
    */
   #pausedPartitions = new Set();
 
+  /**
+   * Contains a list of stored topics/regexes that the user has subscribed to.
+   * @type {(string|RegExp)[]}
+   */
+  #storedSubscriptions = [];
+
   /**
    * @constructor
    * @param {import("../../types/kafkajs").ConsumerConfig} kJSConfig
@@ -425,7 +431,8 @@ class Consumer {
       }
     });
 
-    this.#internalClient.subscribe(topics);
+    this.#storedSubscriptions =  subscription.replace ? topics : this.#storedSubscriptions.concat(topics);
+    this.#internalClient.subscribe(this.#storedSubscriptions);
   }
 
   async stop() {

test/promisified/consumer/pause.spec.js

@@ -68,8 +68,8 @@ describe('Consumer', () => {
             /* Send the first 2 messages to each topic. */
             for (const topic of topics) {
                 await producer.send({ topic, messages: messages.slice(0, 2) });
+                await consumer.subscribe({ topic });
             }
-            await consumer.subscribe({ topics: topics });
 
             let shouldPause = true;
             let pauseMessageRecvd = false;
@@ -141,7 +141,7 @@ describe('Consumer', () => {
             for (const topic of topics) {
                 await producer.send({ topic, messages: messages.slice(0, 2) })
             }
-            await consumer.subscribe({ topics })
+            await consumer.subscribe({ topics, replace: true });
 
             let shouldPause = true
             const messagesConsumed = []

test/promisified/consumer/subscribe.spec.js

@@ -71,7 +71,7 @@ describe('Consumer', () => {
             expect(messagesConsumed.map(m => m.message.value.toString())).toEqual(
                 expect.arrayContaining(['drink', 'your', 'ovaltine'])
             );
-        }, 10000);
+        });
     })
 
     describe('Deprecated "topic" interface', () => {
@@ -151,7 +151,7 @@ describe('Consumer', () => {
                         'value-br',
                         'value-se',
                     ]);
-                }, 10000);
+                });
             });
         });
     });

types/kafkajs.d.ts

@@ -376,9 +376,9 @@ export type EachBatchHandler = (payload: EachBatchPayload) => Promise<void>
 
 export type EachMessageHandler = (payload: EachMessagePayload) => Promise<void>
 
-export type ConsumerSubscribeTopic = { topic: string | RegExp; fromBeginning?: boolean }
+export type ConsumerSubscribeTopic = { topic: string | RegExp; fromBeginning?: boolean, replace?: boolean }
 
-export type ConsumerSubscribeTopics = { topics: (string | RegExp)[]; fromBeginning?: boolean }
+export type ConsumerSubscribeTopics = { topics: (string | RegExp)[]; fromBeginning?: boolean, replace?: boolean }
 
 export type ConsumerRunConfig = {
   autoCommit?: boolean