Use bounded list for offset manager #1567

vprithvi · 2019-05-24T21:22:10Z

Which problem is this PR solving?

Resolves #1109

Short description of the changes

Bound the concurrent_list used by offset manager
Add a ParallelSpanProcessor that can signal when there are errors when processing spans via a channel
Update Consumer to handle errors from ParallelSpanProcessor by closing the partition. This should trigger a rebalance causing reprocessing from the last committed offset.

Signed-off-by: Prithvi Raj p.r@uber.com

codecov · 2019-05-24T21:34:38Z

Codecov Report

Merging #1567 into master will increase coverage by <.01%.
The diff coverage is 100%.

@@            Coverage Diff             @@
##           master    #1567      +/-   ##
==========================================
+ Coverage   98.21%   98.22%   +<.01%     
==========================================
  Files         195      195              
  Lines        9602     9632      +30     
==========================================
+ Hits         9431     9461      +30     
  Misses        134      134              
  Partials       37       37

Impacted Files	Coverage Δ
cmd/ingester/app/processor/span_processor.go	`100% <ø> (ø)`	⬆️
cmd/ingester/app/consumer/offset/manager.go	`100% <100%> (ø)`	⬆️
...md/ingester/app/consumer/offset/concurrent_list.go	`100% <100%> (ø)`	⬆️
cmd/ingester/app/consumer/processor_factory.go	`100% <100%> (ø)`	⬆️
cmd/ingester/app/consumer/consumer.go	`100% <100%> (ø)`	⬆️
cmd/ingester/app/flags.go	`100% <100%> (ø)`	⬆️
cmd/ingester/app/processor/parallel_processor.go	`100% <100%> (ø)`	⬆️
cmd/ingester/app/consumer/committing_processor.go	`100% <100%> (ø)`	⬆️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 6cb3570...38f1df5. Read the comment docs.

cmd/ingester/app/consumer/consumer_metrics_test.go

guanw · 2019-05-26T00:21:17Z

cmd/ingester/app/flags.go

@@ -63,14 +65,17 @@ const (
 	DefaultEncoding = kafka.EncodingProto
 	// DefaultDeadlockInterval is the default deadlock interval
 	DefaultDeadlockInterval = 1 * time.Minute
+	// DefaultMaxOutOfOrderOffsets is the default max out of ourder offsets to maintain before restarting the processor
+	DefaultMaxOutOfOrderOffsets = 500000


How do we determine this default value?

I used an average message rate of 10k qps, so that this list would be exhausted in 50s.
These numbers are arbitrary - using a number that is too small would cause churn by closing and recreating partition consumers unnecessarily

having some explanation of the default value in the comment would be good.

Perhaps it's something we can tweak at runtime?

cmd/ingester/app/consumer/consumer.go

cmd/ingester/app/consumer/offset/concurrent_list.go

cmd/ingester/app/processor/span_processor.go

resolves jaegertracing#1109 Signed-off-by: Prithvi Raj <p.r@uber.com>

Signed-off-by: Prithvi Raj <p.r@uber.com>

yurishkuro

not clear on the design for this change, and some implementation details.

concurrent_list stores non-consecutive offsets, right? So 'capacity' is kind of misleading as a control knob, since it doesn't correlate with the number of messages outstanding

let's discuss offline

yurishkuro · 2019-09-17T19:34:08Z

cmd/ingester/app/consumer/committing_processor.go

@@ -43,7 +43,7 @@ func (d *comittingProcessor) Process(message processor.Message) error {
 	if msg, ok := message.(Message); ok {
 		err := d.processor.Process(message)
 		if err == nil {
-			d.marker.MarkOffset(msg.Offset())
+			return d.marker.MarkOffset(msg.Offset())


if an error is returned here, will the message be re-processed?

No, it should not be. If an error is thrown here, the number of retries have already been exhausted.

yurishkuro · 2019-09-17T19:35:01Z

cmd/ingester/app/consumer/committing_processor.go

@@ -43,7 +43,7 @@ func (d *comittingProcessor) Process(message processor.Message) error {
 	if msg, ok := message.(Message); ok {
 		err := d.processor.Process(message)
 		if err == nil {


nit: switch to more idiomatic

if err != nil { return err } return d.marker.MarkOffset(msg.Offset())

cmd/ingester/app/consumer/consumer.go

yurishkuro · 2019-09-17T19:38:49Z

cmd/ingester/app/consumer/consumer.go

@@ -149,11 +150,17 @@ func (c *Consumer) handleMessages(pc sc.PartitionConsumer) {
 				defer msgProcessor.Close()
 			}

-			msgProcessor.Process(&saramaMessageWrapper{msg})
+			msgProcessor.Process(&saramaMessageWrapper{msg}, func(message processor.Message, e error) {
+				msgErrors <- e


is the callback func() only called on error? If it can be called on success, then we should check for err != nil.

nit: s/e/err/

The callback is only called onError

cmd/ingester/app/consumer/offset/concurrent_list.go

cmd/ingester/app/processor/parallel_processor.go

Signed-off-by: Prithvi Raj <p.r@uber.com>

yurishkuro · 2019-11-11T22:29:19Z

we're considering a different solution

vprithvi requested review from black-adder, jpkrohling, objectiser, pavolloffay, tiffon and yurishkuro as code owners May 24, 2019 21:22

guanw reviewed May 26, 2019

View reviewed changes

vprithvi force-pushed the use-bounded-list branch from 7bc2376 to 37c037a Compare May 28, 2019 14:55

yurishkuro reviewed Jun 2, 2019

View reviewed changes

cmd/ingester/app/consumer/consumer.go Outdated Show resolved Hide resolved

yurishkuro reviewed Jun 2, 2019

View reviewed changes

cmd/ingester/app/consumer/offset/concurrent_list.go Outdated Show resolved Hide resolved

yurishkuro reviewed Jun 2, 2019

View reviewed changes

cmd/ingester/app/processor/span_processor.go Outdated Show resolved Hide resolved

vprithvi force-pushed the use-bounded-list branch from 37c037a to 9df0742 Compare September 12, 2019 22:00

vprithvi changed the title ~~Use bounded list for offset manager~~ [WIP] Use bounded list for offset manager Sep 12, 2019

vprithvi force-pushed the use-bounded-list branch 2 times, most recently from f7a7ca4 to 57b37ae Compare September 16, 2019 16:55

Use bounded list for offset manager

3250b91

resolves jaegertracing#1109 Signed-off-by: Prithvi Raj <p.r@uber.com>

vprithvi force-pushed the use-bounded-list branch from 57b37ae to 3250b91 Compare September 16, 2019 17:12

vprithvi changed the title ~~[WIP] Use bounded list for offset manager~~ Use bounded list for offset manager Sep 16, 2019

Rename size to capacity

0a5e428

Signed-off-by: Prithvi Raj <p.r@uber.com>

yurishkuro reviewed Sep 17, 2019

View reviewed changes

Address feedback

38f1df5

Signed-off-by: Prithvi Raj <p.r@uber.com>

vprithvi closed this Nov 11, 2019

vprithvi deleted the use-bounded-list branch December 2, 2022 15:25

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Use bounded list for offset manager #1567

Use bounded list for offset manager #1567

vprithvi commented May 24, 2019

codecov bot commented May 24, 2019 •

edited

guanw May 26, 2019

vprithvi May 28, 2019

yurishkuro Jun 2, 2019

jpkrohling Jun 3, 2019

yurishkuro left a comment

yurishkuro Sep 17, 2019

vprithvi Sep 19, 2019

yurishkuro Sep 17, 2019

yurishkuro Sep 17, 2019

vprithvi Sep 24, 2019

yurishkuro commented Nov 11, 2019

Use bounded list for offset manager #1567

Use bounded list for offset manager #1567

Conversation

vprithvi commented May 24, 2019

Which problem is this PR solving?

Short description of the changes

codecov bot commented May 24, 2019 • edited

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yurishkuro left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yurishkuro commented Nov 11, 2019

codecov bot commented May 24, 2019 •

edited