Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Two kafka sources with different consumer group names #198

Open
butorine opened this issue Oct 4, 2020 · 0 comments
Open

Two kafka sources with different consumer group names #198

butorine opened this issue Oct 4, 2020 · 0 comments
Labels
enhancement New feature or request M-kafka This issue is related to the kafka module P2 Priority 2

Comments

@butorine
Copy link
Contributor

butorine commented Oct 4, 2020

We have a service that is reading from two kafka topics on the same kafka cluster using two different consumer groups. This is needed because topics are encoded differently. It's configured like so:

    const kafkaJsonConfig: IKafkaBrokerConfiguration & IKafkaSubscriptionConfiguration = {
        broker: "commonBroker",
        group: "consumerGroup_json",
        topics: "topic1",
        encoder: ProtoJsonMessageEncoder(),
    };

    const kafkaProtoConfig: IKafkaBrokerConfiguration & IKafkaSubscriptionConfiguration = {
        broker: "commonBroker",
        group: "consumerGroup_proto",
        topics: "topic2",
        encoder: ProtoMessageEncoder(),
    };
    ...
    .input()
        .add(kafkaSource(kafkaJsonConfig))
        .add(kafkaSource(kafkaProtoConfig))
        .done()
   ...

This works well most of the time, but when kafka cluster goes down and then recovers, this service looses connection to the cluster and never recovers like other more traditional services. The errors we see are like this:

{"level":"ERROR","timestamp":"2020-09-27T13:12:11.194Z","logger":"kafkajs","message":"[Connection] Response JoinGroup(key: 11, version: 2)","broker":"[broker]","clientId":"[client ID]","error":"The coordinator is loading and hence can't process requests for this group","correlationId":0,"size":24}
{"level":"ERROR","timestamp":"2020-09-27T13:12:11.194Z","logger":"kafkajs","message":"[Consumer] Crash: KafkaJSProtocolError: The coordinator is loading and hence can't process requests for this group","groupId":"consumerGroup_json","stack":"KafkaJSProtocolError: The coordinator is loading and hence can't process requests for this group
at createErrorFromCode (/usr/src/node_modules/kafkajs/src/protocol/error.js:537:10)
at Object.parse (/usr/src/node_modules/kafkajs/src/protocol/requests/joinGroup/v0/response.js:37:11)
at Connection.send (/usr/src/node_modules/kafkajs/src/network/connection.js:311:35)
at runMicrotasks (<anonymous>)
at processTicksAndRejections (internal/process/task_queues.js:93:5)
at async Broker.joinGroup (/usr/src/node_modules/kafkajs/src/broker/index.js:351:12)
at async ConsumerGroup.join (/usr/src/node_modules/kafkajs/src/consumer/consumerGroup.js:93:23)
at async /usr/src/node_modules/kafkajs/src/consumer/runner.js:51:9
at async Runner.start (/usr/src/node_modules/kafkajs/src/consumer/runner.js:105:7)
at async start (/usr/src/node_modules/kafkajs/src/consumer/index.js:230:7)"}
@plameniv plameniv added the M-kafka This issue is related to the kafka module label Jun 10, 2021
@plameniv plameniv added enhancement New feature or request P2 Priority 2 labels Jul 27, 2021
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
enhancement New feature or request M-kafka This issue is related to the kafka module P2 Priority 2
Development

No branches or pull requests

2 participants