The dependency async was updated from 3.1.0 to 3.1.1.

🚨 View failing branch.

This version is covered by your current version range and after updating it in your project the build failed.

async is a direct dependency of this project, and it is very likely causing it to break. If other packages depend on yours, this update is probably also breaking those in turn.

Your Greenkeeper Bot 🌴

The devDependency eslint was updated from 5.15.1 to 5.15.2.

🚨 View failing branch.

This version is covered by your current version range and after updating it in your project the build failed.

eslint is a devDependency of this project. It might not break your production code or affect downstream projects, but probably breaks your build or test tools, which may prevent deploying or publishing.

Your Greenkeeper Bot 🌴

Right now if you try to connect with the consumer and it cannot for whatever reason the Consumer never throws an error so there is no way to know that this happened and is indistinguishable from just waiting.

You should be able to provide a connection timeout parameter and have the lib throw an error when that timeout span has elapsed

Hi,

thanks for the awesome library! The default Kafka Java client is using the Murmur2 Hash function for it's partitioning algorithm. Since our system uses different clients, for us it's important, that they use consistent partitioning functions. Unfortunately neither library, nor node-rdkafka, nor librdkafka implement's the algorithm used at the default Java client. Although there is a pull request for a Murmur2 partitioner pending at librdkafka (confluentinc/librdkafka#1468), we are going to implement our own partitioning function and maybe switch to the librdkafka implementation, once it's live and supported by node-sinek and the underlaying library node-rdkafka.
Is a pull request for node-sinek, supporting the Murmur2 partitioning function welcome if well implemented? If yes we are going to fork this library and then send a pull request, otherwise we are going to solve this problem on application level.

Thank you,
Marlon

I'm using the NConsumer as suggested with the following settings
batchSize: 3,
noBatchCommits: true (I want to commit every batch regardless of how many messages it handled, e.g. batchSize is 3, messages are only 2, I want to commit if it handled 2)
manualBatching: true (I want all the messages of the batch to be passed on my sync event)

The behaviour I'm seeing is that I have 3 messages in my topic. I would expect to have my syncevent called exactly once with three messages. I see the code and I discover the _firstMessageConsumed and the first-drain-message. So what this does is,

• it calls my syncevent one time for the first message and a second time with all three messages including the first message.
• After every batch (regardless of actual messages length) I commit manually (synchronously).

So I have three messages, and batchSize 3.

• syncvent called with the first message (let's call it A)
• commit
• check the kafka offset in the gui and the current offset from 0 goes to 3 (although I handled one message)
• syncevent called with all three messages (including message A)

This is very odd. What am I missing?

1. Why the first message is being passed two times?
2. Why the current offset of the consumer goes to 3 when I only handled one message?

The devDependency sinon was updated from 7.2.3 to 7.2.4.

🚨 View failing branch.

This version is covered by your current version range and after updating it in your project the build failed.

sinon is a devDependency of this project. It might not break your production code or affect downstream projects, but probably breaks your build or test tools, which may prevent deploying or publishing.

Your Greenkeeper Bot 🌴

Here is the fix:

1. xcode-select --install
2. (make sure your have the xcode app installed, if not get it from the appstore)
3. brew reinstall openssl (optional)
4. LDFLAGS='-L/usr/local/opt/openssl/lib' CPPFLAGS='-I/usr/local/opt/openssl/include' yarn

OpenSSL might be installed but is not linked anymore when updating to mojave, this way yarn or npm wont be able to install (compile) node-rdkafka via node-gyp. By passing the flags you can solve this issue.

The devDependency mocha was updated from 6.2.0 to 6.2.1.

🚨 View failing branch.

This version is covered by your current version range and after updating it in your project the build failed.

mocha is a devDependency of this project. It might not break your production code or affect downstream projects, but probably breaks your build or test tools, which may prevent deploying or publishing.

Your Greenkeeper Bot 🌴

For example here:

The dependency lodash.merge was updated from 4.6.1 to 4.6.2.

🚨 View failing branch.

This version is covered by your current version range and after updating it in your project the build failed.

lodash.merge is a direct dependency of this project, and it is very likely causing it to break. If other packages depend on yours, this update is probably also breaking those in turn.

Your Greenkeeper Bot 🌴

This part of the connect readme shows an error being passed to callback (albeit null); however, the callback which is passed in the implementation doesn't handle any arguments. Am I missing something? Should it accept errors?

Hi,

I really like your package, great work. It would be even more pleasant if the Publisher class had an auto-flush option when using bufferPublishMessage. That way I would not have to trace the total number of messages myself.

Any chance of adding that feature?

Cheers!

I am using sinek library producer to connect to kafka which is on SASL + SSL. But while connecting producer I am getting below error.

Error: Invalid value "sasl_ssl" for configuration property "security.protocol"

The dependency bluebird was updated from 3.5.2 to 3.5.3.

🚨 View failing branch.

This version is covered by your current version range and after updating it in your project the build failed.

bluebird is a direct dependency of this project, and it is very likely causing it to break. If other packages depend on yours, this update is probably also breaking those in turn.

Your Greenkeeper Bot 🌴

Unhandled rejection Error: Usage of syncEvent is not permitted in streaming mode.

When passing an empty array (removing all subscriptions) to NConsumer#adjustSubscription() the following error occurs.

 Error: Local: Invalid argument or configuration
      at Function.createLibrdkafkaError [as create] (node_modules/node-rdkafka/lib/error.js:260:10)
      at KafkaConsumer.Client._errorWrap (node_modules/node-rdkafka/lib/client.js:470:29)
      at KafkaConsumer.subscribe (node_modules/node-rdkafka/lib/kafka-consumer.js:290:8)
      at NConsumer.adjustSubscription (node_modules/sinek/lib/librdkafka/NConsumer.js:669:19)

I worked around this by calling the unsubscribe() method of the underlying consumer - like adjustSubscription() does before setting the new subscription - in case the new topic list is emtpy.

Thanks and have a nice day!

Opening an issue here to have discussions on moving from yarn to npm.

Some benefits of yarn:

• Pinned Version Resolutions
• Speed
• yarn2 - yarnpkg/yarn#6953

Hi @krystianity it seems I'm losing data. I was hoping to use node-sinek after reading this statement:

auto-commit / manual drain commit in backpressure-mode (dont loose data)

My use-case is a minute-long process for every message. It's important for me to commit every time I finish processing a single message and not before I'm done processing. Currently because of Drainer._q I see I'm buffering tens of messages and the queue is only drained and committed after many are processed. If I try to consumer.consumer._commit then all of my queued messages are committed.

Hi there!

I'm looking into processing partitions in parallel, but each message in a partition one at a time in order.

In the README, I see:

js
const options = {
  batchSize: 500, //grab up to 500 messages per batch round
  commitEveryNBatch: 5, //commit all offsets on every 5th batch
  concurrency: 2, //calls synFunction in parallel * 2 for messages in batch
  // ...
};

myNConsumer.consume(syncFunction, true, false, options);

If I have 4 partitions, will the concurrency: 2 setting mean that I risk processing 2 messages from the same partition in parallel?

Thanks!

Are you open to someone contributing Typescript definitions to sinek?

I have a kafka cluster that is configured using SSL. And what I have for properties are:

• SSL_TRUSTSTORE_LOCATION
• SSL_TRUSTSTORE_PASSWORD
• BROKER_LIST (with SSL://<...> based URLs)

I'm not entirely clear how I would setup a client to consume from that kafka cluster...

Hi.

Sinek could produce incorrect error messages while loading native rdkafka client. E.g. here https://github.com/nodefluent/node-sinek/blob/master/lib/librdkafka/NProducer.js#L73 original error is swallowed and new custom error is thrown.

However as rdkafka is native module it could throw not only "Module not found" exceptions. It may also throw exceptions like: "Error relocating... symbol not found". And this is important to preserve original error message or at least show it in debug.

IT would be nice to have a delivery report emitted on NProducer. Currently it just logs.

In node-sinek-master/lib/librdkafka/NConsumer.js

    lagFetchInterval = lagFetchInterval || 1e6 * 60 * 3; //3 minutes

    this._analyticsIntv = setInterval(this._runAnalytics.bind(this), analyticsInterval);
    this._lagCheckIntv = setInterval(this._runLagCheck.bind(this), lagFetchInterval);

setInterval and setTimeout use milliseconds, not microseconds. Using a parameter like 45000000 will mean "run this every 45000 seconds".

The optionalDependency node-rdkafka was updated from 2.4.1 to 2.4.2.

🚨 View failing branch.

This version is covered by your current version range and after updating it in your project the build failed.

node-rdkafka is a optionalDependency of this project. It might not break your production code or affect downstream projects, but probably breaks your build or test tools, which may prevent deploying or publishing.

Your Greenkeeper Bot 🌴

The dependency uuid was updated from 3.3.2 to 3.3.3.

🚨 View failing branch.

This version is covered by your current version range and after updating it in your project the build failed.

uuid is a direct dependency of this project, and it is very likely causing it to break. If other packages depend on yours, this update is probably also breaking those in turn.

Your Greenkeeper Bot 🌴

... example: The message with the key "thingäöüÄÖÜ" with a partition count of 10 should be written to the partition 2, but gets written to the partition 4 by the current implementation. Although this is not used here, it is due to the same bug described in this issue: b3nj4m/murmurhash-js#1

I propose to use the npm package murmur-hash-js instead of murmurhash and pass a buffer instead of a string, which leads to results, consistent with Kafka's Java implementation.

The bug described got tested with this bug (#39) and that for assuming it was already resolved.

For reference see the following, tested, implementation: https://github.com/vuza/murmur2-partitioner

I'm playing around with your module, but when I try the sasl-ssl-example by calling

DEBUG=* node producer.js

I'm getting back:

sinek:error Error: Local: Broker transport failure
sinek:error at Function.createLibrdkafkaError [as create] (/media/gitprodenv/Elements1/Coding/ApacheKafka/node-sinek/examples/sasl-ssl-example/node_modules/node-rdkafka/lib/error.js:334:10)
sinek:error at /media/gitprodenv/Elements1/Coding/ApacheKafka/node-sinek/examples/sasl-ssl-example/node_modules/node-rdkafka/lib/client.js:339:28 +0ms
sinek:error Error: Local: Broker transport failure
sinek:error at Function.createLibrdkafkaError [as create] (/media/gitprodenv/Elements1/Coding/ApacheKafka/node-sinek/examples/sasl-ssl-example/node_modules/node-rdkafka/lib/error.js:334:10)
sinek:error at /media/gitprodenv/Elements1/Coding/ApacheKafka/node-sinek/examples/sasl-ssl-example/node_modules/node-rdkafka/lib/client.js:339:28 +2ms

Apache Kafka and Zookeeper is set up by calling npm run kafka:start as described in the docs.

Any help is highly appreciated!

Version 3.0.1 of kafka-node was just published.

This version is covered by your current version range and after updating it in your project the build failed.

kafka-node is a direct dependency of this project, and it is very likely causing it to break. If other packages depend on yours, this update is probably also breaking those in turn.

Your Greenkeeper Bot 🌴

Good to see an example for javascript client subscribing to more then one topic like it is possible in kafka-node lib.

Hi!
Is there a specific reason, why you overwrite the commit callback? How can I receive commit errors? Wouldn't it be nice to pass an optional callback to the commit methods, which overwrites the default callback? Or am I overlooking something?

Hi,

I am using sinek's NConsumer to connect to Kafka and came across faulty behavior when trying to disconnect from the broker. It seems that 8fd1846 introduced a race condition between the disconnected event listener (when it calls NConsumer#reset) and checking the abort condition in NConsumer#_singeConsumeRecursive.

The result of this is, that the NConsumer continues trying to read messages from the disconnected node-rdkafka consumer.

{ Error: KafkaConsumer is not connected\n    at Function.createLibrdkafkaError [as create] (/home/node/event-client/node_modules/node-rdkafka/lib/error.js:261:10)\n    at /home/node/event-client/node_modules/node-rdkafka/lib/kafka-consumer.js:448:29\n  message: 'KafkaConsumer is not connected',\n  code: -172,\n  errno: -172,\n  origin: 'kafka' }

Thanks in advance!

The dependency bluebird was updated from 3.5.3 to 3.5.4.

🚨 View failing branch.

This version is covered by your current version range and after updating it in your project the build failed.

bluebird is a direct dependency of this project, and it is very likely causing it to break. If other packages depend on yours, this update is probably also breaking those in turn.

Your Greenkeeper Bot 🌴

The dependency bluebird was updated from 3.7.1 to 3.7.2.

🚨 View failing branch.

This version is covered by your current version range and after updating it in your project the build failed.

bluebird is a direct dependency of this project, and it is very likely causing it to break. If other packages depend on yours, this update is probably also breaking those in turn.

Your Greenkeeper Bot 🌴

The dependency node-rdkafka was updated from 2.7.0 to 2.7.1.

🚨 View failing branch.

This version is covered by your current version range and after updating it in your project the build failed.

node-rdkafka is a direct dependency of this project, and it is very likely causing it to break. If other packages depend on yours, this update is probably also breaking those in turn.

Your Greenkeeper Bot 🌴

Thank you for implementing both the murmur2 and 3 partitioning logic. Unfortunately it does not match the default Java Kafka client's logic.

Example: Key "wu" with a partition count of 10 should be written to partition 0, the current logic returns 5.

The Java client's seed for the murmur2 algorithm wasn't provided, see the Java implementation: https://github.com/apache/kafka/blob/1.0/clients/src/main/java/org/apache/kafka/common/utils/Utils.java#L355

For murmur2, changing https://github.com/nodefluent/node-sinek/blob/master/lib/librdkafka/NProducer.js#L275 to return this._murmur(key, 0x9747b28c) % partitionCount; fixes the issue.

For reference, I've implemented a JS murmur2 partitioner including a testing, testing against Java's partitioner results: https://github.com/vuza/murmur2-partitioner

Here is what I'm trying to do:

I have a topic recipes(3 partitions) and I have 2 consumer groups sweet and savoury. When I'm working with sweet I don't want to consume / process / commit any messages that belong to savoury I simply want to skip that message and vice versa.

Here is my code:

recipesListener('group-sweet', 'recipes', 'sweet');
recipesListener('group-savoury', 'recipes', 'savoury');

const { NConsumer } = require("sinek");
const { logger } = require("../logger");

const recipesListener = ({ groupId, topicName, groupType }) => {
  const consumerConfig = {
    logger: {
      debug: (msg) => logger.debug(msg),
      info: (msg) => logger.info(msg),
      warn: (msg) => logger.warn(msg),
      error: (msg) => logger.error(msg),
    },
    noptions: {
      "metadata.broker.list": "localhost:9092",
      "group.id": groupId,
      "socket.keepalive.enable": true,
      "queued.min.messages": 1000,
      "queued.max.messages.kbytes": 5000,
      "fetch.message.max.bytes": 524288,
    },
    tconf: {
      "auto.offset.reset": "latest",
    },
  };

  (async () => {
    const consumer = new NConsumer([topicName], consumerConfig);

    consumer.on("error", (error) => logger.error(error));
    consumer.on("ready", () =>
      logger.info(`Consumer ready to consume ${topicName} for ${groupId}`)
    );

    await consumer.connect();

    consumer.consume(
      async (messages, callback) => {
        /*
         *   messages are produced with headers that can have either sweet or savoury
         */
        if (messages.headers !== groupType) {
          return;
        }

        callback();
      },
      true,
      false,
      { noBatchCommits: true }
    );
  })().catch(logger.error);
};

module.exports = {
  recipesListener,
};

Currently the below lines halts the consumer completely. Is there a workaround for this? Please let me know.

        /*
         *   messages are produced with headers that can have either sweet or savoury
         */
        if (messages.headers !== groupType) {
          return;
        }

I'm using

Node v12.8.2
"node-rdkafka": "^2.9.1",
"sinek": "^9.1.0",

I'm a Kafka noob. Please let me know if any more details are needed.

Hello @krystianity
Can you please provide any pointers on how to implement backpressure with this client ? We would like to control how many messages we can consume. Is there way to commit them later on once processing is done and get more messages ?

Is it possible to "seek" to the beginning of a topic for a particular consumer group and partition?

The use case is being able to reprocess messages from the start. 😄

The dependency debug was updated from 4.1.0 to 4.1.1.

🚨 View failing branch.

This version is covered by your current version range and after updating it in your project the build failed.

debug is a direct dependency of this project, and it is very likely causing it to break. If other packages depend on yours, this update is probably also breaking those in turn.

Your Greenkeeper Bot 🌴

The dependency kafka-node was updated from 4.0.2 to 4.0.3.

🚨 View failing branch.

This version is covered by your current version range and after updating it in your project the build failed.

kafka-node is a direct dependency of this project, and it is very likely causing it to break. If other packages depend on yours, this update is probably also breaking those in turn.

Your Greenkeeper Bot 🌴

based on: https://kafka.js.org/docs/configuration and tulios/kafkajs#298
We may not have the correct settings for the JSConsumer and JSProducer. This issue is to ensure we have them up to date after #154 has been merged

Hello.

I'm wondering of how to terminate a consumer gracefully, for instance, syncEvent is still running, but consumer.close() has called.

I couldn't find what waits for running tasks until finishing below code.

Is there any code of termination gracefully in rdkafka or something else? Or does it just let them terminate without graceful termination? or would I implement code that clean the resources before shutdown for myself?

Thank you!

The devDependency sinon was updated from 7.3.1 to 7.3.2.

🚨 View failing branch.

This version is covered by your current version range and after updating it in your project the build failed.

sinon is a devDependency of this project. It might not break your production code or affect downstream projects, but probably breaks your build or test tools, which may prevent deploying or publishing.

Your Greenkeeper Bot 🌴

This might very well be the most advanced client for JS/TS. However, to be honest, it's very hard even to get started due to the near-total lack of documentation. You should really take a week off from writing code and just work on docs, in my not so humble opinion. For instance, I struggle to see the difference between a Producer's send, buffer and bufferFormat methods. Please update the docs so we can use this driver!

Apparently there's a lack of good drivers for Kafka in the Node world in general, so if this is as good as you claim, please write documentation which shows it and underpins it.

 if(this._isAutoCommitting !== null && typeof this._isAutoCommitting !== "undefined"){
          this.config.logger.warn("enable.auto.commit has no effect in 1:n consume-mode, set to null or undefined to remove this message." +
            "You can pass 'noBatchCommits' as true via options to .consume(), if you want to commit manually.");
        }

I'm passing noBatchCommits as true in the consume options and I do not pass anything for
enable.auto.commit. If I pass it as null, then kafka responds with error.

NConsumer send: async send(topicName, message, _partition = null, _key = null, _partitionKey = null, _opaqueKey = null) {

rdkafka produce: Producer.prototype.produce = function(topic, partition, message, key, timestamp, opaque, headers) {

There is no way to send custom headers although the underlying implementation supports it.

I am replacing kafka-node with node-sinek in a project,

We had some fail on error logic, (- Netflix philosophy, let the pod die in an error scenario)

My calls to process.exit() do not seem to cause the node application to exit any more..

I am using NConsumer
Up until the point where I call the NConsumer constructor, I can exit the process as normal..

Thanks in advance,
D

https://github.com/nodefluent/node-sinek/blob/master/lib/librdkafka/Health.js#L155