fix(consumer): drain in-flight bootstrap consume workers before unsubscribe

The bootstrap consumer uses setInterval(200ms) to call consume(1, cb),
creating multiple concurrent C++ async workers (each with a 1000ms
timeout). Since librdkafka 2.10.0, these workers survive an
unsubscribe→subscribe transition and can dequeue messages from the next
subscription, causing them to be lost to the normal consume pipeline.

Track in-flight workers with a counter and defer unsubscribe() until
all have completed.

Issue: BB-760

Author: delthas

lib/BackbeatConsumer.js

@@ -873,7 +873,32 @@ class BackbeatConsumer extends EventEmitter {
         let producer; // eslint-disable-line prefer-const
         let producerTimer;
         let consumerTimer;
+        let inFlightConsumes = 0;
+        // Wait for all in-flight consume(1, consumeCb) workers to complete
+        // before calling unsubscribe(). The setInterval(200ms) pattern can
+        // have multiple consume workers running concurrently in the C++
+        // thread pool (each with a 1000ms timeout). Since librdkafka 2.10.0,
+        // these workers survive an unsubscribe→subscribe transition and can
+        // steal messages from the next subscription.
+        function _finishBootstrap(partition, offset) {
+            if (inFlightConsumes > 0) {
+                setTimeout(() => _finishBootstrap(partition, offset), 50);
+                return;
+            }
+            self._consumer.offsetsStore([{
+                topic: self._topic,
+                partition,
+                offset,
+            }]);
+            self._consumer.commit();
+            self._consumer.unsubscribe();
+            producer.close(() => {
+                self._bootstrapping = false;
+                self._onReady();
+            });
+        }
         function consumeCb(err, messages) {
+            inFlightConsumes--;
             if (err) {
                 return undefined;
             }
@@ -888,17 +913,7 @@ class BackbeatConsumer extends EventEmitter {
                             { topic: self._topic, groupId: self._groupId });
                         clearInterval(producerTimer);
                         clearInterval(consumerTimer);
-                        self._consumer.offsetsStore([{
-                            topic: self._topic,
-                            partition: message.partition,
-                            offset: message.offset + 1,
-                        }]);
-                        self._consumer.commit();
-                        self._consumer.unsubscribe();
-                        producer.close(() => {
-                            self._bootstrapping = false;
-                            self._onReady();
-                        });
+                        _finishBootstrap(message.partition, message.offset + 1);
                     }
                 }
             });
@@ -928,6 +943,7 @@ class BackbeatConsumer extends EventEmitter {
                         this._consumer.on('ready', () => {
                             this._consumer.subscribe([this._topic]);
                             consumerTimer = setInterval(() => {
+                                inFlightConsumes++;
                                 this._consumer.consume(1, consumeCb);
                             }, 200);
                         });