meitu/go-consumergroup

golang kafka consumer group

go-consumergroup's Issues

zk 路径问题

如果使用了自己定义的路径，怎么解决，在zk_grou_stroage中固定了这些位置
const (
ownerPath = "/consumers/%s/owners/%s/%d"
consumersDir = "/consumers/%s/ids"
consumersPath = "/consumers/%s/ids/%s"
offsetsPath = "/consumers/%s/offsets/%s/%d"
brokersDir = "/brokers/ids"
brokersPath = "/brokers/ids/%s"
)

如果kafka的资源位于自定义路径下,特别是zk是公共zk，怎么处理
例如/zk_kafka/consumers%s/owners...

Topic was not found in consumergroup

卡夫卡0.11 版本遇到一个错误

go 编译提示

duplicate symbol _ZSTD_freeDCtx in:
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000026.o
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000071.o
duplicate symbol _ZSTD_initStaticDCtx in:
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000026.o
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000071.o
duplicate symbol _ZSTD_sizeof_DCtx in:
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000026.o
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000071.o
duplicate symbol _ZSTD_DCtx_refPrefix in:
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000026.o
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000071.o
duplicate symbol _ZSTD_DCtx_reset in:
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000026.o
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000071.o
duplicate symbol _ZSTD_getDictID_fromDict in:
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000026.o
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000071.o
duplicate symbol _ZSTD_decompress_usingDict in:
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000026.o
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000071.o
duplicate symbol _ZSTD_decompressBegin_usingDict in:
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000026.o
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000071.o
duplicate symbol _ZSTD_initDStream_usingDict in:
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000026.o
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000071.o
duplicate symbol _ZSTD_getDictID_fromDDict in:
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000026.o
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000071.o
duplicate symbol _ZSTD_decompress_usingDDict in:
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000026.o
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000071.o
duplicate symbol _ZSTD_decompressBegin_usingDDict in:
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000026.o
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000071.o
duplicate symbol _ZSTD_initDStream_usingDDict in:
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000026.o
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000071.o
duplicate symbol _ZSTD_DCtx_refDDict in:
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000026.o
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000071.o
duplicate symbol _ZSTD_createDDict in:
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000026.o
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000071.o
duplicate symbol _ZSTD_freeDDict in:
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000026.o
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000071.o
duplicate symbol _ZSTD_initStaticDDict in:
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000026.o
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000071.o
duplicate symbol _ZSTD_sizeof_DDict in:
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000026.o
/var/folders/7g/71m476gx15l0lf147pjmtrlm0000gp/T/go-link-968281994/000071.o

Rebalance when the partition's num was changed

setup a goroutine to watch topic partition's zk dir children changed

问一下，启动了两个consumergroup进程接收消息

问一下，启动了两个consumergroup进程接收消息，消息只打到一个进程里，另一个进程没有任何消息，这是什么原因，怎么才能把消息平均到两个consumer进程里

bug: failed restart the consumer while the partition consumer was releasing

the produce steps:

1. current consumer group was assigned partition [0,3]
2. trigger rebalance
3.  current consumer group was assigned partition [0]
4. releasing partition but didn't complete, and new consumer was started, try to create the partition consumer

got error:

Kafka: invalid configuration (That topic/partition is already being consumed)

[version 0.1.0]About "consumergroup" usage in goroutine

Example code in this repo works fine on Kafka cluster, but gist doesn't work well

I new all the staff in main goroutine by calliing function.
Having closure-style go routines (using for-select) selects message events.
got these logs on stdout.

I wonder if I meet usage problem in my code

[INFO]  [go-consumergroup] [cccc, ni.rx] start to consume
[INFO]  [go-consumergroup] [cccc, ni.rx] partitions [] are assigned to this consumer
[INFO]  [go-consumergroup] [cccc, ni.rx] stop to consume
[INFO]  [go-consumergroup] [cccc, ni.rx] partitions [] are assigned to this consumer
[INFO]  [go-consumergroup] [cccc, ni.rx] stop to consume
[INFO]  [go-consumergroup] [cccc, ni.rx] partitions [] are assigned to this consumer
[INFO]  [go-consumergroup] [cccc, ni.rx] stop to consume
[INFO]  [go-consumergroup] [cccc, ni.rx] start to consume
[INFO]  [go-consumergroup] [cccc, ni.rx] start to consume
[INFO]  [go-consumergroup] [cccc, ni.rx] partitions [] are assigned to this consumer
[INFO]  [go-consumergroup] [cccc, ni.rx] stop to consume
[INFO]  [go-consumergroup] [cccc, ni.rx] partitions [] are assigned to this consumer
[INFO]  [go-consumergroup] [cccc, ni.rx] stop to consume
[INFO]  [go-consumergroup] [cccc, ni.rx] start to consume
[INFO]  [go-consumergroup] [cccc, ni.rx] partitions [] are assigned to this consumer
[INFO]  [go-consumergroup] [cccc, ni.rx] stop to consume
[INFO]  [go-consumergroup] [cccc, ni.rx] start to consume
[INFO]  [go-consumergroup] [cccc, ni.rx] start to consume
[INFO]  [go-consumergroup] [cccc, ni.rx] start to consume
[INFO]  [go-consumergroup] [cccc, ni.rx] partitions [] are assigned to this consumer
[INFO]  [go-consumergroup] [cccc, ni.rx] stop to consume
[INFO]  [go-consumergroup] [cccc, ni.rx] partitions [] are assigned to this consumer
[INFO]  [go-consumergroup] [cccc, ni.rx] stop to consume
[INFO]  [go-consumergroup] [cccc, ni.rx] partitions [] are assigned to this consumer

反馈一个bug, 指定了一个consumer group, 但之后收到的数据不全，怀疑partition分配的不对

反馈一个bug, 指定了一个consumer group, 但之后收到的数据不全，怀疑partition分配的不对。我有两个consumer节点，每个节点都只有一个consumergroup的连接，接收数据不全之后，我换了一个新的consumer group名字，之后两个节点收到的数据就完整了。这个库是否连接到zk之后会向zk写入一些数据？怀疑zk里和之前consumer group相关的数据被污染了。

Rebalance when the partition consumer was error, and set pc.offset with msg.offset

allow consuming different topics in the same group

allow consuming different topics in the same group, as those:

instance-1: consume the topic-A, topic-B
instance-2: consume the topic-A, topic-C

FIX: close sarama consumer after the consumer group was stopped