KAFKA-3197 Fix producer sending records out of order#857
Conversation
There was a problem hiding this comment.
Was this just changed for cleanup reasons?
There was a problem hiding this comment.
Yes. This method seems exposed only for unit test purpose. So protected seems good enough. Is there any other case we need access to this method?
There was a problem hiding this comment.
if it's only for unit tests, is package private good enough?
|
@granthenke Thanks for the review, updated the patch. I don't feel strong on the method variances as they look pretty much like private calls, just like the package name implies. |
There was a problem hiding this comment.
one liner? partitionsInFlight = sendInOrder ? new HashSet : null;
There was a problem hiding this comment.
why twice especially if the timestamp isn't changing?
There was a problem hiding this comment.
You should close the metrics object as is done above.
becketqin
force-pushed
the
KAFKA-3197
branch
from
f84441e to
bf3a014
Compare
| now); | ||
| if (sendInOrder) { | ||
| // Mute all the partitions drained | ||
| for (Map.Entry<Integer, List<RecordBatch>> entry : batches.entrySet()) { |
There was a problem hiding this comment.
Minor: doesn't look like you're using the key, maybe you could use batches.values()?
| private final Time time; | ||
| private final ConcurrentMap<TopicPartition, Deque<RecordBatch>> batches; | ||
| private final IncompleteRecordBatches incomplete; | ||
| private final Set<TopicPartition> muted; |
There was a problem hiding this comment.
It would be good to add a comment why this doesn't have to be thread-safe (unlike incomplete and batches, for example). In fact, it would be good to group the fields that are not thread-safe together (drainIndex seems to be the other one).
7 participants
This patch reuse max.in.flight.request.per.connection. When it equals to one, we take it as user wants order protection. The current approach is make sure there is only one batch per partition on the fly.