Enhance Pulsar consumer #12812
Enhance Pulsar consumer #12812
Conversation
Jackie-Jiang
added
dependencies
Jackie-Jiang
force-pushed
the
enhance_pulsar_consumer
branch
from
88ba624
to
1b57668
Compare
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #12812 +/- ##
============================================
+ Coverage 61.75% 61.99% +0.24%
+ Complexity 207 198 -9
============================================
Files 2436 2462 +26
Lines 133233 134696 +1463
Branches 20636 20812 +176
============================================
+ Hits 82274 83507 +1233
- Misses 44911 45052 +141
- Partials 6048 6137 +89
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. |
Jackie-Jiang
force-pushed
the
enhance_pulsar_consumer
branch
from
1b57668
to
cdf1193
Compare
|
|
||
| // Seek to the start message id if necessary | ||
| // NOTE: Use Objects.equals() to check reference first for performance. | ||
| if (!Objects.equals(startMessageId, _nextMessageId)) { |
There was a problem hiding this comment.
To clarify, when startMessageId and nextMessageId are the same, it means we are done with the current batch and starting from next batch?
There was a problem hiding this comment.
When they are the same, we don't need to seek the message id again
There was a problem hiding this comment.
When they are the same, we don't need to seek the message id again
Ok. Its the caching logic to avoid re-seeking.
| break; | ||
| } | ||
| while (_reader.hasMessageAvailable() && System.currentTimeMillis() < endTimeMs) { | ||
| messages.add(PulsarUtils.buildPulsarStreamMessage(_reader.readNext(), _config)); |
There was a problem hiding this comment.
Assuming that readNext will block.
There was a problem hiding this comment.
readNext() shouldn't block. We call readNext() only when there hasMessageAvailable()
There was a problem hiding this comment.
readNext()shouldn't block. We callreadNext()only when therehasMessageAvailable()
Makes sense if hasMessageAvailable is used. I read this and thought it blocks
https://github.com/apache/pulsar/blob/master/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Reader.java#L49
It doesn't rely on the async task to be interrupted. But essentially both of them will gather as many messages as possible within the timeout. |
java.util.ConcurrentModificationException#10503)