Honghai, You are going to do a vote on this, right? I think the patch is ready to go so we are just waiting on the KIP adoption I think.
-Jay On Thu, May 21, 2015 at 8:50 AM, Jun Rao <j...@confluent.io> wrote: > Honghai, > > Could you update the wiki on the preallocated size? Instead of > config.segmentSize > - 2 * config.maxMessageSize, we just want to use config.segmentSize. > > Thanks, > > Jun > > On Mon, May 11, 2015 at 8:16 PM, Honghai Chen <honghai.c...@microsoft.com> > wrote: > > > All issues fixed, test cases added, performance result on windows > > attached. The patch can help improve the consume performance around > > 25%~50%. > > > > Thanks, Honghai Chen > > > > -----Original Message----- > > From: Jun Rao [mailto:j...@confluent.io] > > Sent: Wednesday, May 6, 2015 5:39 AM > > To: dev@kafka.apache.org > > Subject: Re: [DISCUSS] KIP 20 Enable log preallocate to improve consume > > performance under windows and some old Linux file system > > > > Thanks. Could you updated the wiki? Also, commented on the jira. > > > > Jun > > > > On Tue, May 5, 2015 at 12:48 AM, Honghai Chen < > honghai.c...@microsoft.com> > > wrote: > > > > > Use config.segmentSize should be ok. Previously add that one for make > > > sure the file not exceed config.segmentSize, actually the function > > > maybeRoll already make sure that. > > > When try add test case for recover, blocked by the rename related > > > issue, just open one jira at > > > https://issues.apache.org/jira/browse/KAFKA-2170 , any recommendation > > for fix that issue? > > > > > > Thanks, Honghai Chen > > > > > > -----Original Message----- > > > From: Jun Rao [mailto:j...@confluent.io] > > > Sent: Tuesday, May 5, 2015 12:51 PM > > > To: dev@kafka.apache.org > > > Subject: Re: [DISCUSS] KIP 20 Enable log preallocate to improve > > > consume performance under windows and some old Linux file system > > > > > > This seems similar to what's in > > > https://issues.apache.org/jira/browse/KAFKA-1065. > > > > > > Also, could you explain why the preallocated size is set to > > > config.segmentSize > > > - 2 * config.maxMessageSize, instead of just config.segmentSize? > > > > > > Thanks, > > > > > > Jun > > > > > > On Mon, May 4, 2015 at 8:12 PM, Honghai Chen > > > <honghai.c...@microsoft.com> > > > wrote: > > > > > > > Hi guys, > > > > I'm trying add test cases, but below case crashed at line " > > > > segReopen.recover(64*1024)--> index.trimToValidSize() ", any idea > > > > for > > > it? > > > > Appreciate your help. > > > > The case assume kafka suddenly crash, and need recover the > > > > last segment. > > > > > > > > kafka.log.LogSegmentTest > testCreateWithInitFileSizeCrash FAILED > > > > java.io.IOException: The requested operation cannot be performed > > > > on a file w ith a user-mapped section open > > > > at java.io.RandomAccessFile.setLength(Native Method) > > > > at > > > > kafka.log.OffsetIndex$$anonfun$resize$1.apply(OffsetIndex.scala:292) > > > > at > > > > kafka.log.OffsetIndex$$anonfun$resize$1.apply(OffsetIndex.scala:283) > > > > at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:288) > > > > at kafka.log.OffsetIndex.resize(OffsetIndex.scala:283) > > > > at > > > > kafka.log.OffsetIndex$$anonfun$trimToValidSize$1.apply$mcV$sp(Offset > > > > I > > > > ndex.scala:272) > > > > at > > > > kafka.log.OffsetIndex$$anonfun$trimToValidSize$1.apply(OffsetIndex.s > > > > c > > > > ala:272) > > > > at > > > > kafka.log.OffsetIndex$$anonfun$trimToValidSize$1.apply(OffsetIndex.s > > > > c > > > > ala:272) > > > > at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:288) > > > > at > kafka.log.OffsetIndex.trimToValidSize(OffsetIndex.scala:271) > > > > at kafka.log.LogSegment.recover(LogSegment.scala:199) > > > > at > > > > kafka.log.LogSegmentTest.testCreateWithInitFileSizeCrash(LogSegmentT > > > > e > > > > st.scala:306) > > > > > > > > def recover(maxMessageSize: Int): Int = { > > > > index.truncate() > > > > index.resize(index.maxIndexSize) > > > > var validBytes = 0 > > > > var lastIndexEntry = 0 > > > > val iter = log.iterator(maxMessageSize) > > > > try { > > > > while(iter.hasNext) { > > > > val entry = iter.next > > > > entry.message.ensureValid() > > > > if(validBytes - lastIndexEntry > indexIntervalBytes) { > > > > // we need to decompress the message, if required, to get > > > > the offset of the first uncompressed message > > > > val startOffset = > > > > entry.message.compressionCodec match { > > > > case NoCompressionCodec => > > > > entry.offset > > > > case _ => > > > > > > > > ByteBufferMessageSet.deepIterator(entry.message).next().offset > > > > } > > > > index.append(startOffset, validBytes) > > > > lastIndexEntry = validBytes > > > > } > > > > validBytes += MessageSet.entrySize(entry.message) > > > > } > > > > } catch { > > > > case e: InvalidMessageException => > > > > logger.warn("Found invalid messages in log segment %s at > > > > byte offset %d: %s.".format(log.file.getAbsolutePath, validBytes, > > > e.getMessage)) > > > > } > > > > val truncated = log.sizeInBytes - validBytes > > > > log.truncateTo(validBytes) > > > > index.trimToValidSize() > > > > truncated > > > > } > > > > > > > > /* create a segment with pre allocate and Crash*/ > > > > @Test > > > > def testCreateWithInitFileSizeCrash() { > > > > val tempDir = TestUtils.tempDir() > > > > val seg = new LogSegment(tempDir, 40, 1, 1000, 0, SystemTime, > > > > false, 512*1024*1024, true) > > > > > > > > val ms = messages(50, "hello", "there") > > > > seg.append(50, ms) > > > > val ms2 = messages(60, "alpha", "beta") > > > > seg.append(60, ms2) > > > > val read = seg.read(startOffset = 55, maxSize = 200, maxOffset = > > > None) > > > > assertEquals(ms2.toList, read.messageSet.toList) > > > > val oldSize = seg.log.sizeInBytes() > > > > val oldPosition = seg.log.channel.position > > > > val oldFileSize = seg.log.file.length > > > > assertEquals(512*1024*1024, oldFileSize) > > > > seg.flush() > > > > seg.log.channel.close() > > > > seg.index.close() > > > > > > > > val segReopen = new LogSegment(tempDir, 40, 1, 1000, 0, > > > > SystemTime, > > > > true) > > > > segReopen.recover(64*1024) > > > > val size = segReopen.log.sizeInBytes() > > > > val position = segReopen.log.channel.position > > > > val fileSize = segReopen.log.file.length > > > > assertEquals(oldPosition, position) > > > > assertEquals(oldSize, size) > > > > assertEquals(size, fileSize) > > > > } > > > > > > > > > > > > > > > > Thanks, Honghai Chen > > > > > > > > -----Original Message----- > > > > From: Sriram Subramanian [mailto:srsubraman...@linkedin.com.INVALID] > > > > Sent: Friday, April 24, 2015 12:57 AM > > > > To: dev@kafka.apache.org > > > > Cc: Roshan Naik > > > > Subject: Re: [DISCUSS] KIP 20 Enable log preallocate to improve > > > > consume performance under windows and some old Linux file system > > > > > > > > +1 > > > > > > > > Some information on how this will be tested would be useful. > > > > > > > > On 4/23/15 9:33 AM, "Jay Kreps" <jay.kr...@gmail.com> wrote: > > > > > > > > >Yeah if we understand the optimal policy for a config we always > > > > >want to set it automatically. In this case I don't think we do yet, > > > > >but down the road that could be nice. I think for now we should > > > > >consider this option experimental to give people a chance to try it > > out. > > > > > > > > > >-Jay > > > > > > > > > >On Wed, Apr 22, 2015 at 7:32 PM, Honghai Chen > > > > ><honghai.c...@microsoft.com> > > > > >wrote: > > > > > > > > > >> Hi Roshan, > > > > >> Use the 'auto' value maybe will break the rule and mess > > > > >> up the configuration. @Jay, any thoughts? > > > > >> > > > > >> Thanks, Honghai Chen > > > > >> > > > > >> -----Original Message----- > > > > >> From: Sriharsha Chintalapani [mailto:harsh...@fastmail.fm] > > > > >> Sent: Thursday, April 23, 2015 6:27 AM > > > > >> To: dev@kafka.apache.org; Roshan Naik > > > > >> Subject: Re: [DISCUSS] KIP 20 Enable log preallocate to improve > > > > >> consume performance under windows and some old Linux file system > > > > >> > > > > >> +1 (non-binding). > > > > >> > > > > >> -- > > > > >> Harsha > > > > >> > > > > >> > > > > >> On April 22, 2015 at 2:52:12 PM, Roshan Naik > > > > >> (ros...@hortonworks.com) > > > > >> wrote: > > > > >> > > > > >> I see that it is safe to keep it this off by default due to some > > > > >>concerns. > > > > >> Eventually, for settings such as this whose 'preferred' value is > > > > >>platform specific (or based on other criteria), it might be worth > > > > >>considering having a default value that is not a constant but an > > > > >>'auto' value .. > > > > >>When > > > > >> kafka boots up it can automatically use the preferred value. > > > > >>Ofcourse it would have to documented as to what auto means for a > > > given platform. > > > > >> > > > > >> -roshan > > > > >> > > > > >> > > > > >> On 4/22/15 1:21 PM, "Jakob Homan" <jgho...@gmail.com> wrote: > > > > >> > > > > >> >+1. This is an important performance fix for Windows-based > > clusters. > > > > >> > > > > > >> >-Jakob > > > > >> > > > > > >> >On 22 April 2015 at 03:25, Honghai Chen > > > > >> ><honghai.c...@microsoft.com> > > > > >> >wrote: > > > > >> >> Fix the issue Sriram mentioned. Code review and jira/KIP > updated. > > > > >> >> > > > > >> >> Below are detail description for the scenarios: > > > > >> >> 1.If do clear shutdown, the last log file will be truncated to > > > > >> >>its real size since the close() function of FileMessageSet will > > > > >> >>call > > > > >>trim(), > > > > >> >> 2.If crash, then when restart, will go through the process of > > > > >> >>recover() and the last log file will be truncate to its real > > > > >> >>size, > > > > >>(and > > > > >> >>the position will be moved to end of the file) 3.When service > > > > >> >>start and open existing file a.Will run the LogSegment > > > > >> >>constructor which has NO parameter "preallocate", b.Then in > > > > >> >>FileMessageSet, the "end" in FileMessageSet will be > > > > >> >>Int.MaxValue, and then > > "channel.position(math.min(channel.size().toInt, end))" > > > > >> >>will make the position be end of the file, c.If recover > > > > >> >>needed, the recover function will truncate file to end > > > > >>of > > > > >> >>valid data, and also move the position to it, > > > > >> >> > > > > >> >> 4.When service running and need create new log segment and new > > > > >> >>FileMessageSet > > > > >> >> > > > > >> >> a.If preallocate = truei.the "end" in FileMessageSet will be > > > > >> >>0, the file size will be "initFileSize", and then > > > > >> >>"channel.position(math.min(channel.size().toInt, end))" will > > > > >> >>make the position be 0, > > > > >> >> > > > > >> >> b.Else if preallocate = falsei.backward compatible, the "end" > > > > >> >>in FileMessageSet will be Int.MaxValue, the file size will be > > > > >> >>"0", and then "channel.position(math.min(channel.size().toInt, > > end))" > > > > >> >>will make the position be 0, > > > > >> >> > > > > >> >> > > > > >> >> > > > > >> > > > > >>https://cwiki.apache.org/confluence/display/KAFKA/KIP-20+-+Enable+ > > > > >>lo > > > > >>g+ > > > > >>pre > > > > >> > > > > > > >>>>allocate+to+improve+consume+performance+under+windows+and+some+old+L > > > > > > >>>>allocate+to+improve+consume+performance+under+windows+and+some+old+i > > > > > > >>>>allocate+to+improve+consume+performance+under+windows+and+some+old+n > > > > > > >>>>allocate+to+improve+consume+performance+under+windows+and+some+old+u > > > > >>>>x+ > > > > >> >>file+system > > > > >> >> https://issues.apache.org/jira/browse/KAFKA-1646 > > > > >> >> https://reviews.apache.org/r/33204/diff/2/ > > > > >> >> > > > > >> >> Thanks, Honghai Chen > > > > >> >> http://aka.ms/kafka > > > > >> >> http://aka.ms/manifold > > > > >> >> > > > > >> >> -----Original Message----- > > > > >> >> From: Honghai Chen > > > > >> >> Sent: Wednesday, April 22, 2015 11:12 AM > > > > >> >> To: dev@kafka.apache.org > > > > >> >> Subject: RE: [DISCUSS] KIP 20 Enable log preallocate to > > > > >> >> improve > > > > >>consume > > > > >> >>performance under windows and some old Linux file system > > > > >> >> > > > > >> >> Hi Sriram, > > > > >> >> One sentence of code missed, will update code review board and > > > > >> >>KIP soon. > > > > >> >> For LogSegment and FileMessageSet, must use different > > > > >> >>constructor function for existing file and new file, then the > > code " > > > > >> >>channel.position(math.min(channel.size().toInt, end)) " will > > > > >> >>make sure the position at end of existing file. > > > > >> >> > > > > >> >> Thanks, Honghai Chen > > > > >> >> > > > > >> >> -----Original Message----- > > > > >> >> From: Jay Kreps [mailto:jay.kr...@gmail.com] > > > > >> >> Sent: Wednesday, April 22, 2015 5:22 AM > > > > >> >> To: dev@kafka.apache.org > > > > >> >> Subject: Re: [DISCUSS] KIP 20 Enable log preallocate to > > > > >> >> improve > > > > >>consume > > > > >> >>performance under windows and some old Linux file system > > > > >> >> > > > > >> >> My understanding of the patch is that clean shutdown truncates > > > > >> >> the > > > > >>file > > > > >> >>back to it's true size (and reallocates it on startup). Hard > > > > >> >>crash is handled by the normal recovery which should truncate > > > > >> >>off the empty portion of the file. > > > > >> >> > > > > >> >> On Tue, Apr 21, 2015 at 10:52 AM, Sriram Subramanian < > > > > >> >>srsubraman...@linkedin.com.invalid> wrote: > > > > >> >> > > > > >> >>> Could you describe how recovery works in this mode? Say, we > > > > >> >>> had a > > > > >>250 > > > > >> >>> MB preallocated segment and we wrote till 50MB and crashed. > > > > >> >>> Till > > > > >>what > > > > >> >>> point do we recover? Also, on startup, how is the append end > > > > >> >>> pointer set even on a clean shutdown? How does the > > > > >> >>> FileChannel end position get set to 50 MB instead of 250 MB? > > > > >> >>> The existing code might just > > > > >>work > > > > >> >>> for it but explaining that would be useful. > > > > >> >>> > > > > >> >>> On 4/21/15 9:40 AM, "Neha Narkhede" <n...@confluent.io> > wrote: > > > > >> >>> > > > > >> >>> >+1. I've tried this on Linux and it helps reduce the spikes > > > > >> >>> >+in > > > > >>append > > > > >> >>> >+(and > > > > >> >>> >hence producer) latency for high throughput writes. I am not > > > > >>entirely > > > > >> >>> >sure why but my suspicion is that in the absence of > > > > >> >>> >preallocation, you see spikes writes need to happen faster > > > > >> >>> >than the time it takes Linux to allocate the next block to > > > > >> >>> >the > > > file. > > > > >> >>> > > > > > >> >>> >It will be great to see some performance test results too. > > > > >> >>> > > > > > >> >>> >On Tue, Apr 21, 2015 at 9:23 AM, Jay Kreps > > > > >> >>> ><jay.kr...@gmail.com> > > > > >> >>>wrote: > > > > >> >>> > > > > > >> >>> >> I'm also +1 on this. The change is quite small and may > > > > >> >>> >>actually help perf on Linux as well (we've never tried > this). > > > > >> >>> >> > > > > >> >>> >> I have a lot of concerns on testing the various failure > > > > >>conditions > > > > >> >>> >> but I think since it will be off by default the risk is > > > > >> >>> >> not too > > > > >> >>>high. > > > > >> >>> >> > > > > >> >>> >> -Jay > > > > >> >>> >> > > > > >> >>> >> On Mon, Apr 20, 2015 at 6:58 PM, Honghai Chen > > > > >> >>> >><honghai.c...@microsoft.com> > > > > >> >>> >> wrote: > > > > >> >>> >> > > > > >> >>> >> > I wrote a KIP for this after some discussion on > KAFKA-1646. > > > > >> >>> >> > https://issues.apache.org/jira/browse/KAFKA-1646 > > > > >> >>> >> > > > > > >> >>> >> > > > > > >> >>> >> > > > > >> >>> >> > > > > >> >>> > > > > >>https://cwiki.apache.org/confluence/display/KAFKA/KIP-20+-+Enable+ > > > > >>lo > > > > >>g+ > > > > >> >>> pre > > > > >> >>> > > > > >> > > > > > > > > > > >>>>>>>allocate+to+improve+consume+performance+under+windows+and+some+old+L > > > > >>>>>>>in > > > > >> >>>>>ux+ > > > > >> >>> >>file+system > > > > >> >>> >> > The RB is here: https://reviews.apache.org/r/33204/diff/ > > > > >> >>> >> > > > > > >> >>> >> > Thanks, Honghai > > > > >> >>> >> > > > > > >> >>> >> > > > > > >> >>> >> > > > > >> >>> > > > > > >> >>> > > > > > >> >>> > > > > > >> >>> >-- > > > > >> >>> >Thanks, > > > > >> >>> >Neha > > > > >> >>> > > > > >> >>> > > > > >> > > > > >> > > > > > > > > > > > > > >
