consumer.js

// This will do the 'while(true)' trick for you.
// The callback should return a promise, that's fulfilled when this batch processing is done
// However, I'm not 100% sure this is the API we want. I would better see the Consumer inherit 
// from an EventEmitter, and emit `message` message for each consumed message, and `error` for each error,
// same as kafka-node does. It feels more natural to me, but it's a bigger discussion.

class Consumer {
    constructor(rawConsumer) {
        this.rawConsumer = rawConsumer;
        this.isStarted = false;
    }

    pollAsync(timeout) {
        return new Promise((resolve) => {
            resolve(rawConsumer.poll(timeout));
        });
    }

    start(callback) {
        this.isStarted = true;

        const fetch = () => {
            return this.pollAsync(1000)
            .then(callback)
            .catch((e) => console.log(e))
            .finally(() => {
                if (this.isStarted) {
                    return fetch();
                }
            });
        };

        fetch();
    }

    stop() {
        this.isStarted = false;
    }
}