在Kafkajs中实现Single Partition Assigner遇到问题求助

你的自定义分区分配器未生效，核心问题集中在以下几点：

关键问题点 #

• 硬编码memberId：assign方法里写死了memberId: 'what'，但该ID是Kafka集群动态分配给消费者的固定标识，硬编码会导致分配逻辑与实际消费者实例不匹配。
• this指向错误：你通过箭头函数返回分配器对象，protocol和assign方法中的this.version、this.name无法正确指向分配器自身属性，会引发协议版本、名称不匹配的问题。
• 忽略assign方法入参：自定义分配器的assign方法会接收集群成员、主题分区等关键参数，你完全未利用这些参数，导致分配逻辑脱离实际集群状态。

修复后的自定义分配器代码 #

import { Kafka, PartitionAssigner, AssignerProtocol, KafkaMessage } from 'kafkajs'
import { Buffer } from 'buffer'

const kafka = new Kafka({
    clientId: 'my-app',
    brokers: ['localhost:9092']
})

// 修正后的SinglePartitionAssigner
export const SinglePartitionAssigner: PartitionAssigner = () => {
    const name = 'SinglePartitionAssigner'
    const version = 1

    return {
        name,
        version,
        async assign({ members, topicPartitions }) {
            // 匹配当前消费者的memberId
            const currentMember = members.find(member => member.clientId === 'my-app')
            if (!currentMember) return []

            // 仅分配testTopic的第0个分区给当前消费者
            return [
                {
                    memberId: currentMember.memberId,
                    memberAssignment: AssignerProtocol.MemberAssignment.encode({
                        version,
                        assignment: {
                            'testTopic': [0]
                        },
                        userData: Buffer.from([])
                    })
                }
            ]
        },
        protocol({ topics }) {
            return {
                name,
                metadata: AssignerProtocol.MemberMetadata.encode({
                    version,
                    topics,
                    userData: Buffer.from([])
                }),
            }
        }
    }
}

const consumer = kafka.consumer({
    groupId: 'ashutosh',
    partitionAssigners: [SinglePartitionAssigner]
})

async function sendEvent(message: KafkaMessage) {
    console.log({
        key: message.key?.toString(),
        value: message.value?.toString(),
        headers: message.headers,
    });
}

async function main() {
    await consumer.connect();
    await consumer.subscribe({
        topic: 'testTopic',
        fromBeginning: true
    })
    await consumer.run({
        eachMessage: async ({ message }) => sendEvent(message),
    });
}

main().catch(error => {
    console.error(error);
})

更简洁的替代方案 #

若你的需求仅为消费指定单个分区，无需自定义分配器，直接在subscribe时指定分区即可：

await consumer.subscribe({
    topic: 'testTopic',
    fromBeginning: true,
    partition: 0 // 直接指定目标分区
})

内容的提问来源于stack exchange，提问作者Ashutosh Pandey