bug: some data race found in rocketmq-client-go #790
Description
Running go test with race in ci, and find some data race in rocketmq-client-go
==================
WARNING: DATA RACE
Write at 0x00c0002035f8 by goroutine 90:
github.com/emirpasic/gods/trees/redblacktree.(*Tree).Remove()
/home/runner/go/pkg/mod/github.com/emirpasic/gods@v1.12.0/trees/redblacktree/redblacktree.go:145 +0x256
github.com/emirpasic/gods/maps/treemap.(*Map).Remove()
/home/runner/go/pkg/mod/github.com/emirpasic/gods@v1.12.0/maps/treemap/treemap.go:62 +0x294
github.com/apache/rocketmq-client-go/v2/consumer.(*processQueue).removeMessage()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/process_queue.go:198 +0x205
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).consumeMessageCurrently.func1()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:1040 +0xaf7
github.com/apache/rocketmq-client-go/v2/primitive.WithRecover()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/primitive/base.go:99 +0x5a
Previous read at 0x00c0002035f8 by goroutine 103:
github.com/emirpasic/gods/trees/redblacktree.(*Tree).Size()
/home/runner/go/pkg/mod/github.com/emirpasic/gods@v1.12.0/trees/redblacktree/redblacktree.go:155 +0x424
github.com/emirpasic/gods/maps/treemap.(*Map).Size()
/home/runner/go/pkg/mod/github.com/emirpasic/gods@v1.12.0/maps/treemap/treemap.go:72 +0x3f6
github.com/apache/rocketmq-client-go/v2/consumer.(*processQueue).putMessage()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/process_queue.go:121 +0x3cb
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).pullMessage()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:757 +0x3971
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).Start.func1.1.1()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:165 +0x44
Goroutine 90 (running) created at:
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).consumeMessageCurrently()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:974 +0xe4
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).consumeMessageCurrently-fm()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:959 +0x5e
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).pullMessage.func1()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:533 +0x118
github.com/apache/rocketmq-client-go/v2/primitive.WithRecover()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/primitive/base.go:99 +0x5a
Goroutine 103 (running) created at:
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).Start.func1.1()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:164 +0x70
==================
==================
WARNING: DATA RACE
Read at 0x00c00028a6c8 by goroutine 103:
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).pullMessage()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:5[84](https://github.com/douyu/jupiter/runs/5579931547?check_suite_focus=true#step:10:84) +0x22d9
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).Start.func1.1.1()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:165 +0x44
Previous write at 0x00c00028a6c8 by goroutine 90:
sync/atomic.AddInt64()
/opt/hostedtoolcache/go/1.16.14/x64/src/runtime/race_amd64.s:300 +0xb
github.com/apache/rocketmq-client-go/v2/consumer.(*processQueue).removeMessage()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/process_queue.go:200 +0x2cc
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).consumeMessageCurrently.func1()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:1040 +0xaf7
github.com/apache/rocketmq-client-go/v2/primitive.WithRecover()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/primitive/base.go:99 +0x5a
Goroutine 103 (running) created at:
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).Start.func1.1()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:164 +0x70
Goroutine 90 (finished) created at:
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).consumeMessageCurrently()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:974 +0xe4
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).consumeMessageCurrently-fm()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:959 +0x5e
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).pullMessage.func1()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:533 +0x118
github.com/apache/rocketmq-client-go/v2/primitive.WithRecover()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/primitive/base.go:99 +0x5a
==================
==================
WARNING: DATA RACE
Read at 0x00c00028a6c0 by goroutine 103:
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).pullMessage()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:5[85](https://github.com/douyu/jupiter/runs/5579931547?check_suite_focus=true#step:10:85) +0x22f6
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).Start.func1.1.1()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:165 +0x44
Previous write at 0x00c00028a6c0 by goroutine [90](https://github.com/douyu/jupiter/runs/5579931547?check_suite_focus=true#step:10:90):
sync/atomic.AddInt64()
/opt/hostedtoolcache/go/1.16.14/x64/src/runtime/race_amd64.s:300 +0xb
github.com/apache/rocketmq-client-go/v2/consumer.(*processQueue).removeMessage()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/process_queue.go:202 +0x2ea
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).consumeMessageCurrently.func1()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:1040 +0xaf7
github.com/apache/rocketmq-client-go/v2/primitive.WithRecover()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/primitive/base.go:99 +0x5a
Goroutine 103 (running) created at:
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).Start.func1.1()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:164 +0x70
Goroutine 90 (finished) created at:
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).consumeMessageCurrently()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:974 +0xe4
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).consumeMessageCurrently-fm()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:[95](https://github.com/douyu/jupiter/runs/5579931547?check_suite_focus=true#step:10:95)9 +0x5e
github.com/apache/rocketmq-client-go/v2/consumer.(*pushConsumer).pullMessage.func1()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/consumer/push_consumer.go:533 +0x118
github.com/apache/rocketmq-client-go/v2/primitive.WithRecover()
/home/runner/go/pkg/mod/github.com/apache/rocketmq-client-go/v2@v2.1.1-rc2/primitive/base.go:[99](https://github.com/douyu/jupiter/runs/5579931547?check_suite_focus=true#step:10:99) +0x5a
==================