> On Oct. 25, 2014, 7:52 a.m., Joel Koshy wrote: > >
Joel, thanks a lot for the review! Some comments on your comments. > On Oct. 25, 2014, 7:52 a.m., Joel Koshy wrote: > > core/src/main/scala/kafka/utils/ByteBoundedBlockingQueue.scala, line 18 > > <https://reviews.apache.org/r/26755/diff/3/?file=722245#file722245line18> > > > > I'm wondering if this is specific and nuanced enough to make it > > entirely private to MirrorMaker.scala > > > > OR > > > > if you think it is useful as a generic utility consider putting in > > org.apache.kafka.clients.common.utils My first thinking is that this could help provide better control on memory management in broader cases in addition to mirror maker, such as consumer side data chunk queue, and maybe also for controller message queue. > On Oct. 25, 2014, 7:52 a.m., Joel Koshy wrote: > > core/src/main/scala/kafka/utils/ByteBoundedBlockingQueue.scala, line 36 > > <https://reviews.apache.org/r/26755/diff/3/?file=722245#file722245line36> > > > > This can be paraphrased to be simpler: > > > > "An element can be enqueued provided the current size (in number of > > elements) is within the configured capacity and the current size in bytes > > of the queue is within the configured byte capacity. i.e., the element may > > be enqueued even if adding it causes the queue's size in bytes to exceed > > the byte capacity." > > > > Ok, so while I was thinking through the above: is there any benefit to > > having a count-based capacity when you have a byte-based capacity? i.e., > > why not have byte-capacity only? I think there might be three cases where a queue size in number of message could help: 1. Because the size function is actually provided by user, if the message has a small payload with big overhead on other stuff, the size of bytes might not be working well. 2. The queue num size could be used to control the number of message buffered in the middle, i.e the failure boundry. For instance if one of the mirror maker bounced, if we buffered too many messages in mirror maker, we need to reconsume all of them again. 3. In cases where the byte limit only used to protect against of running out of memory, but users don't expect the queue to consume that much memory all the time. (I'm not sure if it is a valid use case though...) > On Oct. 25, 2014, 7:52 a.m., Joel Koshy wrote: > > core/src/main/scala/kafka/utils/ByteBoundedBlockingQueue.scala, line 82 > > <https://reviews.apache.org/r/26755/diff/3/?file=722245#file722245line82> > > > > One significant caveat to this approach (and in the timed variant > > above) is that if a single large element needs to be enqueued it could > > potentially block a number of smaller elements from being enqueued. This > > may be okay in the case of mirror maker though but would make it less > > useful as a generic utility. I'm not sure why the big put could block small ones... It is possible that there is a super big item put into the queue and makes the queue to pass the byte limit by a lot. In that case, all the put will be blocked until a bunch of small messages are taken out of the queue. But it seems to be the purpose of having a byte limit for the queue. > On Oct. 25, 2014, 7:52 a.m., Joel Koshy wrote: > > core/src/main/scala/kafka/utils/ByteBoundedBlockingQueue.scala, line 100 > > <https://reviews.apache.org/r/26755/diff/3/?file=722245#file722245line100> > > > > Can you clarify what this means? I was trying to say that the poll method does not contend lock with offer. I saw similar description from some queue's javadoc, it is probably better to remove it... > On Oct. 25, 2014, 7:52 a.m., Joel Koshy wrote: > > core/src/main/scala/kafka/utils/ByteBoundedBlockingQueue.scala, line 109 > > <https://reviews.apache.org/r/26755/diff/3/?file=722245#file722245line109> > > > > getAndDecrement(sizeFunction.get(e)) It seems getAndDecrement() does not take argument and will always decrement by 1. - Jiangjie ----------------------------------------------------------- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/26755/#review58497 ----------------------------------------------------------- On Oct. 15, 2014, 4:28 p.m., Jiangjie Qin wrote: > > ----------------------------------------------------------- > This is an automatically generated e-mail. To reply, visit: > https://reviews.apache.org/r/26755/ > ----------------------------------------------------------- > > (Updated Oct. 15, 2014, 4:28 p.m.) > > > Review request for kafka. > > > Bugs: KAFKA-1706 > https://issues.apache.org/jira/browse/KAFKA-1706 > > > Repository: kafka > > > Description > ------- > > changed arguments name > > > correct typo. > > > Diffs > ----- > > core/src/main/scala/kafka/utils/ByteBoundedBlockingQueue.scala PRE-CREATION > > Diff: https://reviews.apache.org/r/26755/diff/ > > > Testing > ------- > > > Thanks, > > Jiangjie Qin > >
