Kafkajs: Consumer error - Cannot read property 'every' of undefined

[paambaati]

I'm on KafkaJS 1.4.4 connecting to a 5-node Kafka 1.1.0 cluster. I have an app that is subscribed to 2 topics (some_topic with 50 partitions and some_other_topics with 400 partitions), and I'm getting this error —

{"level":"DEBUG","timestamp":"2018-11-28T08:08:42.131Z","logger":"kafkajs","message":"[Connection] Connecting","broker":"kafka1.com:9092","clientId":"kafka-reader","ssl":false,"sasl":false}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:42.656Z","logger":"kafkajs","message":"[Connection] Request ApiVersions(key: 18, version: 2)","broker":"kafka1.com:9092","clientId":"kafka-reader","correlationId":0,"expectResponse":true,"size":40}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:42.888Z","logger":"kafkajs","message":"[Connection] Response ApiVersions(key: 18, version: 2)","broker":"kafka1.com:9092","clientId":"kafka-reader","error":"The version of API is not supported","correlationId":0,"payload":{"type":"Buffer","data":"[filtered]"}}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:42.889Z","logger":"kafkajs","message":"[Connection] Request ApiVersions(key: 18, version: 1)","broker":"kafka1.com:9092","clientId":"kafka-reader","correlationId":1,"expectResponse":true,"size":40}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:43.122Z","logger":"kafkajs","message":"[Connection] Response ApiVersions(key: 18, version: 1)","broker":"kafka1.com:9092","clientId":"kafka-reader","correlationId":1,"size":272,"data":{"errorCode":0,"apiVersions":[{"apiKey":0,"minVersion":0,"maxVersion":5},{"apiKey":1,"minVersion":0,"maxVersion":7},{"apiKey":2,"minVersion":0,"maxVersion":2},{"apiKey":3,"minVersion":0,"maxVersion":5},{"apiKey":4,"minVersion":0,"maxVersion":1},{"apiKey":5,"minVersion":0,"maxVersion":0},{"apiKey":6,"minVersion":0,"maxVersion":4},{"apiKey":7,"minVersion":0,"maxVersion":1},{"apiKey":8,"minVersion":0,"maxVersion":3},{"apiKey":9,"minVersion":0,"maxVersion":3},{"apiKey":10,"minVersion":0,"maxVersion":1},{"apiKey":11,"minVersion":0,"maxVersion":2},{"apiKey":12,"minVersion":0,"maxVersion":1},{"apiKey":13,"minVersion":0,"maxVersion":1},{"apiKey":14,"minVersion":0,"maxVersion":1},{"apiKey":15,"minVersion":0,"maxVersion":1},{"apiKey":16,"minVersion":0,"maxVersion":1},{"apiKey":17,"minVersion":0,"maxVersion":1},{"apiKey":18,"minVersion":0,"maxVersion":1},{"apiKey":19,"minVersion":0,"maxVersion":2},{"apiKey":20,"minVersion":0,"maxVersion":1},{"apiKey":21,"minVersion":0,"maxVersion":0},{"apiKey":22,"minVersion":0,"maxVersion":0},{"apiKey":23,"minVersion":0,"maxVersion":0},{"apiKey":24,"minVersion":0,"maxVersion":0},{"apiKey":25,"minVersion":0,"maxVersion":0},{"apiKey":26,"minVersion":0,"maxVersion":0},{"apiKey":27,"minVersion":0,"maxVersion":0},{"apiKey":28,"minVersion":0,"maxVersion":0},{"apiKey":29,"minVersion":0,"maxVersion":0},{"apiKey":30,"minVersion":0,"maxVersion":0},{"apiKey":31,"minVersion":0,"maxVersion":0},{"apiKey":32,"minVersion":0,"maxVersion":1},{"apiKey":33,"minVersion":0,"maxVersion":0},{"apiKey":34,"minVersion":0,"maxVersion":0},{"apiKey":35,"minVersion":0,"maxVersion":0},{"apiKey":36,"minVersion":0,"maxVersion":0},{"apiKey":37,"minVersion":0,"maxVersion":0},{"apiKey":38,"minVersion":0,"maxVersion":0},{"apiKey":39,"minVersion":0,"maxVersion":0},{"apiKey":40,"minVersion":0,"maxVersion":0},{"apiKey":41,"minVersion":0,"maxVersion":0},{"apiKey":42,"minVersion":0,"maxVersion":0}],"throttleTime":0}}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:43.123Z","logger":"kafkajs","message":"[Connection] Connecting","broker":"kafka1.com:9092","clientId":"kafka-reader","ssl":false,"sasl":false}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:43.346Z","logger":"kafkajs","message":"[Connection] Request ApiVersions(key: 18, version: 2)","broker":"kafka1.com:9092","clientId":"kafka-reader","correlationId":0,"expectResponse":true,"size":40}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:43.574Z","logger":"kafkajs","message":"[Connection] Response ApiVersions(key: 18, version: 2)","broker":"kafka1.com:9092","clientId":"kafka-reader","error":"The version of API is not supported","correlationId":0,"payload":{"type":"Buffer","data":"[filtered]"}}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:43.575Z","logger":"kafkajs","message":"[Connection] Request ApiVersions(key: 18, version: 1)","broker":"kafka1.com:9092","clientId":"kafka-reader","correlationId":1,"expectResponse":true,"size":40}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:43.809Z","logger":"kafkajs","message":"[Connection] Response ApiVersions(key: 18, version: 1)","broker":"kafka1.com:9092","clientId":"kafka-reader","correlationId":1,"size":272,"data":{"errorCode":0,"apiVersions":[{"apiKey":0,"minVersion":0,"maxVersion":5},{"apiKey":1,"minVersion":0,"maxVersion":7},{"apiKey":2,"minVersion":0,"maxVersion":2},{"apiKey":3,"minVersion":0,"maxVersion":5},{"apiKey":4,"minVersion":0,"maxVersion":1},{"apiKey":5,"minVersion":0,"maxVersion":0},{"apiKey":6,"minVersion":0,"maxVersion":4},{"apiKey":7,"minVersion":0,"maxVersion":1},{"apiKey":8,"minVersion":0,"maxVersion":3},{"apiKey":9,"minVersion":0,"maxVersion":3},{"apiKey":10,"minVersion":0,"maxVersion":1},{"apiKey":11,"minVersion":0,"maxVersion":2},{"apiKey":12,"minVersion":0,"maxVersion":1},{"apiKey":13,"minVersion":0,"maxVersion":1},{"apiKey":14,"minVersion":0,"maxVersion":1},{"apiKey":15,"minVersion":0,"maxVersion":1},{"apiKey":16,"minVersion":0,"maxVersion":1},{"apiKey":17,"minVersion":0,"maxVersion":1},{"apiKey":18,"minVersion":0,"maxVersion":1},{"apiKey":19,"minVersion":0,"maxVersion":2},{"apiKey":20,"minVersion":0,"maxVersion":1},{"apiKey":21,"minVersion":0,"maxVersion":0},{"apiKey":22,"minVersion":0,"maxVersion":0},{"apiKey":23,"minVersion":0,"maxVersion":0},{"apiKey":24,"minVersion":0,"maxVersion":0},{"apiKey":25,"minVersion":0,"maxVersion":0},{"apiKey":26,"minVersion":0,"maxVersion":0},{"apiKey":27,"minVersion":0,"maxVersion":0},{"apiKey":28,"minVersion":0,"maxVersion":0},{"apiKey":29,"minVersion":0,"maxVersion":0},{"apiKey":30,"minVersion":0,"maxVersion":0},{"apiKey":31,"minVersion":0,"maxVersion":0},{"apiKey":32,"minVersion":0,"maxVersion":1},{"apiKey":33,"minVersion":0,"maxVersion":0},{"apiKey":34,"minVersion":0,"maxVersion":0},{"apiKey":35,"minVersion":0,"maxVersion":0},{"apiKey":36,"minVersion":0,"maxVersion":0},{"apiKey":37,"minVersion":0,"maxVersion":0},{"apiKey":38,"minVersion":0,"maxVersion":0},{"apiKey":39,"minVersion":0,"maxVersion":0},{"apiKey":40,"minVersion":0,"maxVersion":0},{"apiKey":41,"minVersion":0,"maxVersion":0},{"apiKey":42,"minVersion":0,"maxVersion":0}],"throttleTime":0}}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:43.890Z","logger":"kafkajs","message":"[Connection] Request Metadata(key: 3, version: 4)","broker":"kafka1.com:9092","clientId":"kafka-reader","correlationId":2,"expectResponse":true,"size":65}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:44.123Z","logger":"kafkajs","message":"[Connection] Response Metadata(key: 3, version: 4)","broker":"kafka1.com:9092","clientId":"kafka-reader","correlationId":2,"size":1744,"data":{"throttleTime":0,"brokers":[{"nodeId":2,"host":"kafka2.com","port":9092,"rack":null},{"nodeId":5,"host":"kafka5.com","port":9092,"rack":null},{"nodeId":4,"host":"kafka4.com","port":9092,"rack":null},{"nodeId":1,"host":"kafka1.com","port":9092,"rack":null},{"nodeId":3,"host":"kafka3.com","port":9092,"rack":null}],"clusterId":"7QRYYnVdScOZOZEEj9pdpw","controllerId":3,"topicMetadata":[{"topicErrorCode":0,"topic":"some_topic","isInternal":false,"partitionMetadata":[{"partitionErrorCode":0,"partitionId":23,"leader":3,"replicas":[3,4],"isr":[3]},{"partitionErrorCode":0,"partitionId":41,"leader":1,"replicas":[1,2],"isr":[1]},{"partitionErrorCode":0,"partitionId":32,"leader":2,"replicas":[2,5],"isr":[2]},{"partitionErrorCode":0,"partitionId":44,"leader":4,"replicas":[4,5],"isr":[4]},{"partitionErrorCode":0,"partitionId":8,"leader":3,"replicas":[3,5],"isr":[3]},{"partitionErrorCode":0,"partitionId":35,"leader":5,"replicas":[5,4],"isr":[5]},{"partitionErrorCode":0,"partitionId":17,"leader":2,"replicas":[2,1],"isr":[2]},{"partitionErrorCode":0,"partitionId":26,"leader":1,"replicas":[1,3],"isr":[1]},{"partitionErrorCode":0,"partitionId":11,"leader":1,"replicas":[1,4],"isr":[1]},{"partitionErrorCode":0,"partitionId":38,"leader":3,"replicas":[3,2],"isr":[3]},{"partitionErrorCode":0,"partitionId":47,"leader":2,"replicas":[2,4],"isr":[2]},{"partitionErrorCode":0,"partitionId":29,"leader":4,"replicas":[4,1],"isr":[4]},{"partitionErrorCode":0,"partitionId":20,"leader":5,"replicas":[5,1],"isr":[5]},{"partitionErrorCode":0,"partitionId":2,"leader":2,"replicas":[2,3],"isr":[2]},{"partitionErrorCode":0,"partitionId":5,"leader":5,"replicas":[5,2],"isr":[5]},{"partitionErrorCode":0,"partitionId":14,"leader":4,"replicas":[4,2],"isr":[4]},{"partitionErrorCode":0,"partitionId":46,"leader":1,"replicas":[1,3],"isr":[1]},{"partitionErrorCode":0,"partitionId":40,"leader":5,"replicas":[5,1],"isr":[5]},{"partitionErrorCode":0,"partitionId":49,"leader":4,"replicas":[4,1],"isr":[4]},{"partitionErrorCode":0,"partitionId":13,"leader":3,"replicas":[3,1],"isr":[3]},{"partitionErrorCode":0,"partitionId":4,"leader":4,"replicas":[4,5],"isr":[4]},{"partitionErrorCode":0,"partitionId":31,"leader":1,"replicas":[1,4],"isr":[1]},{"partitionErrorCode":0,"partitionId":22,"leader":2,"replicas":[2,3],"isr":[2]},{"partitionErrorCode":0,"partitionId":16,"leader":1,"replicas":[1,5],"isr":[1]},{"partitionErrorCode":0,"partitionId":7,"leader":2,"replicas":[2,4],"isr":[2]},{"partitionErrorCode":0,"partitionId":43,"leader":3,"replicas":[3,4],"isr":[3]},{"partitionErrorCode":0,"partitionId":25,"leader":5,"replicas":[5,2],"isr":[5]},{"partitionErrorCode":0,"partitionId":34,"leader":4,"replicas":[4,2],"isr":[4]},{"partitionErrorCode":0,"partitionId":10,"leader":5,"replicas":[5,3],"isr":[5]},{"partitionErrorCode":0,"partitionId":37,"leader":2,"replicas":[2,1],"isr":[2]},{"partitionErrorCode":0,"partitionId":1,"leader":1,"replicas":[1,2],"isr":[1]},{"partitionErrorCode":0,"partitionId":28,"leader":3,"replicas":[3,5],"isr":[3]},{"partitionErrorCode":0,"partitionId":19,"leader":4,"replicas":[4,3],"isr":[4]},{"partitionErrorCode":0,"partitionId":45,"leader":5,"replicas":[5,2],"isr":[5]},{"partitionErrorCode":0,"partitionId":36,"leader":1,"replicas":[1,5],"isr":[1]},{"partitionErrorCode":0,"partitionId":27,"leader":2,"replicas":[2,4],"isr":[2]},{"partitionErrorCode":0,"partitionId":9,"leader":4,"replicas":[4,1],"isr":[4]},{"partitionErrorCode":0,"partitionId":18,"leader":3,"replicas":[3,2],"isr":[3]},{"partitionErrorCode":0,"partitionId":48,"leader":3,"replicas":[3,5],"isr":[3]},{"partitionErrorCode":0,"partitionId":21,"leader":1,"replicas":[1,2],"isr":[1]},{"partitionErrorCode":0,"partitionId":3,"leader":3,"replicas":[3,4],"isr":[3]},{"partitionErrorCode":0,"partitionId":12,"leader":2,"replicas":[2,5],"isr":[2]},{"partitionErrorCode":0,"partitionId":30,"leader":5,"replicas":[5,3],"isr":[5]},{"partitionErrorCode":0,"partitionId":39,"leader":4,"replicas":[4,3],"isr":[4]},{"partitionErrorCode":0,"partitionId":15,"leader":5,"replicas":[5,4],"isr":[5]},{"partitionErrorCode":0,"partitionId":42,"leader":2,"replicas":[2,3],"isr":[2]},{"partitionErrorCode":0,"partitionId":6,"leader":1,"replicas":[1,3],"isr":[1]},{"partitionErrorCode":0,"partitionId":33,"leader":3,"replicas":[3,1],"isr":[3]},{"partitionErrorCode":0,"partitionId":24,"leader":4,"replicas":[4,5],"isr":[4]},{"partitionErrorCode":0,"partitionId":0,"leader":5,"replicas":[5,1],"isr":[5]}]}]}}
{"level":"INFO","timestamp":"2018-11-28T08:08:44.124Z","logger":"kafkajs","message":"[Consumer] Starting","groupId":"kafka-reader"}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:44.153Z","logger":"kafkajs","message":"[Connection] Request GroupCoordinator(key: 10, version: 0)","broker":"kafka1.com:9092","clientId":"kafka-reader","correlationId":3,"expectResponse":true,"size":68}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:44.385Z","logger":"kafkajs","message":"[Connection] Response GroupCoordinator(key: 10, version: 0)","broker":"kafka1.com:9092","clientId":"kafka-reader","correlationId":3,"size":29,"data":{"errorCode":0,"coordinator":{"nodeId":3,"host":"kafka3.com","port":9092}}}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:44.385Z","logger":"kafkajs","message":"[Cluster] Found group coordinator","nodeId":3}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:44.386Z","logger":"kafkajs","message":"[Connection] Connecting","broker":"kafka3.com:9092","clientId":"kafka-reader","ssl":false,"sasl":false}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:44.669Z","logger":"kafkajs","message":"[Connection] Request JoinGroup(key: 11, version: 0)","broker":"kafka3.com:9092","clientId":"kafka-reader","correlationId":0,"expectResponse":true,"size":142}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:53.434Z","logger":"kafkajs","message":"[Connection] Response JoinGroup(key: 11, version: 0)","broker":"kafka3.com:9092","clientId":"kafka-reader","correlationId":0,"size":164,"data":{"errorCode":0,"generationId":47,"groupProtocol":"RoundRobinAssigner","leaderId":"kafka-reader-895368cf-bd5e-4a39-842e-9df4f98a14e2","memberId":"kafka-reader-d3fb14e2-4c54-4f30-8fb4-a2e8230a6575","members":[]}}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:53.451Z","logger":"kafkajs","message":"[Connection] Request SyncGroup(key: 14, version: 0)","broker":"kafka3.com:9092","clientId":"kafka-reader","correlationId":1,"expectResponse":true,"size":141}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:53.675Z","logger":"kafkajs","message":"[Connection] Response SyncGroup(key: 14, version: 0)","broker":"kafka3.com:9092","clientId":"kafka-reader","correlationId":1,"size":89,"data":{"errorCode":0,"memberAssignment":{"type":"Buffer","data":[0,1,0,0,0,2,0,15,112,114,111,100,117,99,116,105,111,110,95,104,116,109,108,0,0,0,6,0,0,1,129,0,0,0,253,0,0,1,63,0,0,0,187,0,0,0,55,0,0,0,121,0,18,112,114,111,100,117,99,116,105,111,110,95,114,101,112,97,114,115,101,0,0,0,0,0,0,0,0]}}}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:53.676Z","logger":"kafkajs","message":"[ConsumerGroup] Received assignment","groupId":"kafka-reader","generationId":47,"memberId":"kafka-reader-d3fb14e2-4c54-4f30-8fb4-a2e8230a6575","memberAssignment":{"some_other_topic":[385,253,319,187,55,121],"some_topic":[]}}
{"level":"WARN","timestamp":"2018-11-28T08:08:53.676Z","logger":"kafkajs","message":"[ConsumerGroup] Consumer group received unsubscribed topics","groupId":"kafka-reader","generationId":47,"memberId":"kafka-reader-d3fb14e2-4c54-4f30-8fb4-a2e8230a6575","assignedTopics":["some_other_topic","some_topic"],"topicsSubscribed":["some_topic"],"topicsNotSubscribed":["some_other_topic"]}
{"level":"ERROR","timestamp":"2018-11-28T08:08:53.676Z","logger":"kafkajs","message":"[Consumer] Crash: TypeError: Cannot read property 'every' of undefined","groupId":"kafka-reader","stack":"TypeError: Cannot read property 'every' of undefined\n    at ConsumerGroup.sync (/Users/indix/Projects/firestar-service/service/node_modules/kafkajs/src/consumer/consumerGroup.js:172:65)\n    at process._tickCallback (internal/process/next_tick.js:68:7)"}
{"level":"INFO","timestamp":"2018-11-28T08:08:53.677Z","logger":"kafkajs","message":"[Consumer] Stopped","groupId":"kafka-reader"}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:53.677Z","logger":"kafkajs","message":"[Consumer] consumer has stopped, disconnecting","groupId":"kafka-reader"}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:53.677Z","logger":"kafkajs","message":"[Connection] disconnecting...","broker":"kafka1.com:9092","clientId":"kafka-reader"}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:53.677Z","logger":"kafkajs","message":"[Connection] disconnected","broker":"kafka1.com:9092","clientId":"kafka-reader"}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:53.677Z","logger":"kafkajs","message":"[Connection] disconnecting...","broker":"kafka3.com:9092","clientId":"kafka-reader"}
{"level":"DEBUG","timestamp":"2018-11-28T08:08:53.677Z","logger":"kafkajs","message":"[Connection] disconnected","broker":"kafka3.com:9092","clientId":"kafka-reader"}
{"level":"INFO","timestamp":"2018-11-28T08:10:45.884Z","logger":"kafkajs","message":"[Consumer] Stopped","groupId":"kafka-reader"}
{"level":"DEBUG","timestamp":"2018-11-28T08:10:45.884Z","logger":"kafkajs","message":"[Consumer] consumer has stopped, disconnecting","groupId":"kafka-reader"}
{"level":"DEBUG","timestamp":"2018-11-28T08:10:45.885Z","logger":"kafkajs","message":"[Connection] disconnecting...","broker":"kafka1.com:9092","clientId":"kafka-reader"}
{"level":"DEBUG","timestamp":"2018-11-28T08:10:45.885Z","logger":"kafkajs","message":"[Connection] disconnected","broker":"kafka1.com:9092","clientId":"kafka-reader"}

Relevant code

await producer.connect();
await consumer.connect();
const inputTopics = ['some_topic'];
for (const topic of inputTopics) {
    await consumer.subscribe({
        topic,
        fromBeginning: true,
    });
}

I face the error only when I run the above code; when I change inputTopics to [, 'some_topic', 'some_other_topic'] and re-run it, the app proceeds but only consumers from some_other_topic.

Somehow I can never consume from some_topic!

[Nevon]

This is a bug in the code where we try to handle the case where you receive assignments for a topic you are not subscribed to. I'll be creating a PR to handle this case shortly.

However, you shouldn't be receiving assignments for a topic you're not subscribed to in the first case, which leads me to believe that there is something wonky going on with your configuration. The last time we saw this (see #83), it was because the same groupId was being used for two different services, which meant that service A got assignments for a topic that service B was subscribed to. If that's the case, make sure you're not re-using the same groupId between different services.

[paambaati]

I'll be creating a PR to handle this case shortly.

@Nevon Thanks!

I have a few more updates -

1. I already have the app deployed, subscribed to 1 topic. I just ran a local copy of it (same groupId) subscribing to 2 topics and ran into this.
2. some_topic is actually empty, as all the old data got evicted by the retention policy; I verified this by running the console consumer.

I think this is no longer an issue, and I can close it; unless you want this to link to your upcoming PR?

[tulios]

@paambaati let's keep it open since this is a bug

I already have the app deployed, subscribed to 1 topic. I just ran a local copy of it (same groupId) subscribing to 2 topics and ran into this

This was probably the cause, but we shouldn't crash thus the fix.
Thanks again for the reporting!

[Nevon]

@paambaati: I just released v1.4.5 with a fix for this issue.