unistack-org/micro-broker-segmentio
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 8 Wiki Activity

improve consumer speed #41

Merged
vtolstov merged 3 commits from speedup into master 2021-07-20 15:05:42 +03:00
Conversation 0 Commits 3 Files Changed 1 +1 -12
2 changed files with 1 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit a9541f7c4f - Show all commits

12
segmentio.go
View File

@ -59,7 +59,7 @@ func (p *publication) Message() *broker.Message {
func (p *publication) Ack() error { func (p *publication) Ack() error {
if atomic.LoadInt32(p.readerDone) == 1 { if atomic.LoadInt32(p.readerDone) == 1 {
return fmt.Errorf("kafka reader done") return kafka.ErrGroupClosed
} }
p.ackCh <- map[string]map[int]int64{p.topic: {p.partition: p.offset}} p.ackCh <- map[string]map[int]int64{p.topic: {p.partition: p.offset}}
if cerr := p.ackErr.Load(); cerr != nil { if cerr := p.ackErr.Load(); cerr != nil {
@ -87,11 +87,9 @@ func (s *subscriber) Unsubscribe(ctx context.Context) error {
group := s.group group := s.group
s.Unlock() s.Unlock()
//fmt.Printf("unsub start\n")
if group != nil { if group != nil {
err = group.Close() err = group.Close()
} }
//fmt.Printf("unsub end\n")
return err return err
} }
@ -466,7 +464,6 @@ func (k *kBroker) commitLoop(generation *kafka.Generation, commitInterval time.D
select { select {
case <-checkTicker.C: case <-checkTicker.C:
if atomic.LoadInt32(cntWait) == 0 { if atomic.LoadInt32(cntWait) == 0 {
//fmt.Printf("cntWait IS 0\n")
mapMu.Lock() mapMu.Lock()
if len(offsets) > 0 { if len(offsets) > 0 {
if err := generation.CommitOffsets(offsets); err != nil { if err := generation.CommitOffsets(offsets); err != nil {
@ -511,7 +508,6 @@ func (k *kBroker) commitLoop(generation *kafka.Generation, commitInterval time.D
} }
// check for readers done and commit offsets // check for readers done and commit offsets
if atomic.LoadInt32(cntWait) == 0 { if atomic.LoadInt32(cntWait) == 0 {
//fmt.Printf("cntWait IS 0\n")
mapMu.Lock() mapMu.Lock()
if len(offsets) > 0 { if len(offsets) > 0 {
if err := generation.CommitOffsets(offsets); err != nil { if err := generation.CommitOffsets(offsets); err != nil {
@ -527,7 +523,6 @@ func (k *kBroker) commitLoop(generation *kafka.Generation, commitInterval time.D
} }
return return
} }
//fmt.Printf("cntWait NOT 0\n")
} }
} }
}() }()
@ -544,7 +539,6 @@ func (k *kBroker) commitLoop(generation *kafka.Generation, commitInterval time.D
if atomic.LoadInt32(readerDone) == 1 { if atomic.LoadInt32(readerDone) == 1 {
mapMu.Lock() mapMu.Lock()
if len(offsets) == 0 { if len(offsets) == 0 {
//fmt.Printf("close all on <-readerDoneCh\n")
defer ticker.Stop() defer ticker.Stop()
return return
} }
@ -570,7 +564,6 @@ func (k *kBroker) commitLoop(generation *kafka.Generation, commitInterval time.D
offsets = make(map[string]map[int]int64, 4) offsets = make(map[string]map[int]int64, 4)
mapMu.Unlock() mapMu.Unlock()
if atomic.LoadInt32(readerDone) == 1 && atomic.LoadInt32(cntWait) == 0 { if atomic.LoadInt32(readerDone) == 1 && atomic.LoadInt32(cntWait) == 0 {
//fmt.Printf("close all on <-ticker.C\n")
return return
} }
} }
@ -626,21 +619,18 @@ func (h *cgHandler) run(ctx context.Context) {
if h.brokerOpts.Logger.V(logger.ErrorLevel) { if h.brokerOpts.Logger.V(logger.ErrorLevel) {
h.brokerOpts.Logger.Errorf(h.brokerOpts.Context, "[segmentio] unexpected error type: %T err: %v", err, err) h.brokerOpts.Logger.Errorf(h.brokerOpts.Context, "[segmentio] unexpected error type: %T err: %v", err, err)
} }
//fmt.Printf("exit from readMessage loop\n")
return return
case kafka.ErrGenerationEnded: case kafka.ErrGenerationEnded:
// generation has ended // generation has ended
if h.brokerOpts.Logger.V(logger.TraceLevel) { if h.brokerOpts.Logger.V(logger.TraceLevel) {
h.brokerOpts.Logger.Trace(h.brokerOpts.Context, "[segmentio] generation ended, rebalance or close") h.brokerOpts.Logger.Trace(h.brokerOpts.Context, "[segmentio] generation ended, rebalance or close")
} }
//fmt.Printf("exit from readMessage loop\n")
return return
case nil: case nil:
if cerr := commitErr.Load(); cerr != nil { if cerr := commitErr.Load(); cerr != nil {
if h.brokerOpts.Logger.V(logger.ErrorLevel) { if h.brokerOpts.Logger.V(logger.ErrorLevel) {
h.brokerOpts.Logger.Errorf(h.brokerOpts.Context, "[segmentio] commit error: %v", cerr) h.brokerOpts.Logger.Errorf(h.brokerOpts.Context, "[segmentio] commit error: %v", cerr)
} }
//fmt.Printf("exit from readMessage loop\n")
return return
} }

1
stats.go
View File

@ -15,7 +15,6 @@ func readerStats(ctx context.Context, r *kafka.Reader, td time.Duration, m meter
for { for {
select { select {
case <-ctx.Done(): case <-ctx.Done():
//fmt.Printf("done reader stats\n")
return return
case <-ticker.C: case <-ticker.C:
if r == nil { if r == nil {