Consumer timestamp search using metadata implementation #43
Conversation
jeffxiang
changed the title
| S3Records segmentRecords = null; | ||
| try { | ||
| segmentRecords = S3Records.open( | ||
| logObject.getLeft(), | ||
| logObject.getMiddle(), | ||
| startPosition, | ||
| false, | ||
| true, | ||
| logObject.getRight().intValue(), | ||
| true); | ||
|
|
||
| Iterator<S3ChannelRecordBatch> batches = segmentRecords.batchesFrom(startPosition).iterator(); | ||
| while (batches.hasNext()) { | ||
| S3ChannelRecordBatch batch = batches.next(); | ||
| for (Record record : batch) { | ||
| if (record.timestamp() >= targetTimestamp) { | ||
| return Optional.of(new OffsetAndTimestamp(record.offset(), record.timestamp(), Optional.empty())); | ||
| } | ||
| } | ||
| } | ||
| } catch (IOException e) { |
There was a problem hiding this comment.
Any chance of resource leak in case of an exception? Should this be a try-with-resources block instead?
There was a problem hiding this comment.
in the finally block we close the segmentRecords
| return Optional.empty(); | ||
| } | ||
|
|
||
| TimeIndex timeIndex = metadataOptional.get().getTimeIndex(); |
There was a problem hiding this comment.
Do we need a null check for metadataOptional.get() to avoid NPE?
There was a problem hiding this comment.
we never return a null for the metadataOptional or put a null value into the Optional<TopicPartitionMetadata>. The only check needed is Optional.isPresent() which is done above.
| * 2. Find the largest timestamp in the segment's timeindex file | ||
| * 3. If the largest timestamp in the segment's timeindex file is less than the target timestamp, we continue to the next segment. | ||
| * 4. If the largest timestamp in the segment's timeindex file is greater than or equal to the target timestamp, we have found the segment. | ||
| * 5. Find the segment timeindex entry with a timestamp less than or equal to the target timestamp. |
There was a problem hiding this comment.
For step 5, do we perform a binary search?
There was a problem hiding this comment.
Implicitly, yes. We do this by putting the entries into a sorted ConcurrentSkipListSet within the TimeIndex.load() method and then calling ConcurrentSkipListSet.floor() method. This is done in https://github.com/pinterest/tiered-storage/blob/consumer_seek/ts-common/src/main/java/com/pinterest/kafka/tieredstorage/common/metadata/TimeIndex.java#L296
ts-consumer/src/main/java/com/pinterest/kafka/tieredstorage/consumer/S3Utils.java
Show resolved
Hide resolved
offsetsForTimes() is now implemented using the
metadatafile in S3 to perform a more efficient search process. The mechanism is as follows:metadatafile from S3 for the particular topic-partitiontimeindexfrom the metadata, which provides each segment's last modified timestamp on Kafkaand we query
offsetsForTimes(timestamp=250), we start searching from000200.log. Even though000200.logdoesn't contain our target offset (the last modified timestamp of this segment is 200 while we're searching for timestamp=250), we still want to start our search here in case thetimeindexhas a gap immediately following this segment..timeindexfile and looking at the last entry, corresponding to the last record in that segment. If that entry's timestamp is less than the target timestamp, we continue onto the next segment. Usually, the very next segment is the target segment unless there is a gap in themetadatatimeindex entries