Skip to content

Commit

Permalink
Provide the _subscribed_ topics to the protocol() function
Browse files Browse the repository at this point in the history
The `topics` field may change depending on the actually assigned topics, and we might effectively
lose topics here when the assigner tries to build a full view of the topics used by the members.

See also https://kafkajs.slack.com/archives/CF6RFPF6K/p1572986319131500
  • Loading branch information
ankon committed Jan 2, 2020
1 parent afc283f commit c995a18
Showing 1 changed file with 5 additions and 1 deletion.
6 changes: 5 additions & 1 deletion src/consumer/consumerGroup.js
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,11 @@ module.exports = class ConsumerGroup {
sessionTimeout,
rebalanceTimeout,
memberId: this.memberId || '',
groupProtocols: this.assigners.map(assigner => assigner.protocol({ topics: this.topics })),
groupProtocols: this.assigners.map(assigner =>
assigner.protocol({
topics: this.topicsSubscribed,
})
),
})

this.generationId = groupData.generationId
Expand Down

0 comments on commit c995a18

Please sign in to comment.