HighLevelConsumer.pause() does not work

[sergeyjsg]

"use strict";

let kafka = require('kafka-node');
var conf = require('./providers/Config');

let client = new kafka.Client(conf.kafka.connectionString, conf.kafka.clientName);
let consumer = new kafka.HighLevelConsumer(client, [ { topic: conf.kafka.readTopic } ], { groupId: conf.kafka.clientName, paused: true });

let threads = 0;

consumer.on('message', function(message) {
    threads++;

    if (threads > 10) consumer.pause();

    if (threads > 50) process.exit(1);

    console.log(threads + " >>> " + message.value);
});

consumer.resume();

So I see 50 messages in console and process exits by termination statement

[shaharmor]

Each bulk load is fetching more than 50 messages, so pause will not stop those messages, only future fetches

[sergeyjsg]

I understand the behaviour. But code above - should work. It have to fixed.

[BadLambdaJamma]

Pause and Resume are a dicey way to control concurrency. Make sure your not pausing and resuming during a rebalance!

[sergeyjsg]

And what is the best approach to control concurrency? Obviously my writer is slower then reader. What should I do in context of kafka? And how can I control rebalance?

[BadLambdaJamma]

Rebalance occurs with a high level consumer, whenever any consumer starts. Set the message size to control how many messages.

consumer.on('rebalancing', function () {
self.rebalancing = true;
});

[BadLambdaJamma]

Pauses only stops after the first fetch request. so if you have a a bunch of messages at the broker when the client first connects....... you'll exceed your limit. you can throttle the first request by message size.......