Skip to content
This repository has been archived by the owner on Sep 15, 2023. It is now read-only.

[fix(ekafka)] originErr 没有正确赋值 #146

Merged
merged 2 commits into from
Aug 23, 2021
Merged

[fix(ekafka)] originErr 没有正确赋值 #146

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2b026c5
[fix(ekafka)] 调用 Component Stop() 方法后，没有正确关闭 kafka consumer
Aug 20, 2021
dbafbac
[fix(ekafka)] 需要等 handler 退出
Aug 23, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 12 additions & 12 deletions ekafka/consumerserver/component.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,20 +149,20 @@ func (cmp *Component) launchOnConsumerGroupStart() error {

var originErr error
select {
case originErr := <-handlerExit:
case originErr = <-handlerExit:
if originErr != nil {
cmp.logger.Error("terminating ConsumerServer because an error", elog.FieldErr(originErr))
cmp.logger.Error("terminating ConsumerGroup because an error", elog.FieldErr(originErr))
} else {
cmp.logger.Info("message handler exited without any error, terminating ConsumerServer")
cmp.logger.Info("message handler exited without any error, terminating ConsumerGroup")
}
cmp.stopServer()
case <-cmp.ServerCtx.Done():
originErr := cmp.ServerCtx.Err()
cmp.logger.Error("terminating ConsumerServer because a context error", elog.FieldErr(originErr))
originErr = cmp.ServerCtx.Err()
cmp.logger.Error("terminating ConsumerGroup because a context error", elog.FieldErr(originErr))

err := <-handlerExit
if err != nil {
cmp.logger.Error("terminating ConsumerServer because an error", elog.FieldErr(err))
cmp.logger.Error("terminating ConsumerGroup because an error", elog.FieldErr(err))
} else {
cmp.logger.Info("message handler exited without any error")
}
Expand Down Expand Up @@ -195,20 +195,20 @@ func (cmp *Component) launchOnConsumerStart() error {

var originErr error
select {
case originErr := <-handlerExit:
case originErr = <-handlerExit:
if originErr != nil {
cmp.logger.Error("terminating ConsumerGroup because an error", elog.FieldErr(originErr))
cmp.logger.Error("terminating ConsumerServer because an error", elog.FieldErr(originErr))
} else {
cmp.logger.Info("message handler exited without any error, terminating ConsumerGroup")
cmp.logger.Info("message handler exited without any error, terminating ConsumerServer")
}
cmp.stopServer()
case <-cmp.ServerCtx.Done():
originErr := cmp.ServerCtx.Err()
cmp.logger.Error("terminating ConsumerGroup because a context error", elog.FieldErr(originErr))
originErr = cmp.ServerCtx.Err()
cmp.logger.Error("terminating ConsumerServer because a context error", elog.FieldErr(originErr))

err := <-handlerExit
if err != nil {
cmp.logger.Error("terminating ConsumerGroup because an error", elog.FieldErr(err))
cmp.logger.Error("terminating ConsumerServer because an error", elog.FieldErr(err))
} else {
cmp.logger.Info("message handler exited without any error")
}
Expand Down
2 changes: 1 addition & 1 deletion ekafka/test/e2e/consumer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ func Test_ConsumeWithConsumer(t *testing.T) {
)
consumer := cmp.Consumer("c1")
for {
msg, err := consumer.ReadMessage(ctx)
msg, _, err := consumer.ReadMessage(ctx)
if err != nil {
consumerGroupErr <- err
return
Expand Down
2 changes: 1 addition & 1 deletion ekafka/test/e2e/consumerserver_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,7 @@ func Test_ConsumeServer_OnConsumerStart(t *testing.T) {
consumerServerComponent.OnStart(
func(ctx context.Context, consumer *ekafka.Consumer) error {
for {
msg, err := consumer.ReadMessage(ctx)
msg, _, err := consumer.ReadMessage(ctx)
if err != nil {
return err
}
Expand Down