Skip to content
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

feat: Implement SingleSubscriptionConsumerImpl #281

Merged
merged 5 commits into from
Oct 8, 2020

Conversation

dpcollins-google
Copy link
Collaborator

This is an API that can be used to implement the kafka Consumer API.

@dpcollins-google dpcollins-google requested a review from a team as a code owner October 5, 2020 15:41
@google-cla google-cla bot added the cla: yes This human has signed the Contributor License Agreement. label Oct 5, 2020
@codecov
Copy link

codecov bot commented Oct 5, 2020

Codecov Report

❗ No coverage uploaded for pull request base (master@30c6f83). Click here to learn what that means.
The diff coverage is 90.97%.

Impacted file tree graph

@@            Coverage Diff            @@
##             master     #281   +/-   ##
=========================================
  Coverage          ?   61.68%           
  Complexity        ?      984           
=========================================
  Files             ?      189           
  Lines             ?     6063           
  Branches          ?      525           
=========================================
  Hits              ?     3740           
  Misses            ?     2017           
  Partials          ?      306           
Impacted Files Coverage Δ Complexity Δ
...bsublite/kafka/SingleSubscriptionConsumerImpl.java 90.97% <90.97%> (ø) 36.00 <36.00> (?)

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 30c6f83...6b0b78d. Read the comment docs.

@product-auto-label product-auto-label bot added the api: pubsublite Issues related to the googleapis/java-pubsublite API. label Oct 6, 2020
public ConsumerRecords<byte[], byte[]> poll(Duration duration) {
Map<Partition, Queue<SequencedMessage>> partitionQueues = new HashMap<>();
try {
if (wakeupTriggered) throw new WakeupException();
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you read this without the mutex?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No. Fixed.

try {
if (wakeupTriggered) throw new WakeupException();
while (!duration.isZero()) {
Duration sleepFor = Collections.min(ImmutableList.of(duration, Duration.ofMillis(100)));
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm worried that this will result in very high e2e latency for consumers that set high poll durations. IIUC, Kafka returns from Poll as soon as any data is available, and does not wait until "max.poll.records" records are available. Ideally, poll() would return as soon as any one partition has any data unless "fetch.min.bytes" is set to >1.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, I was misreading the behavior. I've restructured this to return as soon as any messages are available, PTAL.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This works well for cases where messages are already in the buffer when the client calls poll(), but it will still have high latency when messages are delivered during the 100ms sleep. I can think of some cases where this will matter: imagine messages being published uniformly every 10ms. Also, we need to remember that clients are used to single-digit milliseconds latency for Kafka. Would it be possible to have some kind of notification when any of the SinglePartitionSubscribers has messages ready to be delivered?

}));
}
}
try (CloseableMonitor.Hold h = monitor.enter()) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need this last read?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes. Added an explanatory comment as to why.

Copy link
Contributor

@manuelmenzella-google manuelmenzella-google left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I have a single comment that I would like to discuss regarding latency for Kafka subscribers. Everything else looks great, thanks Dan.

try {
if (wakeupTriggered) throw new WakeupException();
while (!duration.isZero()) {
Duration sleepFor = Collections.min(ImmutableList.of(duration, Duration.ofMillis(100)));
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This works well for cases where messages are already in the buffer when the client calls poll(), but it will still have high latency when messages are delivered during the 100ms sleep. I can think of some cases where this will matter: imagine messages being published uniformly every 10ms. Also, we need to remember that clients are used to single-digit milliseconds latency for Kafka. Would it be possible to have some kind of notification when any of the SinglePartitionSubscribers has messages ready to be delivered?

@dpcollins-google
Copy link
Collaborator Author

I have a single comment that I would like to discuss regarding latency for Kafka subscribers. Everything else looks great, thanks Dan.

Ack. As discussed offline, this requires a bit of a gross "register watch" type api to implement. I will add a FR for this behavior and lower the polling duration to 10 ms in a subsequent PR.

@dpcollins-google dpcollins-google merged commit 0e409c2 into master Oct 8, 2020
@dpcollins-google dpcollins-google deleted the single-consumer-impl branch October 8, 2020 18:27
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
api: pubsublite Issues related to the googleapis/java-pubsublite API. cla: yes This human has signed the Contributor License Agreement.
Projects
None yet
Development

Successfully merging this pull request may close these issues.

None yet

2 participants