-
Notifications
You must be signed in to change notification settings - Fork 3.5k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Issue #80: Use receiveAsync() to aggregate messages into shared queue for partitioned consumer #106
Issue #80: Use receiveAsync() to aggregate messages into shared queue for partitioned consumer #106
Conversation
CLA is valid! |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LGTM.
} else { | ||
// Schedule next receiveAsync() if the incoming queue is not full. Use a different thread to avoid | ||
// recursion and stack overflow | ||
ForkJoinPool.commonPool().execute(() -> { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
can't we use iothreads here?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
good point, changed that
this.consumers = Lists.newArrayListWithCapacity(numPartitions); | ||
this.pausedConsumers = new ConcurrentLinkedQueue<>(); | ||
this.sharedQueueResumeThreshold = maxReceiverQueueSize / 2; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
wouldn't it be better if the threshold was maxReceiverQueueSeize - partitions
?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
having the threshold too tight would make flip-flopping easy: receiving a single message would make pausing the consumer and then when the application process it, resume the consumer. Using half-the queue size is to reduce that contention, and at the same time avoids the queue to get empty and starve the application
… queue for partitioned consumer
50adcf4
to
ff8318b
Compare
Co-authored-by: yfuruta <yfuruta@yahoo-corp.jp>
Remove the paths restrict for the CI.
Motivation
Context can be found in #80. This is a slightly different approach from #83, using
receiveAsync()
and pausing and resuming reads depending on the shared queue utilization.