Node-rdkafka: Unknown Broker error when using seek

[FuzzOli87]

Using your docker-compose.yml file:

I run this producer to create this topic:

const kafka = require('node-rdkafka');

const producer = new kafka.Producer({
  'metadata.broker.list': 'localhost:9092',
  'broker.version.fallback': '0.10.0', // If kafka node doesn't have API, use this instead
  'api.version.request': true // Request the api version of Kafka node
});

producer.connect();
producer.on('ready', () => {
  console.log('producer is ready');
  try {
    producer.produce('TEST_TOPIC', null, Buffer.from(JSON.stringify({ test: 'test' })));
  } catch (e) {
    console.error(e);
  }
});

I then boot up my consumer:

const kafka = require('node-rdkafka');

const consumer = new kafka.KafkaConsumer({
  'group.id': 'consumer',
  'metadata.broker.list': 'localhost:9092',
  'broker.version.fallback': '0.10.0', // If kafka node doesn't have API, use this instead
  'api.version.request': true // Request the api version of Kafka node
});

consumer.connect();

consumer
  .on('ready', () => {
    console.log('consumer is ready');
    consumer.subscribe(['TEST_TOPIC']);

    consumer.seek(
      {
        topic: 'TEST_TOPIC',
        partition: 0,
        offset: 0
      },
      0,
      (err) => {
        console.log('seek', err);
        consumer.consume();
      }
    );
  })
  .on('data', (data) => {
    console.log('DATA RECEIVED', data);
    const dataMsg = JSON.parse(data.value.toString());
    console.log('parsed data', dataMsg);
  });

When I do this, I expect to process the first message that the producer created first. However, I instead get a "Unknown broker error".

Is this not how seek is meant to be used?

[FuzzOli87]

Update:

I installed from the repo and began to use librdkafka 11 because I noticed another issue mentioned it didn't make it to 9.5. I also updated my docker-compose.yml file to use kafka 11 as well.

What I'm seeing is that when I run the consumer now, I get an "unknown partition" error which I expect because there hasn't been any messages posted to the topic the consumer subscribed to. However, after I do post some messages and restart the consumer, it fails due to a segmentation fault error. This only happens AFTER I create a few messages.

[FuzzOli87]

Here is the DUMP for when I get the seg fault:

PID 40421 received SIGSEGV for address: 0x0
0   segfault-handler.node               0x00000001038f3178 _ZL16segfault_handleriP9__siginfoPv + 280
1   libsystem_platform.dylib            0x00007fff8e123b3a _sigtramp + 26
2   ???                                 0x0000700004bb3be0 0x0 + 123145381690336
3   node-librdkafka.node                0x0000000105e8aed5 rd_kafka_seek + 164
4   node-librdkafka.node                0x0000000105e75add _ZN7RdKafka17KafkaConsumerImpl4seekERKNS_14TopicPartitionEi + 115
5   node-librdkafka.node                0x0000000105e6ace0 _ZN9NodeKafka13KafkaConsumer4SeekERKN7RdKafka14TopicPartitionEi + 172
6   node-librdkafka.node                0x0000000105e73520 _ZN9NodeKafka7Workers17KafkaConsumerSeek7ExecuteEv + 46
7   node                                0x0000000100891b54 worker + 90
8   libsystem_pthread.dylib             0x00007fff8e12d93b _pthread_body + 180
9   libsystem_pthread.dylib             0x00007fff8e12d887 _pthread_body + 0
10  libsystem_pthread.dylib             0x00007fff8e12d08d thread_start + 13
[1]    40421 abort      node consumer.js

[webmakersteve]

Seek has been largely untested and is hidden behind a precompiler flag right now. I am going to investigate it more when I can upgrade to librdkafka 0.11 after a potential bug fix in the library.

I'll take a look at what you've done to see if I can reproduce it. It could have something to do with the way topics are getting passed into the function, or it could be something internal to the library.

[FuzzOli87]

@webmakersteve Thanks! When you say it's hidden, is there something I have to do extra when I install the library to unhide it?

[webmakersteve]

Should be nothing extra to do. It uses precompiler guards to ensure the function is defined.

I am in the middle of a few updates to the library - will add testing seek to my list before the next publish of a major version.

[webmakersteve]

I just upgrade librdkafka to 0.11 in the most recent 2.1.0 release. Can you try this again? I am in the process of adding tests around seek to ensure it works as expected.

[mimaison]

I tried with 2.1.1 and I see the same issue. It either segfaults or I get the same Unknown broker error message.

[webmakersteve]

Alright. I think this is fixed on master. The timeout was not being passed correctly, and when librdkafka is given a timeout of 0 it runs the request as an asynchronous one, so the cleanup I was doing to delete the topic partition object was causing problems.

If you try from master you can see if it fixes your issue.

[mimaison]

Thanks for taking a look!
I don't see the error anymore on MacOS (I've not checked if Ubuntu still segfaults) but it does not seem to work as expected (maybe my expectations are wrong ?).

Using:

console.log("Position before seek: " + JSON.stringify(consumer.position()));
consumer.seek({ topic: topicName, partition: 0, offset: 159588 }, 0, function(err) {
    if (err) {
        console.log(err);
    } else {
        console.log("Position after seek: " + JSON.stringify(consumer.position()));
    }
});

But the position printed is the same before and after.

[edenhill]

seek() may only be used for partitions that are currently being fetched, i.e., they've been assign()ed - either explicitly or implicitly (by subscription).

[battlecow]

The seek function no longer segfaults for me on master branch within a Debian docker thanks to the helpful suggestion above to subscribe to the topic first. The method works when creating utilizing a new group.id only it would seem, when I attempt to resuse the method on a current group.id (or re-run the same seek code) I get the error:
{ Error: Local: Erroneous state at Function.createLibrdkafkaError [as create] (/node_modules/node-rdkafka/lib/error.js:260:10) at /node_modules/node-rdkafka/lib/kafka-consumer.js:209:26 message: 'Local: Erroneous state', code: -172, errno: -172, origin: 'kafka' }

I am not familiar enough with Kafka and the seek function to know if this is intended or a side effect.

[loganbfisher]

@webmakersteve I am seeing the same issue as above. It is no longer seg faulting but I am getting the erroneous state error message. here is my code.

const connection = process.env.EVENT_STORE_CONNS;

const consumer = new kafka.KafkaConsumer(
  {
    "group.id": "testthis",
    "metadata.broker.list": connection
  },
  {}
);

// Non-flowing mode
consumer.connect();

consumer
  .on("ready", function() {
    console.log("READY TO CONSUME:");
    // Subscribe to the librdtesting-01 topic
    // This makes subsequent consumes read from that topic.
    consumer.subscribe(["Devices"]);

    consumer.seek({ topic: "Devices", partition: 0, offset: 0 }, 0, function(
      err,
      result
    ) {
      if (err) {
        console.log(err);
      }

      console.log("THIS IS THE MESSAGE: ", result);
    });
  })
  .on("data", function(data) {
    console.log("Message found!  Contents below.");
    console.log(data.value.toString());
  });

Here is my Dockerfile

FROM node:8.8.0
LABEL maintainer Daniel Olivares "[email protected]"

ENV TINI_VERSION v0.16.1
ADD https://github.com/krallin/tini/releases/download/${TINI_VERSION}/tini /tini
RUN chmod +x /tini

RUN mkdir /home/app

WORKDIR /home/app

COPY package.json package-lock.json ./

RUN npm install

ENTRYPOINT ["/tini", "--"]

CMD ["npm", "start"]

Is there anything you see wrong? Thanks for the help!

[grubed]

2018-02-06 13:51:02,575 ERROR 17438 nodejs.unhandledExceptionError: Unknown bro ker error
at Function.createLibrdkafkaError [as create] (/node_modules/node-rdkafka/lib/error.js:260:10)
at /test/node_modules/node-rdkafka/lib/kafka-consumer.js:441:29
message: 'Unknown broker error'
code: -1
errno: -1
origin: 'local'
name: 'unhandledExceptionError'
pid: 17438
hostname: hzgtlfrontend

node: v8.9.4
"node-rdkafka": "^2.2.1"
linux centos 7

[Pchelolo]

I'm seeing the same Local: Erroneous state issue. What I'm trying to do is to seek to the committed offsets when renaming a consumer group - so I create 2 consumers, subscribe both of them with different groups, take committed offsets for one group and transfer them using seek to another group.

[thebigredgeek]

Bump. Is this still an issue?

[SeanYHe]

I'm also getting the Local: Erroneous state issue. After some investigation it seems that calling subscribe doesn't set the assignments immediately, which causes the error in seek.

this.client.subscribe(topics);
console.log(this.client.subscription());
console.log(this.client.assignments());

output:

[ 'TOPIC1', 'TOPIC2' ]
[]

Any suggestions?

[meysamabl]

I was also getting the Local: Erroneous state. I realized that in case of implicit assign (subscribe), you would have to wait for rebalance_cb to kick off and then call seek. And in case of explicit assign, you would have to specify the offset in your topicPartition object.
Here is the sample solution I came up with:
Implicit call to assign:

const topics = ['test'];
const consumer = new Kafka.KafkaConsumer({
    'group.id': 'kafka',
    'enable.auto.commit': false,
    'enable.auto.offset.store': false,
    'auto.offset.reset': 'earliest',
    'metadata.broker.list': 'localhost:9092',
    'rebalance_cb': function(err, assignments) {
        if (err.code === Kafka.CODES.ERRORS.ERR__ASSIGN_PARTITIONS) {
           // stuff you wanna do before assignment
          this.assign(assignments);
          this.emit('rebalanced'); // emit event 
        } else if (err.code === Kafka.CODES.ERRORS.ERR__REVOKE_PARTITIONS) {
            // Same as above, this can throw if we are disconnected.
            this.unassign();
        } else {
            // We had a real error.
            console.error(err);
        }
    }
}, {});

consumer.connect();
consumer.on('ready', function() {
    consumer.subscribe(topics);
    // now listen to rebalanced event
    consumer.on('rebalanced', function () {
        consumer.seek({topic: 'test', partition: 0, offset: 100}, 10, function (err) {
            console.log(err);
        }).on('data', (data) => {
            console.log(`${data.topic}  :  ${data.offset}`);
        });
    });

    consumer.consume();
});

In explicit call to assign, you would need to specify the offset and then consume it, or use seek to change the offset (without specifying the offset in topicPartition object, you would still get an error if you call seek function):

const consumer = new Kafka.KafkaConsumer({
    'group.id': 'kafka',
    'enable.auto.commit': false,
    'enable.auto.offset.store': false,
    'auto.offset.reset': 'earliest',
    'metadata.broker.list': 'localhost:9092',
}, {});

consumer.connect();
consumer.on('ready', function() {
    consumer.assign([{topic: 'test', partition: 0, offset: 100}]);
   // this becomes unnecessary as we already explicitly assign
   // but still you can call seek with no error
   /* 
   consumer.seek({topic: 'test', partition: 0, offset: 100}, 10, function (err) {
            console.log(err);
    });
   */
    consumer.consume();
    consumer.on('data', (data) => {
            console.log(`${data.topic}  :  ${data.offset}`);
    })
});

I hope this helps.

Thanks

[stale[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs.