KAFKA-3163: Add time based index to Kafka. #1215

becketqin · 2016-04-11T22:59:35Z

This patch is for KIP-33.
https://cwiki.apache.org/confluence/display/KAFKA/KIP-33+-+Add+a+time+based+log+index

becketqin · 2016-04-11T23:14:18Z

It seems this conflicts with KAFKA-3510. I'll do the rebase.

becketqin · 2016-04-12T00:34:13Z

cc @junrao. Will you be able to help review this patch? Thanks!

becketqin · 2016-04-12T00:37:28Z

core/src/main/scala/kafka/log/Log.scala

        val segment = new LogSegment(dir = dir,
                                     startOffset = start,
                                     indexIntervalBytes = config.indexInterval,
                                     maxIndexSize = config.maxIndexSize,
                                     rollJitterMs = config.randomSegmentJitter,
                                     time = time,
                                     fileAlreadyExists = true)
-
-        if(indexFile.exists()) {
+        if (indexFile.exists() && timeIndexFile.exists()) {


I have a question regarding the code here. On line 199 of this patch (line 192 of the original code), we check if the index file exists or not. However, right before this, we have already created the log segment which will create the index file if it does not exist. So it seems the test here will never be false?

That's a good point. It seems that we should check the existence of the index file before creating LogSegment.

If we check the existence of the index files before creating log segment, would it be a little difficult to distinguish between 1) the upgrade case and 2) time index file is really missing? In (1), we want to just create an empty time index without rebuilding the time index. In (2), we want to rebuild the entire time index.

I am wondering in which case will we miss a index? Is it only when the index is deleted manually?

junrao · 2016-08-18T19:58:31Z

@becketqin : Thanks. Could you confirm the number of log dirs you have and the number of disks per dir? Intuitively, if there are multiple disks per dir, it seems using more than one 1 recovery thread should lead to better performance.

becketqin · 2016-08-18T21:15:27Z

@junrao We only have one log directory on a RAID 10 containing 10 disk. So in our case, multiple thread might not help. In a JBOD environment, I agree that multiple threads would probably help.

junrao · 2016-08-18T22:01:14Z

@becketqin : Interesting, in theory, using more than 1 thread should still help since those threads can drive the I/Os on different disks in parallel.

becketqin · 2016-08-18T23:04:45Z

@junrao I am not sure, one thing I noticed is that when there is one thread, the log loading time of each partition is pretty linear to the number of log segments. But if multiple threads are used, the linearity goes away.

For example, the following logs are from running the code using 10 log recovery threads without time index (Old_Mark indicates no time index, and the same random Old_Mark means they are from the same test run).

For TOPIC1, it only took ~400 ms to load about 2000 log segments, at about the same time (300 ms earlier) TOPIC2 takes 22 ms to load ~70 log segments. However, later on the same run (3 seconds later), for some other partitions of TOPIC2 that has about the same number of segments, it takes much longer (~400 ms) to load. This was the reason that I suspected that the later mmap() calls are more expensive, which seems also proven from profiling by measuring the time on mmap calls in different buckets.

I am not sure why this happens, but I didn't see such issue when there is one log loading thread.

2016/08/16 20:57:28.793 INFO [Log] [pool-9-thread-3] [kafka-server] [] (22 ms) Completed load of log TOPIC2-7 with 74 log segments and log end offset 1835960333 (Old_Mark 5303310516390717976)
2016/08/16 20:57:28.794 INFO [Log] [pool-9-thread-1] [kafka-server] [] (22 ms) Completed load of log TOPIC2-4 with 84 log segments and log end offset 1829077019 (Old_Mark 5303310516390717976)
2016/08/16 20:57:28.795 INFO [Log] [pool-9-thread-2] [kafka-server] [] (23 ms) Completed load of log TOPIC2-8 with 68 log segments and log end offset 1831459833 (Old_Mark 5303310516390717976)
2016/08/16 20:57:28.795 INFO [Log] [pool-9-thread-4] [kafka-server] [] (24 ms) Completed load of log TOPIC2-2 with 79 log segments and log end offset 1836145525 (Old_Mark 5303310516390717976)
......
2016/08/16 20:57:29.133 INFO [Log] [pool-9-thread-1] [kafka-server] [] (337 ms) Completed load of log TOPIC1-2 with 1973 log segments and log end offset 7832957736 (Old_Mark 5303310516390717976)
2016/08/16 20:57:29.177 INFO [Log] [pool-9-thread-2] [kafka-server] [] (380 ms) Completed load of log TOPIC1-9 with 2020 log segments and log end offset 7871065990 (Old_Mark 5303310516390717976)
2016/08/16 20:57:29.203 INFO [Log] [pool-9-thread-5] [kafka-server] [] (407 ms) Completed load of log TOPIC1-7 with 2076 log segments and log end offset 7839205001 (Old_Mark 5303310516390717976)
2016/08/16 20:57:29.245 INFO [Log] [pool-9-thread-3] [kafka-server] [] (447 ms) Completed load of log TOPIC1-0 with 2078 log segments and log end offset 7852775406 (Old_Mark 5303310516390717976)
...
2016/08/16 20:57:32.100 INFO [Log] [pool-9-thread-1] [kafka-server] [] (387 ms) Completed load of log TOPIC2-112 with 77 log segments and log end offset 371172232 (Old_Mark 5303310516390717976)
2016/08/16 20:57:32.100 INFO [Log] [pool-9-thread-2] [kafka-server] [] (383 ms) Completed load of log TOPIC2-122 with 68 log segments and log end offset 373822294 (Old_Mark 5303310516390717976)
2016/08/16 20:57:32.110 INFO [Log] [pool-9-thread-4] [kafka-server] [] (393 ms) Completed load of log TOPIC2-27 with 81 log segments and log end offset 375406677 (Old_Mark 5303310516390717976)
2016/08/16 20:57:32.124 INFO [Log] [pool-9-thread-3] [kafka-server] [] (406 ms) Completed load of log TOPIC2-17 with 72 log segments and log end offset 371089327 (Old_Mark 5303310516390717976)

junrao · 2016-08-18T23:50:45Z

docs/upgrade.html

+<h5><a id="upgrade_10_1_breaking" href="#upgrade_10_1_breaking">Potential breaking changes in 0.10.1.0</a></h5>
+<ul>
+    <li> The log retention time is no longer based on last modified time of the log segments. Instead it will be based on the largest timestamp of the messages in a log segment.</li>
+    <li> The log rolling time is no longer depending on log segment create time. Instead it is now based on the timestamp of the first message in a log segment. i.e. if the timestmp of the first message in the segment is T, the log will be rolled out at T + log.roll.ms </li>


typo timestmp

junrao · 2016-08-19T00:22:28Z

@becketqin : Thanks for the latest patch. I made another pass and only had a few minor comments. Once those are addressed, I can merge in the patch.

becketqin · 2016-08-19T05:25:29Z

@junrao Thanks a lot for all the patient reviews. I updated the patch to address you latest comments and added a few unit tests. Thanks again.

junrao · 2016-08-19T17:05:05Z

Thanks for the patch. LGTM

junrao · 2016-08-19T17:08:21Z

core/src/main/scala/kafka/log/LogSegment.scala

+      if (iter.hasNext)
+        rollingBasedTimestamp = Some(iter.next.message.timestamp)
+      else
+        // If the log is empty, we return 0 as time waited.


The comment is no longer valid. We can probably just remove it.

junrao · 2016-08-19T17:13:31Z

@becketqin : Thanks a lot for working on this diligently. A couple of followups.

I left a couple of minor comments after merging. Could you address them in a followup jira?
Do you plan to work on a followup KIP to add a new seekToTimestamp() api to the consumer and a new request for getting offsets based on timestamp?

becketqin · 2016-08-22T16:33:50Z

@junrao Thanks a lot for the patient review. I will submit a follow up PR soon. And yes, I will start working on the KIP to add seekToTimestamp() API now and post the wiki this week.

ijuma · 2016-08-22T21:31:19Z

@becketqin, it would be good to update the Kafka documentation to mention the time index. Maybe an easy way is to search it for offset index and see if it makes sense to mention the time index in those cases as well. Would you be willing to file a JIRA and take this on as well?

…d commas (apache#1215)

becketqin force-pushed the KAFKA-3163 branch from 09d3505 to 06ace99 Compare April 12, 2016 00:29

becketqin reviewed Apr 12, 2016
View reviewed changes

junrao reviewed Aug 18, 2016
View reviewed changes

Addressed Jun's comments.

89a8ac9

asfgit closed this in 79d3fd2 Aug 19, 2016

junrao reviewed Aug 19, 2016
View reviewed changes

efeg pushed a commit to efeg/kafka that referenced this pull request May 29, 2024

Fix Java11 JavaDoc errors (apache#1215)

221fcae

mumrah pushed a commit to mumrah/kafka that referenced this pull request Aug 14, 2024

MINOR: InflightBatch and InflightState refactor for print - spaces an…

ec6d652

…d commas (apache#1215)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-3163: Add time based index to Kafka. #1215

KAFKA-3163: Add time based index to Kafka. #1215

becketqin commented Apr 11, 2016

becketqin commented Apr 11, 2016

becketqin commented Apr 12, 2016

becketqin Apr 12, 2016

junrao Apr 25, 2016

becketqin Apr 25, 2016

junrao commented Aug 18, 2016

becketqin commented Aug 18, 2016

junrao commented Aug 18, 2016

becketqin commented Aug 18, 2016 •

edited

Loading

junrao Aug 18, 2016

junrao commented Aug 19, 2016

becketqin commented Aug 19, 2016

junrao commented Aug 19, 2016

junrao Aug 19, 2016

junrao commented Aug 19, 2016

becketqin commented Aug 22, 2016

ijuma commented Aug 22, 2016

KAFKA-3163: Add time based index to Kafka. #1215

KAFKA-3163: Add time based index to Kafka. #1215

Conversation

becketqin commented Apr 11, 2016

becketqin commented Apr 11, 2016

becketqin commented Apr 12, 2016

becketqin Apr 12, 2016

Choose a reason for hiding this comment

junrao Apr 25, 2016

Choose a reason for hiding this comment

becketqin Apr 25, 2016

Choose a reason for hiding this comment

junrao commented Aug 18, 2016

becketqin commented Aug 18, 2016

junrao commented Aug 18, 2016

becketqin commented Aug 18, 2016 • edited Loading

junrao Aug 18, 2016

Choose a reason for hiding this comment

junrao commented Aug 19, 2016

becketqin commented Aug 19, 2016

junrao commented Aug 19, 2016

junrao Aug 19, 2016

Choose a reason for hiding this comment

junrao commented Aug 19, 2016

becketqin commented Aug 22, 2016

ijuma commented Aug 22, 2016

becketqin commented Aug 18, 2016 •

edited

Loading