My hesitation for the changed protocol is that I think If we will have
topic configuration returned in the topic metadata, the current protocol
makes more sense. Because the timestamp type is a topic level setting so we
don't need to put it into each message. That is assuming the timestamp type
change on a topic rarely happens and if it is ever needed, the existing
data should be wiped out.
Thanks,
Jiangjie (Becket) Qin
On Tue, Jan 26, 2016 at 2:07 PM, Becket Qin <becket....@gmail.com> wrote:
> Bump up this thread per discussion on the KIP hangout.
>
> During the implementation of the KIP, Guozhang raised another proposal on
> how to indicate the message timestamp type used by messages. So we want to
> see people's opinion on this proposal.
>
> The difference between current and the new proposal only differs on
> messages that are a) compressed, and b) using LogAppendTime
>
> For compressed messages using LogAppendTime, the timestamps in the current
> proposal is as below:
> 1. When a producer produces the messages, it tries to set timestamp to -1
> for inner messages if it knows LogAppendTime is used.
> 2. When a broker receives the messages, it will overwrite the timestamp of
> inner message to -1 if needed and write server time to the wrapper message.
> Broker will do re-compression if inner message timestamp is overwritten.
> 3. When a consumer receives the messages, it will see the inner message
> timestamp is -1 so the wrapper message timestamp is used.
>
> Implementation wise, this proposal requires the producer to set timestamp
> for inner messages correctly to avoid broker side re-compression. To do
> that, the short term solution is to let producer infer the timestamp type
> returned by broker in ProduceResponse and set correct timestamp afterwards.
> This means the first few batches will still need re-compression on the
> broker. The long term solution is to have producer get topic configuration
> during metadata update.
>
>
> The proposed modification is:
> 1. When a producer produces the messages, it always using create time.
> 2. When a broker receives the messages, it ignores the inner messages
> timestamp, but simply set a wrapper message timestamp type attribute bit to
> 1 and set the timestamp of the wrapper message to server time. (The broker
> will also set the timesatmp type attribute bit accordingly for
> non-compressed messages using LogAppendTime).
> 3. When a consumer receives the messages, it checks timestamp type
> attribute bit of wrapper message. If it is set to 1, the inner message's
> timestamp will be ignored and the wrapper message's timestamp will be used.
>
> This approach uses an extra attribute bit. The good thing of the modified
> protocol is consumers will be able to know the timestamp type. And
> re-compression on broker side is completely avoided no matter what value is
> sent by the producer. In this approach the inner messages will have wrong
> timestamps.
>
> We want to see if people have concerns over the modified approach.
>
> Thanks,
>
> Jiangjie (Becket) Qin
>
>
>
>
>
> On Tue, Dec 15, 2015 at 11:45 AM, Becket Qin <becket....@gmail.com> wrote:
>
>> Jun,
>>
>> 1. I agree it would be nice to have the timestamps used in a unified way.
>> My concern is that if we let server change timestamp of the inner message
>> for LogAppendTime, that will enforce the user who are using LogAppendTime
>> to always pay the recompression penalty. So using LogAppendTime makes
>> KIP-31 in vain.
>>
>> 4. If there are no entries in the log segment, we can read from the time
>> index before the previous log segment. If there is no previous entry
>> avaliable after we search until the earliest log segment, that means all
>> the previous log segment with a valid time index entry has been deleted. In
>> that case supposedly there should be only one log segment left - the active
>> log segment, we can simply set the latest timestamp to 0.
>>
>> Guozhang,
>>
>> Sorry for the confusion. by "the timestamp of the latest message" I
>> actually meant "the timestamp of the message with largest timestamp". So in
>> your example the "latest message" is 5.
>>
>> Thanks,
>>
>> Jiangjie (Becket) Qin
>>
>>
>>
>> On Tue, Dec 15, 2015 at 10:02 AM, Guozhang Wang <wangg...@gmail.com>
>> wrote:
>>
>>> Jun, Jiangjie,
>>>
>>> I am confused about 3) here, if we use "the timestamp of the latest
>>> message"
>>> then doesn't this mean we will roll the log whenever a message delayed by
>>> rolling time is received as well? Just to clarify, my understanding of
>>> "the
>>> timestamp of the latest message", for example in the following log, is 1,
>>> not 5:
>>>
>>> 2, 3, 4, 5, 1
>>>
>>> Guozhang
>>>
>>>
>>> On Mon, Dec 14, 2015 at 10:05 PM, Jun Rao <j...@confluent.io> wrote:
>>>
>>> > 1. Hmm, it's more intuitive if the consumer sees the same timestamp
>>> whether
>>> > the messages are compressed or not. When
>>> > message.timestamp.type=LogAppendTime,
>>> > we will need to set timestamp in each message if messages are not
>>> > compressed, so that the follower can get the same timestamp. So, it
>>> seems
>>> > that we should do the same thing for inner messages when messages are
>>> > compressed.
>>> >
>>> > 4. I thought on startup, we restore the timestamp of the latest
>>> message by
>>> > reading from the time index of the last log segment. So, what happens
>>> if
>>> > there are no index entries?
>>> >
>>> > Thanks,
>>> >
>>> > Jun
>>> >
>>> > On Mon, Dec 14, 2015 at 6:28 PM, Becket Qin <becket....@gmail.com>
>>> wrote:
>>> >
>>> > > Thanks for the explanation, Jun.
>>> > >
>>> > > 1. That makes sense. So maybe we can do the following:
>>> > > (a) Set the timestamp in the compressed message to latest timestamp
>>> of
>>> > all
>>> > > its inner messages. This works for both LogAppendTime and CreateTime.
>>> > > (b) If message.timestamp.type=LogAppendTime, the broker will
>>> overwrite
>>> > all
>>> > > the inner message timestamp to -1 if they are not set to -1. This is
>>> > mainly
>>> > > for topics that are using LogAppendTime. Hopefully the producer will
>>> set
>>> > > the timestamp to -1 in the ProducerRecord to avoid server side
>>> > > recompression.
>>> > >
>>> > > 3. I see. That works. So the semantic of log rolling becomes "roll
>>> out
>>> > the
>>> > > log segment if it has been inactive since the latest message has
>>> > arrived."
>>> > >
>>> > > 4. Yes. If the largest timestamp is in previous log segment. The time
>>> > index
>>> > > for the current log segment does not have a valid offset in current
>>> log
>>> > > segment to point to. Maybe in that case we should build an empty log
>>> > index.
>>> > >
>>> > > Thanks,
>>> > >
>>> > > Jiangjie (Becket) Qin
>>> > >
>>> > >
>>> > >
>>> > > On Mon, Dec 14, 2015 at 5:51 PM, Jun Rao <j...@confluent.io> wrote:
>>> > >
>>> > > > 1. I was thinking more about saving the decompression overhead in
>>> the
>>> > > > follower. Currently, the follower doesn't decompress the messages.
>>> To
>>> > > keep
>>> > > > it that way, the outer message needs to include the timestamp of
>>> the
>>> > > latest
>>> > > > inner message to build the time index in the follower. The simplest
>>> > thing
>>> > > > to do is to change the timestamp in the inner messages if
>>> necessary, in
>>> > > > which case there will be the recompression overhead. However, in
>>> the
>>> > case
>>> > > > when the timestamp of the inner messages don't have to be changed
>>> > > > (hopefully more common), there won't be the recompression
>>> overhead. In
>>> > > > either case, we always set the timestamp in the outer message to
>>> be the
>>> > > > timestamp of the latest inner message, in the leader.
>>> > > >
>>> > > > 3. Basically, in each log segment, we keep track of the timestamp
>>> of
>>> > the
>>> > > > latest message. If current time - timestamp of latest message > log
>>> > > rolling
>>> > > > interval, we roll a new log segment. So, if messages with later
>>> > > timestamps
>>> > > > keep getting added, we only roll new log segments based on size.
>>> On the
>>> > > > other hand, if no new messages are added to a log, we can force a
>>> log
>>> > > roll
>>> > > > based on time, which addresses the issue in (b).
>>> > > >
>>> > > > 4. Hmm, the index is per segment and should only point to
>>> positions in
>>> > > the
>>> > > > corresponding .log file, not previous ones, right?
>>> > > >
>>> > > > Thanks,
>>> > > >
>>> > > > Jun
>>> > > >
>>> > > >
>>> > > >
>>> > > > On Mon, Dec 14, 2015 at 3:10 PM, Becket Qin <becket....@gmail.com>
>>> > > wrote:
>>> > > >
>>> > > > > Hi Jun,
>>> > > > >
>>> > > > > Thanks a lot for the comments. Please see inline replies.
>>> > > > >
>>> > > > > Thanks,
>>> > > > >
>>> > > > > Jiangjie (Becket) Qin
>>> > > > >
>>> > > > > On Mon, Dec 14, 2015 at 10:19 AM, Jun Rao <j...@confluent.io>
>>> wrote:
>>> > > > >
>>> > > > > > Hi, Becket,
>>> > > > > >
>>> > > > > > Thanks for the proposal. Looks good overall. A few comments
>>> below.
>>> > > > > >
>>> > > > > > 1. KIP-32 didn't say what timestamp should be set in a
>>> compressed
>>> > > > > message.
>>> > > > > > We probably should set it to the timestamp of the latest
>>> messages
>>> > > > > included
>>> > > > > > in the compressed one. This way, during indexing, we don't
>>> have to
>>> > > > > > decompress the message.
>>> > > > > >
>>> > > > > That is a good point.
>>> > > > > In normal cases, broker needs to decompress the message for
>>> > > verification
>>> > > > > purpose anyway. So building time index does not add additional
>>> > > > > decompression.
>>> > > > > During time index recovery, however, having a timestamp in
>>> compressed
>>> > > > > message might save the decompression.
>>> > > > >
>>> > > > > Another thing I am thinking is we should make sure KIP-32 works
>>> well
>>> > > with
>>> > > > > KIP-31. i.e. we don't want to do recompression in order to add
>>> > > timestamp
>>> > > > to
>>> > > > > messages.
>>> > > > > Take the approach in my last email, the timestamp in the messages
>>> > will
>>> > > > > either all be overwritten by server if
>>> > > > > message.timestamp.type=LogAppendTime, or they will not be
>>> overwritten
>>> > > if
>>> > > > > message.timestamp.type=CreateTime.
>>> > > > >
>>> > > > > Maybe we can use the timestamp in compressed messages in the
>>> > following
>>> > > > way:
>>> > > > > If message.timestamp.type=LogAppendTime, we have to overwrite
>>> > > timestamps
>>> > > > > for all the messages. We can simply write the timestamp in the
>>> > > compressed
>>> > > > > message to avoid recompression.
>>> > > > > If message.timestamp.type=CreateTime, we do not need to
>>> overwrite the
>>> > > > > timestamps. We either reject the entire compressed message or We
>>> just
>>> > > > leave
>>> > > > > the compressed message timestamp to be -1.
>>> > > > >
>>> > > > > So the semantic of the timestamp field in compressed message
>>> field
>>> > > > becomes:
>>> > > > > if it is greater than 0, that means LogAppendTime is used, the
>>> > > timestamp
>>> > > > of
>>> > > > > the inner messages is the compressed message LogAppendTime. If
>>> it is
>>> > > -1,
>>> > > > > that means the CreateTime is used, the timestamp is in each
>>> > individual
>>> > > > > inner message.
>>> > > > >
>>> > > > > This sacrifice the speed of recovery but seems worthy because we
>>> > avoid
>>> > > > > recompression.
>>> > > > >
>>> > > > >
>>> > > > > > 2. In KIP-33, should we make the time-based index interval
>>> > > > configurable?
>>> > > > > > Perhaps we can default it 60 secs, but allow users to
>>> configure it
>>> > to
>>> > > > > > smaller values if they want more precision.
>>> > > > > >
>>> > > > > Yes, we can do that.
>>> > > > >
>>> > > > >
>>> > > > > > 3. In KIP-33, I am not sure if log rolling should be based on
>>> the
>>> > > > > earliest
>>> > > > > > message. This would mean that we will need to roll a log
>>> segment
>>> > > every
>>> > > > > time
>>> > > > > > we get a message delayed by the log rolling time interval.
>>> Also, on
>>> > > > > broker
>>> > > > > > startup, we can get the timestamp of the latest message in a
>>> log
>>> > > > segment
>>> > > > > > pretty efficiently by just looking at the last time index
>>> entry.
>>> > But
>>> > > > > > getting the timestamp of the earliest timestamp requires a full
>>> > scan
>>> > > of
>>> > > > > all
>>> > > > > > log segments, which can be expensive. Previously, there were
>>> two
>>> > use
>>> > > > > cases
>>> > > > > > of time-based rolling: (a) more accurate time-based indexing
>>> and
>>> > (b)
>>> > > > > > retaining data by time (since the active segment is never
>>> deleted).
>>> > > (a)
>>> > > > > is
>>> > > > > > already solved with a time-based index. For (b), if the
>>> retention
>>> > is
>>> > > > > based
>>> > > > > > on the timestamp of the latest message in a log segment,
>>> perhaps
>>> > log
>>> > > > > > rolling should be based on that too.
>>> > > > > >
>>> > > > > I am not sure how to make log rolling work with the latest
>>> timestamp
>>> > in
>>> > > > > current log segment. Do you mean the log rolling can based on the
>>> > last
>>> > > > log
>>> > > > > segment's latest timestamp? If so how do we roll out the first
>>> > segment?
>>> > > > >
>>> > > > >
>>> > > > > > 4. In KIP-33, I presume the timestamp in the time index will be
>>> > > > > > monotonically increasing. So, if all messages in a log segment
>>> > have a
>>> > > > > > timestamp less than the largest timestamp in the previous log
>>> > > segment,
>>> > > > we
>>> > > > > > will use the latter to index this log segment?
>>> > > > > >
>>> > > > > Yes. The timestamps are monotonically increasing. If the largest
>>> > > > timestamp
>>> > > > > in the previous segment is very big, it is possible the time
>>> index of
>>> > > the
>>> > > > > current segment only have two index entries (inserted during
>>> segment
>>> > > > > creation and roll out), both are pointing to a message in the
>>> > previous
>>> > > > log
>>> > > > > segment. This is the corner case I mentioned before that we
>>> should
>>> > > expire
>>> > > > > the next log segment even before expiring the previous log
>>> segment
>>> > just
>>> > > > > because the largest timestamp is in previous log segment. In
>>> current
>>> > > > > approach, we will wait until the previous log segment expires,
>>> and
>>> > then
>>> > > > > delete both the previous log segment and the next log segment.
>>> > > > >
>>> > > > >
>>> > > > > > 5. In KIP-32, in the wire protocol, we mention both timestamp
>>> and
>>> > > time.
>>> > > > > > They should be consistent.
>>> > > > > >
>>> > > > > Will fix the wiki page.
>>> > > > >
>>> > > > >
>>> > > > > > Jun
>>> > > > > >
>>> > > > > >
>>> > > > > >
>>> > > > > > On Thu, Dec 10, 2015 at 10:13 AM, Becket Qin <
>>> becket....@gmail.com
>>> > >
>>> > > > > wrote:
>>> > > > > >
>>> > > > > > > Hey Jay,
>>> > > > > > >
>>> > > > > > > Thanks for the comments.
>>> > > > > > >
>>> > > > > > > Good point about the actions after when
>>> > max.message.time.difference
>>> > > > is
>>> > > > > > > exceeded. Rejection is a useful behavior although I cannot
>>> think
>>> > of
>>> > > > use
>>> > > > > > > case at LinkedIn at this moment. I think it makes sense to
>>> add a
>>> > > > > > > configuration.
>>> > > > > > >
>>> > > > > > > How about the following configurations?
>>> > > > > > > 1. message.timestamp.type=CreateTime/LogAppendTime
>>> > > > > > > 2. max.message.time.difference.ms
>>> > > > > > >
>>> > > > > > > if message.timestamp.type is set to CreateTime, when the
>>> broker
>>> > > > > receives
>>> > > > > > a
>>> > > > > > > message, it will further check
>>> max.message.time.difference.ms,
>>> > and
>>> > > > > will
>>> > > > > > > reject the message it the time difference exceeds the
>>> threshold.
>>> > > > > > > If message.timestamp.type is set to LogAppendTime, the broker
>>> > will
>>> > > > > always
>>> > > > > > > stamp the message with current server time, regardless the
>>> value
>>> > of
>>> > > > > > > max.message.time.difference.ms
>>> > > > > > >
>>> > > > > > > This will make sure the message on the broker is either
>>> > CreateTime
>>> > > or
>>> > > > > > > LogAppendTime, but not mixture of both.
>>> > > > > > >
>>> > > > > > > What do you think?
>>> > > > > > >
>>> > > > > > > Thanks,
>>> > > > > > >
>>> > > > > > > Jiangjie (Becket) Qin
>>> > > > > > >
>>> > > > > > >
>>> > > > > > > On Wed, Dec 9, 2015 at 2:42 PM, Jay Kreps <j...@confluent.io>
>>> > > wrote:
>>> > > > > > >
>>> > > > > > > > Hey Becket,
>>> > > > > > > >
>>> > > > > > > > That summary of pros and cons sounds about right to me.
>>> > > > > > > >
>>> > > > > > > > There are potentially two actions you could take when
>>> > > > > > > > max.message.time.difference is exceeded--override it or
>>> reject
>>> > > the
>>> > > > > > > > message entirely. Can we pick one of these or does the
>>> action
>>> > > need
>>> > > > to
>>> > > > > > > > be configurable too? (I'm not sure). The downside of more
>>> > > > > > > > configuration is that it is more fiddly and has more modes.
>>> > > > > > > >
>>> > > > > > > > I suppose the reason I was thinking of this as a
>>> "difference"
>>> > > > rather
>>> > > > > > > > than a hard type was that if you were going to go the
>>> reject
>>> > mode
>>> > > > you
>>> > > > > > > > would need some tolerance setting (i.e. if your SLA is
>>> that if
>>> > > your
>>> > > > > > > > timestamp is off by more than 10 minutes I give you an
>>> error).
>>> > I
>>> > > > > agree
>>> > > > > > > > with you that having one field that is potentially
>>> containing a
>>> > > mix
>>> > > > > of
>>> > > > > > > > two values is a bit weird.
>>> > > > > > > >
>>> > > > > > > > -Jay
>>> > > > > > > >
>>> > > > > > > > On Mon, Dec 7, 2015 at 5:17 PM, Becket Qin <
>>> > becket....@gmail.com
>>> > > >
>>> > > > > > wrote:
>>> > > > > > > > > It looks the format of the previous email was messed up.
>>> Send
>>> > > it
>>> > > > > > again.
>>> > > > > > > > >
>>> > > > > > > > > Just to recap, the last proposal Jay made (with some
>>> > > > implementation
>>> > > > > > > > > details added)
>>> > > > > > > > > was:
>>> > > > > > > > >
>>> > > > > > > > > 1. Allow user to stamp the message when produce
>>> > > > > > > > >
>>> > > > > > > > > 2. When broker receives a message it take a look at the
>>> > > > difference
>>> > > > > > > > between
>>> > > > > > > > > its local time and the timestamp in the message.
>>> > > > > > > > > a. If the time difference is within a configurable
>>> > > > > > > > > max.message.time.difference.ms, the server will accept
>>> it
>>> > and
>>> > > > > append
>>> > > > > > > it
>>> > > > > > > > to
>>> > > > > > > > > the log.
>>> > > > > > > > > b. If the time difference is beyond the configured
>>> > > > > > > > > max.message.time.difference.ms, the server will
>>> override the
>>> > > > > > timestamp
>>> > > > > > > > with
>>> > > > > > > > > its current local time and append the message to the log.
>>> > > > > > > > > c. The default value of max.message.time.difference
>>> would
>>> > be
>>> > > > set
>>> > > > > to
>>> > > > > > > > > Long.MaxValue.
>>> > > > > > > > >
>>> > > > > > > > > 3. The configurable time difference threshold
>>> > > > > > > > > max.message.time.difference.ms will
>>> > > > > > > > > be a per topic configuration.
>>> > > > > > > > >
>>> > > > > > > > > 4. The indexed will be built so it has the following
>>> > guarantee.
>>> > > > > > > > > a. If user search by time stamp:
>>> > > > > > > > > - all the messages after that timestamp will be
>>> > consumed.
>>> > > > > > > > > - user might see earlier messages.
>>> > > > > > > > > b. The log retention will take a look at the last time
>>> > index
>>> > > > > entry
>>> > > > > > in
>>> > > > > > > > the
>>> > > > > > > > > time index file. Because the last entry will be the
>>> latest
>>> > > > > timestamp
>>> > > > > > in
>>> > > > > > > > the
>>> > > > > > > > > entire log segment. If that entry expires, the log
>>> segment
>>> > will
>>> > > > be
>>> > > > > > > > deleted.
>>> > > > > > > > > c. The log rolling has to depend on the earliest
>>> timestamp.
>>> > > In
>>> > > > > this
>>> > > > > > > > case
>>> > > > > > > > > we may need to keep a in memory timestamp only for the
>>> > current
>>> > > > > active
>>> > > > > > > > log.
>>> > > > > > > > > On recover, we will need to read the active log segment
>>> to
>>> > get
>>> > > > this
>>> > > > > > > > timestamp
>>> > > > > > > > > of the earliest messages.
>>> > > > > > > > >
>>> > > > > > > > > 5. The downside of this proposal are:
>>> > > > > > > > > a. The timestamp might not be monotonically increasing.
>>> > > > > > > > > b. The log retention might become non-deterministic.
>>> i.e.
>>> > > When
>>> > > > a
>>> > > > > > > > message
>>> > > > > > > > > will be deleted now depends on the timestamp of the other
>>> > > > messages
>>> > > > > in
>>> > > > > > > the
>>> > > > > > > > > same log segment. And those timestamps are provided by
>>> > > > > > > > > user within a range depending on what the time difference
>>> > > > threshold
>>> > > > > > > > > configuration is.
>>> > > > > > > > > c. The semantic meaning of the timestamp in the
>>> messages
>>> > > could
>>> > > > > be a
>>> > > > > > > > little
>>> > > > > > > > > bit vague because some of them come from the producer and
>>> > some
>>> > > of
>>> > > > > > them
>>> > > > > > > > are
>>> > > > > > > > > overwritten by brokers.
>>> > > > > > > > >
>>> > > > > > > > > 6. Although the proposal has some downsides, it gives
>>> user
>>> > the
>>> > > > > > > > flexibility
>>> > > > > > > > > to use the timestamp.
>>> > > > > > > > > a. If the threshold is set to Long.MaxValue. The
>>> timestamp
>>> > in
>>> > > > the
>>> > > > > > > > message is
>>> > > > > > > > > equivalent to CreateTime.
>>> > > > > > > > > b. If the threshold is set to 0. The timestamp in the
>>> > message
>>> > > > is
>>> > > > > > > > equivalent
>>> > > > > > > > > to LogAppendTime.
>>> > > > > > > > >
>>> > > > > > > > > This proposal actually allows user to use either
>>> CreateTime
>>> > or
>>> > > > > > > > LogAppendTime
>>> > > > > > > > > without introducing two timestamp concept at the same
>>> time. I
>>> > > > have
>>> > > > > > > > updated
>>> > > > > > > > > the wiki for KIP-32 and KIP-33 with this proposal.
>>> > > > > > > > >
>>> > > > > > > > > One thing I am thinking is that instead of having a time
>>> > > > difference
>>> > > > > > > > threshold,
>>> > > > > > > > > should we simply set have a TimestampType configuration?
>>> > > Because
>>> > > > in
>>> > > > > > > most
>>> > > > > > > > > cases, people will either set the threshold to 0 or
>>> > > > Long.MaxValue.
>>> > > > > > > > Setting
>>> > > > > > > > > anything in between will make the timestamp in the
>>> message
>>> > > > > > meaningless
>>> > > > > > > to
>>> > > > > > > > > user - user don't know if the timestamp has been
>>> overwritten
>>> > by
>>> > > > the
>>> > > > > > > > brokers.
>>> > > > > > > > >
>>> > > > > > > > > Any thoughts?
>>> > > > > > > > >
>>> > > > > > > > > Thanks,
>>> > > > > > > > > Jiangjie (Becket) Qin
>>> > > > > > > > >
>>> > > > > > > > > On Mon, Dec 7, 2015 at 10:33 AM, Jiangjie Qin
>>> > > > > > > <j...@linkedin.com.invalid
>>> > > > > > > > >
>>> > > > > > > > > wrote:
>>> > > > > > > > >
>>> > > > > > > > >> Bump up this thread.
>>> > > > > > > > >>
>>> > > > > > > > >> Just to recap, the last proposal Jay made (with some
>>> > > > > implementation
>>> > > > > > > > details
>>> > > > > > > > >> added) was:
>>> > > > > > > > >>
>>> > > > > > > > >> 1. Allow user to stamp the message when produce
>>> > > > > > > > >> 2. When broker receives a message it take a look at
>>> the
>>> > > > > > difference
>>> > > > > > > > >> between its local time and the timestamp in the
>>> message.
>>> > > > > > > > >> - If the time difference is within a configurable
>>> > > > > > > > >> max.message.time.difference.ms, the server will
>>> > accept
>>> > > it
>>> > > > > and
>>> > > > > > > > append
>>> > > > > > > > >> it to the log.
>>> > > > > > > > >> - If the time difference is beyond the configured
>>> > > > > > > > >> max.message.time.difference.ms, the server will
>>> > > override
>>> > > > > the
>>> > > > > > > > >> timestamp with its current local time and append
>>> the
>>> > > > message
>>> > > > > > to
>>> > > > > > > > the
>>> > > > > > > > >> log.
>>> > > > > > > > >> - The default value of max.message.time.difference
>>> > would
>>> > > > be
>>> > > > > > set
>>> > > > > > > to
>>> > > > > > > > >> Long.MaxValue.
>>> > > > > > > > >> 3. The configurable time difference threshold
>>> > > > > > > > >> max.message.time.difference.ms will be a per topic
>>> > > > > > configuration.
>>> > > > > > > > >> 4. The indexed will be built so it has the following
>>> > > > guarantee.
>>> > > > > > > > >> - If user search by time stamp:
>>> > > > > > > > >> - all the messages after that timestamp will be
>>> consumed.
>>> > > > > > > > >> - user might see earlier messages.
>>> > > > > > > > >> - The log retention will take a look at the last
>>> time
>>> > > > index
>>> > > > > > > entry
>>> > > > > > > > in
>>> > > > > > > > >> the time index file. Because the last entry will
>>> be
>>> > the
>>> > > > > latest
>>> > > > > > > > >> timestamp in
>>> > > > > > > > >> the entire log segment. If that entry expires,
>>> the log
>>> > > > > segment
>>> > > > > > > > will
>>> > > > > > > > >> be
>>> > > > > > > > >> deleted.
>>> > > > > > > > >> - The log rolling has to depend on the earliest
>>> > > timestamp.
>>> > > > > In
>>> > > > > > > this
>>> > > > > > > > >> case we may need to keep a in memory timestamp
>>> only
>>> > for
>>> > > > the
>>> > > > > > > > >> current active
>>> > > > > > > > >> log. On recover, we will need to read the active
>>> log
>>> > > > segment
>>> > > > > > to
>>> > > > > > > > get
>>> > > > > > > > >> this
>>> > > > > > > > >> timestamp of the earliest messages.
>>> > > > > > > > >> 5. The downside of this proposal are:
>>> > > > > > > > >> - The timestamp might not be monotonically
>>> increasing.
>>> > > > > > > > >> - The log retention might become
>>> non-deterministic.
>>> > i.e.
>>> > > > > When
>>> > > > > > a
>>> > > > > > > > >> message will be deleted now depends on the
>>> timestamp
>>> > of
>>> > > > the
>>> > > > > > > > >> other messages
>>> > > > > > > > >> in the same log segment. And those timestamps are
>>> > > provided
>>> > > > > by
>>> > > > > > > > >> user within a
>>> > > > > > > > >> range depending on what the time difference
>>> threshold
>>> > > > > > > > configuration
>>> > > > > > > > >> is.
>>> > > > > > > > >> - The semantic meaning of the timestamp in the
>>> > messages
>>> > > > > could
>>> > > > > > > be a
>>> > > > > > > > >> little bit vague because some of them come from
>>> the
>>> > > > producer
>>> > > > > > and
>>> > > > > > > > >> some of
>>> > > > > > > > >> them are overwritten by brokers.
>>> > > > > > > > >> 6. Although the proposal has some downsides, it
>>> gives
>>> > > user
>>> > > > > the
>>> > > > > > > > >> flexibility to use the timestamp.
>>> > > > > > > > >> - If the threshold is set to Long.MaxValue. The
>>> timestamp
>>> > > in
>>> > > > > the
>>> > > > > > > > message
>>> > > > > > > > >> is equivalent to CreateTime.
>>> > > > > > > > >> - If the threshold is set to 0. The timestamp in
>>> the
>>> > > > message
>>> > > > > > is
>>> > > > > > > > >> equivalent to LogAppendTime.
>>> > > > > > > > >>
>>> > > > > > > > >> This proposal actually allows user to use either
>>> CreateTime
>>> > or
>>> > > > > > > > >> LogAppendTime without introducing two timestamp concept
>>> at
>>> > the
>>> > > > > same
>>> > > > > > > > time. I
>>> > > > > > > > >> have updated the wiki for KIP-32 and KIP-33 with this
>>> > > proposal.
>>> > > > > > > > >>
>>> > > > > > > > >> One thing I am thinking is that instead of having a time
>>> > > > > difference
>>> > > > > > > > >> threshold, should we simply set have a TimestampType
>>> > > > > configuration?
>>> > > > > > > > Because
>>> > > > > > > > >> in most cases, people will either set the threshold to
>>> 0 or
>>> > > > > > > > Long.MaxValue.
>>> > > > > > > > >> Setting anything in between will make the timestamp in
>>> the
>>> > > > message
>>> > > > > > > > >> meaningless to user - user don't know if the timestamp
>>> has
>>> > > been
>>> > > > > > > > overwritten
>>> > > > > > > > >> by the brokers.
>>> > > > > > > > >>
>>> > > > > > > > >> Any thoughts?
>>> > > > > > > > >>
>>> > > > > > > > >> Thanks,
>>> > > > > > > > >> Jiangjie (Becket) Qin
>>> > > > > > > > >>
>>> > > > > > > > >> On Mon, Oct 26, 2015 at 1:23 PM, Jiangjie Qin <
>>> > > > j...@linkedin.com>
>>> > > > > > > > wrote:
>>> > > > > > > > >>
>>> > > > > > > > >> > Hi Jay,
>>> > > > > > > > >> >
>>> > > > > > > > >> > Thanks for such detailed explanation. I think we both
>>> are
>>> > > > trying
>>> > > > > > to
>>> > > > > > > > make
>>> > > > > > > > >> > CreateTime work for us if possible. To me by "work" it
>>> > means
>>> > > > > clear
>>> > > > > > > > >> > guarantees on:
>>> > > > > > > > >> > 1. Log Retention Time enforcement.
>>> > > > > > > > >> > 2. Log Rolling time enforcement (This might be less a
>>> > > concern
>>> > > > as
>>> > > > > > you
>>> > > > > > > > >> > pointed out)
>>> > > > > > > > >> > 3. Application search message by time.
>>> > > > > > > > >> >
>>> > > > > > > > >> > WRT (1), I agree the expectation for log retention
>>> might
>>> > be
>>> > > > > > > different
>>> > > > > > > > >> > depending on who we ask. But my concern is about the
>>> level
>>> > > of
>>> > > > > > > > guarantee
>>> > > > > > > > >> we
>>> > > > > > > > >> > give to user. My observation is that a clear
>>> guarantee to
>>> > > user
>>> > > > > is
>>> > > > > > > > >> critical
>>> > > > > > > > >> > regardless of the mechanism we choose. And this is the
>>> > > subtle
>>> > > > > but
>>> > > > > > > > >> important
>>> > > > > > > > >> > difference between using LogAppendTime and CreateTime.
>>> > > > > > > > >> >
>>> > > > > > > > >> > Let's say user asks this question: How long will my
>>> > message
>>> > > > stay
>>> > > > > > in
>>> > > > > > > > >> Kafka?
>>> > > > > > > > >> >
>>> > > > > > > > >> > If we use LogAppendTime for log retention, the answer
>>> is
>>> > > > message
>>> > > > > > > will
>>> > > > > > > > >> stay
>>> > > > > > > > >> > in Kafka for retention time after the message is
>>> produced
>>> > > (to
>>> > > > be
>>> > > > > > > more
>>> > > > > > > > >> > precise, upper bounded by log.rolling.ms +
>>> > log.retention.ms
>>> > > ).
>>> > > > > > User
>>> > > > > > > > has a
>>> > > > > > > > >> > clear guarantee and they may decide whether or not to
>>> put
>>> > > the
>>> > > > > > > message
>>> > > > > > > > >> into
>>> > > > > > > > >> > Kafka. Or how to adjust the retention time according
>>> to
>>> > > their
>>> > > > > > > > >> requirements.
>>> > > > > > > > >> > If we use create time for log retention, the answer
>>> would
>>> > be
>>> > > > it
>>> > > > > > > > depends.
>>> > > > > > > > >> > The best answer we can give is at least retention.ms
>>> > > because
>>> > > > > > there
>>> > > > > > > > is no
>>> > > > > > > > >> > guarantee when the messages will be deleted after
>>> that.
>>> > If a
>>> > > > > > message
>>> > > > > > > > sits
>>> > > > > > > > >> > somewhere behind a larger create time, the message
>>> might
>>> > > stay
>>> > > > > > longer
>>> > > > > > > > than
>>> > > > > > > > >> > expected. But we don't know how longer it would be
>>> because
>>> > > it
>>> > > > > > > depends
>>> > > > > > > > on
>>> > > > > > > > >> > the create time. In this case, it is hard for user to
>>> > decide
>>> > > > > what
>>> > > > > > to
>>> > > > > > > > do.
>>> > > > > > > > >> >
>>> > > > > > > > >> > I am worrying about this because a blurring guarantee
>>> has
>>> > > > bitten
>>> > > > > > us
>>> > > > > > > > >> > before, e.g. Topic creation. We have received many
>>> > questions
>>> > > > > like
>>> > > > > > > > "why my
>>> > > > > > > > >> > topic is not there after I created it". I can imagine
>>> we
>>> > > > receive
>>> > > > > > > > similar
>>> > > > > > > > >> > question asking "why my message is still there after
>>> > > retention
>>> > > > > > time
>>> > > > > > > > has
>>> > > > > > > > >> > reached". So my understanding is that a clear and
>>> solid
>>> > > > > guarantee
>>> > > > > > is
>>> > > > > > > > >> better
>>> > > > > > > > >> > than having a mechanism that works in most cases but
>>> > > > > occasionally
>>> > > > > > > does
>>> > > > > > > > >> not
>>> > > > > > > > >> > work.
>>> > > > > > > > >> >
>>> > > > > > > > >> > If we think of the retention guarantee we provide with
>>> > > > > > > LogAppendTime,
>>> > > > > > > > it
>>> > > > > > > > >> > is not broken as you said, because we are telling
>>> user the
>>> > > log
>>> > > > > > > > retention
>>> > > > > > > > >> is
>>> > > > > > > > >> > NOT based on create time at the first place.
>>> > > > > > > > >> >
>>> > > > > > > > >> > WRT (3), no matter whether we index on LogAppendTime
>>> or
>>> > > > > > CreateTime,
>>> > > > > > > > the
>>> > > > > > > > >> > best guarantee we can provide with user is "not
>>> missing
>>> > > > message
>>> > > > > > > after
>>> > > > > > > > a
>>> > > > > > > > >> > certain timestamp". Therefore I actually really like
>>> to
>>> > > index
>>> > > > on
>>> > > > > > > > >> CreateTime
>>> > > > > > > > >> > because that is the timestamp we provide to user, and
>>> we
>>> > can
>>> > > > > have
>>> > > > > > > the
>>> > > > > > > > >> solid
>>> > > > > > > > >> > guarantee.
>>> > > > > > > > >> > On the other hand, indexing on LogAppendTime and
>>> giving
>>> > user
>>> > > > > > > > CreateTime
>>> > > > > > > > >> > does not provide solid guarantee when user do search
>>> based
>>> > > on
>>> > > > > > > > timestamp.
>>> > > > > > > > >> It
>>> > > > > > > > >> > only works when LogAppendTime is always no earlier
>>> than
>>> > > > > > CreateTime.
>>> > > > > > > > This
>>> > > > > > > > >> is
>>> > > > > > > > >> > a reasonable assumption and we can easily enforce it.
>>> > > > > > > > >> >
>>> > > > > > > > >> > With above, I am not sure if we can avoid server
>>> timestamp
>>> > > to
>>> > > > > make
>>> > > > > > > log
>>> > > > > > > > >> > retention work with a clear guarantee. For searching
>>> by
>>> > > > > timestamp
>>> > > > > > > use
>>> > > > > > > > >> case,
>>> > > > > > > > >> > I really want to have the index built on CreateTime.
>>> But
>>> > > with
>>> > > > a
>>> > > > > > > > >> reasonable
>>> > > > > > > > >> > assumption and timestamp enforcement, a LogAppendTime
>>> > index
>>> > > > > would
>>> > > > > > > also
>>> > > > > > > > >> work.
>>> > > > > > > > >> >
>>> > > > > > > > >> > Thanks,
>>> > > > > > > > >> >
>>> > > > > > > > >> > Jiangjie (Becket) Qin
>>> > > > > > > > >> >
>>> > > > > > > > >> >
>>> > > > > > > > >> >
>>> > > > > > > > >> > On Thu, Oct 22, 2015 at 10:48 AM, Jay Kreps <
>>> > > j...@confluent.io
>>> > > > >
>>> > > > > > > wrote:
>>> > > > > > > > >> >
>>> > > > > > > > >> >> Hey Becket,
>>> > > > > > > > >> >>
>>> > > > > > > > >> >> Let me see if I can address your concerns:
>>> > > > > > > > >> >>
>>> > > > > > > > >> >> 1. Let's say we have two source clusters that are
>>> > mirrored
>>> > > to
>>> > > > > the
>>> > > > > > > > same
>>> > > > > > > > >> >> > target cluster. For some reason one of the mirror
>>> maker
>>> > > > from
>>> > > > > a
>>> > > > > > > > cluster
>>> > > > > > > > >> >> dies
>>> > > > > > > > >> >> > and after fix the issue we want to resume
>>> mirroring. In
>>> > > > this
>>> > > > > > case
>>> > > > > > > > it
>>> > > > > > > > >> is
>>> > > > > > > > >> >> > possible that when the mirror maker resumes
>>> mirroring,
>>> > > the
>>> > > > > > > > timestamp
>>> > > > > > > > >> of
>>> > > > > > > > >> >> the
>>> > > > > > > > >> >> > messages have already gone beyond the acceptable
>>> > > timestamp
>>> > > > > > range
>>> > > > > > > on
>>> > > > > > > > >> >> broker.
>>> > > > > > > > >> >> > In order to let those messages go through, we have
>>> to
>>> > > bump
>>> > > > up
>>> > > > > > the
>>> > > > > > > > >> >> > *max.append.delay
>>> > > > > > > > >> >> > *for all the topics on the target broker. This
>>> could be
>>> > > > > > painful.
>>> > > > > > > > >> >>
>>> > > > > > > > >> >>
>>> > > > > > > > >> >> Actually what I was suggesting was different. Here
>>> is my
>>> > > > > > > observation:
>>> > > > > > > > >> >> clusters/topics directly produced to by applications
>>> > have a
>>> > > > > valid
>>> > > > > > > > >> >> assertion
>>> > > > > > > > >> >> that log append time and create time are similar
>>> (let's
>>> > > call
>>> > > > > > these
>>> > > > > > > > >> >> "unbuffered"); other cluster/topic such as those that
>>> > > receive
>>> > > > > > data
>>> > > > > > > > from
>>> > > > > > > > >> a
>>> > > > > > > > >> >> database, a log file, or another kafka cluster don't
>>> have
>>> > > > that
>>> > > > > > > > >> assertion,
>>> > > > > > > > >> >> for these "buffered" clusters data can be arbitrarily
>>> > late.
>>> > > > > This
>>> > > > > > > > means
>>> > > > > > > > >> any
>>> > > > > > > > >> >> use of log append time on these buffered clusters is
>>> not
>>> > > very
>>> > > > > > > > >> meaningful,
>>> > > > > > > > >> >> and create time and log append time "should" be
>>> similar
>>> > on
>>> > > > > > > unbuffered
>>> > > > > > > > >> >> clusters so you can probably use either.
>>> > > > > > > > >> >>
>>> > > > > > > > >> >> Using log append time on buffered clusters actually
>>> > results
>>> > > > in
>>> > > > > > bad
>>> > > > > > > > >> things.
>>> > > > > > > > >> >> If you request the offset for a given time you get
>>> don't
>>> > > end
>>> > > > up
>>> > > > > > > > getting
>>> > > > > > > > >> >> data for that time but rather data that showed up at
>>> that
>>> > > > time.
>>> > > > > > If
>>> > > > > > > > you
>>> > > > > > > > >> try
>>> > > > > > > > >> >> to retain 7 days of data it may mostly work but any
>>> kind
>>> > of
>>> > > > > > > > >> bootstrapping
>>> > > > > > > > >> >> will result in retaining much more (potentially the
>>> whole
>>> > > > > > database
>>> > > > > > > > >> >> contents!).
>>> > > > > > > > >> >>
>>> > > > > > > > >> >> So what I am suggesting in terms of the use of the
>>> > > > > > max.append.delay
>>> > > > > > > > is
>>> > > > > > > > >> >> that
>>> > > > > > > > >> >> unbuffered clusters would have this set and buffered
>>> > > clusters
>>> > > > > > would
>>> > > > > > > > not.
>>> > > > > > > > >> >> In
>>> > > > > > > > >> >> other words, in LI terminology, tracking and metrics
>>> > > clusters
>>> > > > > > would
>>> > > > > > > > have
>>> > > > > > > > >> >> this enforced, aggregate and replica clusters
>>> wouldn't.
>>> > > > > > > > >> >>
>>> > > > > > > > >> >> So you DO have the issue of potentially maintaining
>>> more
>>> > > data
>>> > > > > > than
>>> > > > > > > > you
>>> > > > > > > > >> >> need
>>> > > > > > > > >> >> to on aggregate clusters if your mirroring skews,
>>> but you
>>> > > > DON'T
>>> > > > > > > need
>>> > > > > > > > to
>>> > > > > > > > >> >> tweak the setting as you described.
>>> > > > > > > > >> >>
>>> > > > > > > > >> >> 2. Let's say in the above scenario we let the
>>> messages
>>> > in,
>>> > > at
>>> > > > > > that
>>> > > > > > > > point
>>> > > > > > > > >> >> > some log segments in the target cluster might have
>>> a
>>> > wide
>>> > > > > range
>>> > > > > > > of
>>> > > > > > > > >> >> > timestamps, like Guozhang mentioned the log rolling
>>> > could
>>> > > > be
>>> > > > > > > tricky
>>> > > > > > > > >> >> because
>>> > > > > > > > >> >> > the first time index entry does not necessarily
>>> have
>>> > the
>>> > > > > > smallest
>>> > > > > > > > >> >> timestamp
>>> > > > > > > > >> >> > of all the messages in the log segment. Instead,
>>> it is
>>> > > the
>>> > > > > > > largest
>>> > > > > > > > >> >> > timestamp ever seen. We have to scan the entire
>>> log to
>>> > > find
>>> > > > > the
>>> > > > > > > > >> message
>>> > > > > > > > >> >> > with smallest offset to see if we should roll.
>>> > > > > > > > >> >>
>>> > > > > > > > >> >>
>>> > > > > > > > >> >> I think there are two uses for time-based log
>>> rolling:
>>> > > > > > > > >> >> 1. Making the offset lookup by timestamp work
>>> > > > > > > > >> >> 2. Ensuring we don't retain data indefinitely if it
>>> is
>>> > > > supposed
>>> > > > > > to
>>> > > > > > > > get
>>> > > > > > > > >> >> purged after 7 days
>>> > > > > > > > >> >>
>>> > > > > > > > >> >> But think about these two use cases. (1) is totally
>>> > > obviated
>>> > > > by
>>> > > > > > the
>>> > > > > > > > >> >> time=>offset index we are adding which yields much
>>> more
>>> > > > > granular
>>> > > > > > > > offset
>>> > > > > > > > >> >> lookups. (2) Is actually totally broken if you
>>> switch to
>>> > > > append
>>> > > > > > > time,
>>> > > > > > > > >> >> right? If you want to be sure for security/privacy
>>> > reasons
>>> > > > you
>>> > > > > > only
>>> > > > > > > > >> retain
>>> > > > > > > > >> >> 7 days of data then if the log append and create time
>>> > > diverge
>>> > > > > you
>>> > > > > > > > >> actually
>>> > > > > > > > >> >> violate this requirement.
>>> > > > > > > > >> >>
>>> > > > > > > > >> >> I think 95% of people care about (1) which is solved
>>> in
>>> > the
>>> > > > > > > proposal
>>> > > > > > > > and
>>> > > > > > > > >> >> (2) is actually broken today as well as in both
>>> > proposals.
>>> > > > > > > > >> >>
>>> > > > > > > > >> >> 3. Theoretically it is possible that an older log
>>> segment
>>> > > > > > contains
>>> > > > > > > > >> >> > timestamps that are older than all the messages in
>>> a
>>> > > newer
>>> > > > > log
>>> > > > > > > > >> segment.
>>> > > > > > > > >> >> It
>>> > > > > > > > >> >> > would be weird that we are supposed to delete the
>>> newer
>>> > > log
>>> > > > > > > segment
>>> > > > > > > > >> >> before
>>> > > > > > > > >> >> > we delete the older log segment.
>>> > > > > > > > >> >>
>>> > > > > > > > >> >>
>>> > > > > > > > >> >> The index timestamps would always be a lower bound
>>> (i.e.
>>> > > the
>>> > > > > > > maximum
>>> > > > > > > > at
>>> > > > > > > > >> >> that time) so I don't think that is possible.
>>> > > > > > > > >> >>
>>> > > > > > > > >> >> 4. In bootstrap case, if we reload the data to a
>>> Kafka
>>> > > > > cluster,
>>> > > > > > we
>>> > > > > > > > have
>>> > > > > > > > >> >> to
>>> > > > > > > > >> >> > make sure we configure the topic correctly before
>>> we
>>> > load
>>> > > > the
>>> > > > > > > data.
>>> > > > > > > > >> >> > Otherwise the message might either be rejected
>>> because
>>> > > the
>>> > > > > > > > timestamp
>>> > > > > > > > >> is
>>> > > > > > > > >> >> too
>>> > > > > > > > >> >> > old, or it might be deleted immediately because the
>>> > > > retention
>>> > > > > > > time
>>> > > > > > > > has
>>> > > > > > > > >> >> > reached.
>>> > > > > > > > >> >>
>>> > > > > > > > >> >>
>>> > > > > > > > >> >> See (1).
>>> > > > > > > > >> >>
>>> > > > > > > > >> >> -Jay
>>> > > > > > > > >> >>
>>> > > > > > > > >> >> On Tue, Oct 13, 2015 at 7:30 PM, Jiangjie Qin
>>> > > > > > > > <j...@linkedin.com.invalid
>>> > > > > > > > >> >
>>> > > > > > > > >> >> wrote:
>>> > > > > > > > >> >>
>>> > > > > > > > >> >> > Hey Jay and Guozhang,
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >> > Thanks a lot for the reply. So if I understand
>>> > correctly,
>>> > > > > Jay's
>>> > > > > > > > >> proposal
>>> > > > > > > > >> >> > is:
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >> > 1. Let client stamp the message create time.
>>> > > > > > > > >> >> > 2. Broker build index based on client-stamped
>>> message
>>> > > > create
>>> > > > > > > time.
>>> > > > > > > > >> >> > 3. Broker only takes message whose create time is
>>> > withing
>>> > > > > > current
>>> > > > > > > > time
>>> > > > > > > > >> >> > plus/minus T (T is a configuration
>>> *max.append.delay*,
>>> > > > could
>>> > > > > be
>>> > > > > > > > topic
>>> > > > > > > > >> >> level
>>> > > > > > > > >> >> > configuration), if the timestamp is out of this
>>> range,
>>> > > > broker
>>> > > > > > > > rejects
>>> > > > > > > > >> >> the
>>> > > > > > > > >> >> > message.
>>> > > > > > > > >> >> > 4. Because the create time of messages can be out
>>> of
>>> > > order,
>>> > > > > > when
>>> > > > > > > > >> broker
>>> > > > > > > > >> >> > builds the time based index it only provides the
>>> > > guarantee
>>> > > > > that
>>> > > > > > > if
>>> > > > > > > > a
>>> > > > > > > > >> >> > consumer starts consuming from the offset returned
>>> by
>>> > > > > searching
>>> > > > > > > by
>>> > > > > > > > >> >> > timestamp t, they will not miss any message created
>>> > after
>>> > > > t,
>>> > > > > > but
>>> > > > > > > > might
>>> > > > > > > > >> >> see
>>> > > > > > > > >> >> > some messages created before t.
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >> > To build the time based index, every time when a
>>> broker
>>> > > > needs
>>> > > > > > to
>>> > > > > > > > >> insert
>>> > > > > > > > >> >> a
>>> > > > > > > > >> >> > new time index entry, the entry would be
>>> > > > > > > > {Largest_Timestamp_Ever_Seen
>>> > > > > > > > >> ->
>>> > > > > > > > >> >> > Current_Offset}. This basically means any timestamp
>>> > > larger
>>> > > > > than
>>> > > > > > > the
>>> > > > > > > > >> >> > Largest_Timestamp_Ever_Seen must come after this
>>> offset
>>> > > > > because
>>> > > > > > > it
>>> > > > > > > > >> never
>>> > > > > > > > >> >> > saw them before. So we don't miss any message with
>>> > larger
>>> > > > > > > > timestamp.
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >> > (@Guozhang, in this case, for log retention we only
>>> > need
>>> > > to
>>> > > > > > take
>>> > > > > > > a
>>> > > > > > > > >> look
>>> > > > > > > > >> >> at
>>> > > > > > > > >> >> > the last time index entry, because it must be the
>>> > largest
>>> > > > > > > timestamp
>>> > > > > > > > >> >> ever,
>>> > > > > > > > >> >> > if that timestamp is overdue, we can safely delete
>>> any
>>> > > log
>>> > > > > > > segment
>>> > > > > > > > >> >> before
>>> > > > > > > > >> >> > that. So we don't need to scan the log segment
>>> file for
>>> > > log
>>> > > > > > > > retention)
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >> > I assume that we are still going to have the new
>>> > > > FetchRequest
>>> > > > > > to
>>> > > > > > > > allow
>>> > > > > > > > >> >> the
>>> > > > > > > > >> >> > time index replication for replicas.
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >> > I think Jay's main point here is that we don't
>>> want to
>>> > > have
>>> > > > > two
>>> > > > > > > > >> >> timestamp
>>> > > > > > > > >> >> > concepts in Kafka, which I agree is a reasonable
>>> > concern.
>>> > > > > And I
>>> > > > > > > > also
>>> > > > > > > > >> >> agree
>>> > > > > > > > >> >> > that create time is more meaningful than
>>> LogAppendTime
>>> > > for
>>> > > > > > users.
>>> > > > > > > > But
>>> > > > > > > > >> I
>>> > > > > > > > >> >> am
>>> > > > > > > > >> >> > not sure if making everything base on Create Time
>>> would
>>> > > > work
>>> > > > > in
>>> > > > > > > all
>>> > > > > > > > >> >> cases.
>>> > > > > > > > >> >> > Here are my questions about this approach:
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >> > 1. Let's say we have two source clusters that are
>>> > > mirrored
>>> > > > to
>>> > > > > > the
>>> > > > > > > > same
>>> > > > > > > > >> >> > target cluster. For some reason one of the mirror
>>> maker
>>> > > > from
>>> > > > > a
>>> > > > > > > > cluster
>>> > > > > > > > >> >> dies
>>> > > > > > > > >> >> > and after fix the issue we want to resume
>>> mirroring. In
>>> > > > this
>>> > > > > > case
>>> > > > > > > > it
>>> > > > > > > > >> is
>>> > > > > > > > >> >> > possible that when the mirror maker resumes
>>> mirroring,
>>> > > the
>>> > > > > > > > timestamp
>>> > > > > > > > >> of
>>> > > > > > > > >> >> the
>>> > > > > > > > >> >> > messages have already gone beyond the acceptable
>>> > > timestamp
>>> > > > > > range
>>> > > > > > > on
>>> > > > > > > > >> >> broker.
>>> > > > > > > > >> >> > In order to let those messages go through, we have
>>> to
>>> > > bump
>>> > > > up
>>> > > > > > the
>>> > > > > > > > >> >> > *max.append.delay
>>> > > > > > > > >> >> > *for all the topics on the target broker. This
>>> could be
>>> > > > > > painful.
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >> > 2. Let's say in the above scenario we let the
>>> messages
>>> > > in,
>>> > > > at
>>> > > > > > > that
>>> > > > > > > > >> point
>>> > > > > > > > >> >> > some log segments in the target cluster might have
>>> a
>>> > wide
>>> > > > > range
>>> > > > > > > of
>>> > > > > > > > >> >> > timestamps, like Guozhang mentioned the log rolling
>>> > could
>>> > > > be
>>> > > > > > > tricky
>>> > > > > > > > >> >> because
>>> > > > > > > > >> >> > the first time index entry does not necessarily
>>> have
>>> > the
>>> > > > > > smallest
>>> > > > > > > > >> >> timestamp
>>> > > > > > > > >> >> > of all the messages in the log segment. Instead,
>>> it is
>>> > > the
>>> > > > > > > largest
>>> > > > > > > > >> >> > timestamp ever seen. We have to scan the entire
>>> log to
>>> > > find
>>> > > > > the
>>> > > > > > > > >> message
>>> > > > > > > > >> >> > with smallest offset to see if we should roll.
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >> > 3. Theoretically it is possible that an older log
>>> > segment
>>> > > > > > > contains
>>> > > > > > > > >> >> > timestamps that are older than all the messages in
>>> a
>>> > > newer
>>> > > > > log
>>> > > > > > > > >> segment.
>>> > > > > > > > >> >> It
>>> > > > > > > > >> >> > would be weird that we are supposed to delete the
>>> newer
>>> > > log
>>> > > > > > > segment
>>> > > > > > > > >> >> before
>>> > > > > > > > >> >> > we delete the older log segment.
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >> > 4. In bootstrap case, if we reload the data to a
>>> Kafka
>>> > > > > cluster,
>>> > > > > > > we
>>> > > > > > > > >> have
>>> > > > > > > > >> >> to
>>> > > > > > > > >> >> > make sure we configure the topic correctly before
>>> we
>>> > load
>>> > > > the
>>> > > > > > > data.
>>> > > > > > > > >> >> > Otherwise the message might either be rejected
>>> because
>>> > > the
>>> > > > > > > > timestamp
>>> > > > > > > > >> is
>>> > > > > > > > >> >> too
>>> > > > > > > > >> >> > old, or it might be deleted immediately because the
>>> > > > retention
>>> > > > > > > time
>>> > > > > > > > has
>>> > > > > > > > >> >> > reached.
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >> > I am very concerned about the operational overhead
>>> and
>>> > > the
>>> > > > > > > > ambiguity
>>> > > > > > > > >> of
>>> > > > > > > > >> >> > guarantees we introduce if we purely rely on
>>> > CreateTime.
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >> > It looks to me that the biggest issue of adopting
>>> > > > CreateTime
>>> > > > > > > > >> everywhere
>>> > > > > > > > >> >> is
>>> > > > > > > > >> >> > CreateTime can have big gaps. These gaps could be
>>> > caused
>>> > > by
>>> > > > > > > several
>>> > > > > > > > >> >> cases:
>>> > > > > > > > >> >> > [1]. Faulty clients
>>> > > > > > > > >> >> > [2]. Natural delays from different source
>>> > > > > > > > >> >> > [3]. Bootstrap
>>> > > > > > > > >> >> > [4]. Failure recovery
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >> > Jay's alternative proposal solves [1], perhaps
>>> solve
>>> > [2]
>>> > > as
>>> > > > > > well
>>> > > > > > > > if we
>>> > > > > > > > >> >> are
>>> > > > > > > > >> >> > able to set a reasonable max.append.delay. But it
>>> does
>>> > > not
>>> > > > > seem
>>> > > > > > > > >> address
>>> > > > > > > > >> >> [3]
>>> > > > > > > > >> >> > and [4]. I actually doubt if [3] and [4] are
>>> solvable
>>> > > > because
>>> > > > > > it
>>> > > > > > > > looks
>>> > > > > > > > >> >> the
>>> > > > > > > > >> >> > CreateTime gap is unavoidable in those two cases.
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >> > Thanks,
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >> > Jiangjie (Becket) Qin
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >> > On Tue, Oct 13, 2015 at 3:23 PM, Guozhang Wang <
>>> > > > > > > wangg...@gmail.com
>>> > > > > > > > >
>>> > > > > > > > >> >> wrote:
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >> > > Just to complete Jay's option, here is my
>>> > > understanding:
>>> > > > > > > > >> >> > >
>>> > > > > > > > >> >> > > 1. For log retention: if we want to remove data
>>> > before
>>> > > > time
>>> > > > > > t,
>>> > > > > > > we
>>> > > > > > > > >> look
>>> > > > > > > > >> >> > into
>>> > > > > > > > >> >> > > the index file of each segment and find the
>>> largest
>>> > > > > timestamp
>>> > > > > > > t'
>>> > > > > > > > <
>>> > > > > > > > >> t,
>>> > > > > > > > >> >> > find
>>> > > > > > > > >> >> > > the corresponding timestamp and start scanning
>>> to the
>>> > > end
>>> > > > > of
>>> > > > > > > the
>>> > > > > > > > >> >> segment,
>>> > > > > > > > >> >> > > if there is no entry with timestamp >= t, we can
>>> > delete
>>> > > > > this
>>> > > > > > > > >> segment;
>>> > > > > > > > >> >> if
>>> > > > > > > > >> >> > a
>>> > > > > > > > >> >> > > segment's index smallest timestamp is larger
>>> than t,
>>> > we
>>> > > > can
>>> > > > > > > skip
>>> > > > > > > > >> that
>>> > > > > > > > >> >> > > segment.
>>> > > > > > > > >> >> > >
>>> > > > > > > > >> >> > > 2. For log rolling: if we want to start a new
>>> segment
>>> > > > after
>>> > > > > > > time
>>> > > > > > > > t,
>>> > > > > > > > >> we
>>> > > > > > > > >> >> > look
>>> > > > > > > > >> >> > > into the active segment's index file, if the
>>> largest
>>> > > > > > timestamp
>>> > > > > > > is
>>> > > > > > > > >> >> > already >
>>> > > > > > > > >> >> > > t, we can roll a new segment immediately; if it
>>> is <
>>> > t,
>>> > > > we
>>> > > > > > read
>>> > > > > > > > its
>>> > > > > > > > >> >> > > corresponding offset and start scanning to the
>>> end of
>>> > > the
>>> > > > > > > > segment,
>>> > > > > > > > >> if
>>> > > > > > > > >> >> we
>>> > > > > > > > >> >> > > find a record whose timestamp > t, we can roll a
>>> new
>>> > > > > segment.
>>> > > > > > > > >> >> > >
>>> > > > > > > > >> >> > > For log rolling we only need to possibly scan a
>>> small
>>> > > > > portion
>>> > > > > > > the
>>> > > > > > > > >> >> active
>>> > > > > > > > >> >> > > segment, which should be fine; for log retention
>>> we
>>> > may
>>> > > > in
>>> > > > > > the
>>> > > > > > > > worst
>>> > > > > > > > >> >> case
>>> > > > > > > > >> >> > > end up scanning all segments, but in practice we
>>> may
>>> > > skip
>>> > > > > > most
>>> > > > > > > of
>>> > > > > > > > >> them
>>> > > > > > > > >> >> > > since their smallest timestamp in the index file
>>> is
>>> > > > larger
>>> > > > > > than
>>> > > > > > > > t.
>>> > > > > > > > >> >> > >
>>> > > > > > > > >> >> > > Guozhang
>>> > > > > > > > >> >> > >
>>> > > > > > > > >> >> > >
>>> > > > > > > > >> >> > > On Tue, Oct 13, 2015 at 12:52 AM, Jay Kreps <
>>> > > > > > j...@confluent.io>
>>> > > > > > > > >> wrote:
>>> > > > > > > > >> >> > >
>>> > > > > > > > >> >> > > > I think it should be possible to index
>>> out-of-order
>>> > > > > > > timestamps.
>>> > > > > > > > >> The
>>> > > > > > > > >> >> > > > timestamp index would be similar to the offset
>>> > > index, a
>>> > > > > > > memory
>>> > > > > > > > >> >> mapped
>>> > > > > > > > >> >> > > file
>>> > > > > > > > >> >> > > > appended to as part of the log append, but
>>> would
>>> > have
>>> > > > the
>>> > > > > > > > format
>>> > > > > > > > >> >> > > > timestamp offset
>>> > > > > > > > >> >> > > > The timestamp entries would be monotonic and as
>>> > with
>>> > > > the
>>> > > > > > > offset
>>> > > > > > > > >> >> index
>>> > > > > > > > >> >> > > would
>>> > > > > > > > >> >> > > > be no more often then every 4k (or some
>>> > configurable
>>> > > > > > > threshold
>>> > > > > > > > to
>>> > > > > > > > >> >> keep
>>> > > > > > > > >> >> > > the
>>> > > > > > > > >> >> > > > index small--actually for timestamp it could
>>> > probably
>>> > > > be
>>> > > > > > much
>>> > > > > > > > more
>>> > > > > > > > >> >> > sparse
>>> > > > > > > > >> >> > > > than 4k).
>>> > > > > > > > >> >> > > >
>>> > > > > > > > >> >> > > > A search for a timestamp t yields an offset o
>>> > before
>>> > > > > which
>>> > > > > > no
>>> > > > > > > > >> prior
>>> > > > > > > > >> >> > > message
>>> > > > > > > > >> >> > > > has a timestamp >= t. In other words if you
>>> read
>>> > the
>>> > > > log
>>> > > > > > > > starting
>>> > > > > > > > >> >> with
>>> > > > > > > > >> >> > o
>>> > > > > > > > >> >> > > > you are guaranteed not to miss any messages
>>> > occurring
>>> > > > at
>>> > > > > t
>>> > > > > > or
>>> > > > > > > > >> later
>>> > > > > > > > >> >> > > though
>>> > > > > > > > >> >> > > > you may get many before t (due to
>>> > out-of-orderness).
>>> > > > > Unlike
>>> > > > > > > the
>>> > > > > > > > >> >> offset
>>> > > > > > > > >> >> > > > index this bound doesn't really have to be
>>> tight
>>> > > (i.e.
>>> > > > > > > > probably no
>>> > > > > > > > >> >> need
>>> > > > > > > > >> >> > > to
>>> > > > > > > > >> >> > > > go search the log itself, though you could).
>>> > > > > > > > >> >> > > >
>>> > > > > > > > >> >> > > > -Jay
>>> > > > > > > > >> >> > > >
>>> > > > > > > > >> >> > > > On Tue, Oct 13, 2015 at 12:32 AM, Jay Kreps <
>>> > > > > > > j...@confluent.io>
>>> > > > > > > > >> >> wrote:
>>> > > > > > > > >> >> > > >
>>> > > > > > > > >> >> > > > > Here's my basic take:
>>> > > > > > > > >> >> > > > > - I agree it would be nice to have a notion
>>> of
>>> > time
>>> > > > > baked
>>> > > > > > > in
>>> > > > > > > > if
>>> > > > > > > > >> it
>>> > > > > > > > >> >> > were
>>> > > > > > > > >> >> > > > > done right
>>> > > > > > > > >> >> > > > > - All the proposals so far seem pretty
>>> complex--I
>>> > > > think
>>> > > > > > > they
>>> > > > > > > > >> might
>>> > > > > > > > >> >> > make
>>> > > > > > > > >> >> > > > > things worse rather than better overall
>>> > > > > > > > >> >> > > > > - I think adding 2x8 byte timestamps to the
>>> > message
>>> > > > is
>>> > > > > > > > probably
>>> > > > > > > > >> a
>>> > > > > > > > >> >> > > > > non-starter from a size perspective
>>> > > > > > > > >> >> > > > > - Even if it isn't in the message, having two
>>> > > notions
>>> > > > > of
>>> > > > > > > time
>>> > > > > > > > >> that
>>> > > > > > > > >> >> > > > control
>>> > > > > > > > >> >> > > > > different things is a bit confusing
>>> > > > > > > > >> >> > > > > - The mechanics of basing retention etc on
>>> log
>>> > > append
>>> > > > > > time
>>> > > > > > > > when
>>> > > > > > > > >> >> > that's
>>> > > > > > > > >> >> > > > not
>>> > > > > > > > >> >> > > > > in the log seem complicated
>>> > > > > > > > >> >> > > > >
>>> > > > > > > > >> >> > > > > To that end here is a possible 4th option.
>>> Let me
>>> > > > know
>>> > > > > > what
>>> > > > > > > > you
>>> > > > > > > > >> >> > think.
>>> > > > > > > > >> >> > > > >
>>> > > > > > > > >> >> > > > > The basic idea is that the message creation
>>> time
>>> > is
>>> > > > > > closest
>>> > > > > > > > to
>>> > > > > > > > >> >> what
>>> > > > > > > > >> >> > the
>>> > > > > > > > >> >> > > > > user actually cares about but is dangerous
>>> if set
>>> > > > > wrong.
>>> > > > > > So
>>> > > > > > > > >> rather
>>> > > > > > > > >> >> > than
>>> > > > > > > > >> >> > > > > substitute another notion of time, let's try
>>> to
>>> > > > ensure
>>> > > > > > the
>>> > > > > > > > >> >> > correctness
>>> > > > > > > > >> >> > > of
>>> > > > > > > > >> >> > > > > message creation time by preventing
>>> arbitrarily
>>> > bad
>>> > > > > > message
>>> > > > > > > > >> >> creation
>>> > > > > > > > >> >> > > > times.
>>> > > > > > > > >> >> > > > >
>>> > > > > > > > >> >> > > > > First, let's see if we can agree that log
>>> append
>>> > > time
>>> > > > > is
>>> > > > > > > not
>>> > > > > > > > >> >> > something
>>> > > > > > > > >> >> > > > > anyone really cares about but rather an
>>> > > > implementation
>>> > > > > > > > detail.
>>> > > > > > > > >> The
>>> > > > > > > > >> >> > > > > timestamp that matters to the user is when
>>> the
>>> > > > message
>>> > > > > > > > occurred
>>> > > > > > > > >> >> (the
>>> > > > > > > > >> >> > > > > creation time). The log append time is
>>> basically
>>> > > just
>>> > > > > an
>>> > > > > > > > >> >> > approximation
>>> > > > > > > > >> >> > > to
>>> > > > > > > > >> >> > > > > this on the assumption that the message
>>> creation
>>> > > and
>>> > > > > the
>>> > > > > > > > message
>>> > > > > > > > >> >> > > receive
>>> > > > > > > > >> >> > > > on
>>> > > > > > > > >> >> > > > > the server occur pretty close together and
>>> the
>>> > > reason
>>> > > > > to
>>> > > > > > > > prefer
>>> > > > > > > > >> .
>>> > > > > > > > >> >> > > > >
>>> > > > > > > > >> >> > > > > But as these values diverge the issue starts
>>> to
>>> > > > become
>>> > > > > > > > apparent.
>>> > > > > > > > >> >> Say
>>> > > > > > > > >> >> > > you
>>> > > > > > > > >> >> > > > > set the retention to one week and then mirror
>>> > data
>>> > > > > from a
>>> > > > > > > > topic
>>> > > > > > > > >> >> > > > containing
>>> > > > > > > > >> >> > > > > two years of retention. Your intention is
>>> clearly
>>> > > to
>>> > > > > keep
>>> > > > > > > the
>>> > > > > > > > >> last
>>> > > > > > > > >> >> > > week,
>>> > > > > > > > >> >> > > > > but because the mirroring is appending right
>>> now
>>> > > you
>>> > > > > will
>>> > > > > > > > keep
>>> > > > > > > > >> two
>>> > > > > > > > >> >> > > years.
>>> > > > > > > > >> >> > > > >
>>> > > > > > > > >> >> > > > > The reason we are liking log append time is
>>> > because
>>> > > > we
>>> > > > > > are
>>> > > > > > > > >> >> > > (justifiably)
>>> > > > > > > > >> >> > > > > concerned that in certain situations the
>>> creation
>>> > > > time
>>> > > > > > may
>>> > > > > > > > not
>>> > > > > > > > >> be
>>> > > > > > > > >> >> > > > > trustworthy. This same problem exists on the
>>> > > servers
>>> > > > > but
>>> > > > > > > > there
>>> > > > > > > > >> are
>>> > > > > > > > >> >> > > fewer
>>> > > > > > > > >> >> > > > > servers and they just run the kafka code so
>>> it is
>>> > > > less
>>> > > > > of
>>> > > > > > > an
>>> > > > > > > > >> >> issue.
>>> > > > > > > > >> >> > > > >
>>> > > > > > > > >> >> > > > > There are two possible ways to handle this:
>>> > > > > > > > >> >> > > > >
>>> > > > > > > > >> >> > > > > 1. Just tell people to add size based
>>> > > retention. I
>>> > > > > > think
>>> > > > > > > > this
>>> > > > > > > > >> >> is
>>> > > > > > > > >> >> > not
>>> > > > > > > > >> >> > > > > entirely unreasonable, we're basically
>>> saying
>>> > we
>>> > > > > > retain
>>> > > > > > > > data
>>> > > > > > > > >> >> based
>>> > > > > > > > >> >> > > on
>>> > > > > > > > >> >> > > > the
>>> > > > > > > > >> >> > > > > timestamp you give us in the data. If you
>>> give
>>> > > us
>>> > > > > bad
>>> > > > > > > > data we
>>> > > > > > > > >> >> will
>>> > > > > > > > >> >> > > > retain
>>> > > > > > > > >> >> > > > > it for a bad amount of time. If you want
>>> to
>>> > > ensure
>>> > > > > we
>>> > > > > > > > don't
>>> > > > > > > > >> >> retain
>>> > > > > > > > >> >> > > > "too
>>> > > > > > > > >> >> > > > > much" data, define "too much" by setting a
>>> > > > > time-based
>>> > > > > > > > >> retention
>>> > > > > > > > >> >> > > > setting.
>>> > > > > > > > >> >> > > > > This is not entirely unreasonable but
>>> kind of
>>> > > > > suffers
>>> > > > > > > > from a
>>> > > > > > > > >> >> "one
>>> > > > > > > > >> >> > > bad
>>> > > > > > > > >> >> > > > > apple" problem in a very large
>>> environment.
>>> > > > > > > > >> >> > > > > 2. Prevent bad timestamps. In general we
>>> can't
>>> > > > say a
>>> > > > > > > > >> timestamp
>>> > > > > > > > >> >> is
>>> > > > > > > > >> >> > > bad.
>>> > > > > > > > >> >> > > > > However the definition we're implicitly
>>> using
>>> > is
>>> > > > > that
>>> > > > > > we
>>> > > > > > > > >> think
>>> > > > > > > > >> >> > there
>>> > > > > > > > >> >> > > > are a
>>> > > > > > > > >> >> > > > > set of topics/clusters where the creation
>>> > > > timestamp
>>> > > > > > > should
>>> > > > > > > > >> >> always
>>> > > > > > > > >> >> > be
>>> > > > > > > > >> >> > > > "very
>>> > > > > > > > >> >> > > > > close" to the log append timestamp. This
>>> is
>>> > true
>>> > > > for
>>> > > > > > > data
>>> > > > > > > > >> >> sources
>>> > > > > > > > >> >> > > > that have
>>> > > > > > > > >> >> > > > > no buffering capability (which at
>>> LinkedIn is
>>> > > very
>>> > > > > > > common,
>>> > > > > > > > >> but
>>> > > > > > > > >> >> is
>>> > > > > > > > >> >> > > > more rare
>>> > > > > > > > >> >> > > > > elsewhere). The solution in this case
>>> would be
>>> > > to
>>> > > > > > allow
>>> > > > > > > a
>>> > > > > > > > >> >> setting
>>> > > > > > > > >> >> > > > along the
>>> > > > > > > > >> >> > > > > lines of max.append.delay which checks the
>>> > > > creation
>>> > > > > > > > timestamp
>>> > > > > > > > >> >> > > against
>>> > > > > > > > >> >> > > > the
>>> > > > > > > > >> >> > > > > server time to look for too large a
>>> > divergence.
>>> > > > The
>>> > > > > > > > solution
>>> > > > > > > > >> >> would
>>> > > > > > > > >> >> > > > either
>>> > > > > > > > >> >> > > > > be to reject the message or to override it
>>> > with
>>> > > > the
>>> > > > > > > server
>>> > > > > > > > >> >> time.
>>> > > > > > > > >> >> > > > >
>>> > > > > > > > >> >> > > > > So in LI's environment you would configure
>>> the
>>> > > > clusters
>>> > > > > > > used
>>> > > > > > > > for
>>> > > > > > > > >> >> > > direct,
>>> > > > > > > > >> >> > > > > unbuffered, message production (e.g.
>>> tracking and
>>> > > > > metrics
>>> > > > > > > > local)
>>> > > > > > > > >> >> to
>>> > > > > > > > >> >> > > > enforce
>>> > > > > > > > >> >> > > > > a reasonably aggressive timestamp bound (say
>>> 10
>>> > > > mins),
>>> > > > > > and
>>> > > > > > > > all
>>> > > > > > > > >> >> other
>>> > > > > > > > >> >> > > > > clusters would just inherent these.
>>> > > > > > > > >> >> > > > >
>>> > > > > > > > >> >> > > > > The downside of this approach is requiring
>>> the
>>> > > > special
>>> > > > > > > > >> >> configuration.
>>> > > > > > > > >> >> > > > > However I think in 99% of environments this
>>> could
>>> > > be
>>> > > > > > > skipped
>>> > > > > > > > >> >> > entirely,
>>> > > > > > > > >> >> > > > it's
>>> > > > > > > > >> >> > > > > only when the ratio of clients to servers
>>> gets so
>>> > > > > massive
>>> > > > > > > > that
>>> > > > > > > > >> you
>>> > > > > > > > >> >> > need
>>> > > > > > > > >> >> > > > to
>>> > > > > > > > >> >> > > > > do this. The primary upside is that you have
>>> a
>>> > > single
>>> > > > > > > > >> >> authoritative
>>> > > > > > > > >> >> > > > notion
>>> > > > > > > > >> >> > > > > of time which is closest to what a user would
>>> > want
>>> > > > and
>>> > > > > is
>>> > > > > > > > stored
>>> > > > > > > > >> >> > > directly
>>> > > > > > > > >> >> > > > > in the message.
>>> > > > > > > > >> >> > > > >
>>> > > > > > > > >> >> > > > > I'm also assuming there is a workable
>>> approach
>>> > for
>>> > > > > > indexing
>>> > > > > > > > >> >> > > non-monotonic
>>> > > > > > > > >> >> > > > > timestamps, though I haven't actually worked
>>> that
>>> > > > out.
>>> > > > > > > > >> >> > > > >
>>> > > > > > > > >> >> > > > > -Jay
>>> > > > > > > > >> >> > > > >
>>> > > > > > > > >> >> > > > > On Mon, Oct 5, 2015 at 8:52 PM, Jiangjie Qin
>>> > > > > > > > >> >> > <j...@linkedin.com.invalid
>>> > > > > > > > >> >> > > >
>>> > > > > > > > >> >> > > > > wrote:
>>> > > > > > > > >> >> > > > >
>>> > > > > > > > >> >> > > > >> Bumping up this thread although most of the
>>> > > > discussion
>>> > > > > > > were
>>> > > > > > > > on
>>> > > > > > > > >> >> the
>>> > > > > > > > >> >> > > > >> discussion thread of KIP-31 :)
>>> > > > > > > > >> >> > > > >>
>>> > > > > > > > >> >> > > > >> I just updated the KIP page to add detailed
>>> > > solution
>>> > > > > for
>>> > > > > > > the
>>> > > > > > > > >> >> option
>>> > > > > > > > >> >> > > > >> (Option
>>> > > > > > > > >> >> > > > >> 3) that does not expose the LogAppendTime to
>>> > user.
>>> > > > > > > > >> >> > > > >>
>>> > > > > > > > >> >> > > > >>
>>> > > > > > > > >> >> > > > >>
>>> > > > > > > > >> >> > > >
>>> > > > > > > > >> >> > >
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >>
>>> > > > > > > > >>
>>> > > > > > > >
>>> > > > > > >
>>> > > > > >
>>> > > > >
>>> > > >
>>> > >
>>> >
>>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-32+-+Add+CreateTime+and+LogAppendTime+to+Kafka+message
>>> > > > > > > > >> >> > > > >>
>>> > > > > > > > >> >> > > > >> The option has a minor change to the fetch
>>> > request
>>> > > > to
>>> > > > > > > allow
>>> > > > > > > > >> >> fetching
>>> > > > > > > > >> >> > > > time
>>> > > > > > > > >> >> > > > >> index entry as well. I kind of like this
>>> > solution
>>> > > > > > because
>>> > > > > > > > its
>>> > > > > > > > >> >> just
>>> > > > > > > > >> >> > > doing
>>> > > > > > > > >> >> > > > >> what we need without introducing other
>>> things.
>>> > > > > > > > >> >> > > > >>
>>> > > > > > > > >> >> > > > >> It will be great to see what are the
>>> feedback. I
>>> > > can
>>> > > > > > > explain
>>> > > > > > > > >> more
>>> > > > > > > > >> >> > > during
>>> > > > > > > > >> >> > > > >> tomorrow's KIP hangout.
>>> > > > > > > > >> >> > > > >>
>>> > > > > > > > >> >> > > > >> Thanks,
>>> > > > > > > > >> >> > > > >>
>>> > > > > > > > >> >> > > > >> Jiangjie (Becket) Qin
>>> > > > > > > > >> >> > > > >>
>>> > > > > > > > >> >> > > > >> On Thu, Sep 10, 2015 at 2:47 PM, Jiangjie
>>> Qin <
>>> > > > > > > > >> j...@linkedin.com
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >> > > > wrote:
>>> > > > > > > > >> >> > > > >>
>>> > > > > > > > >> >> > > > >> > Hi Jay,
>>> > > > > > > > >> >> > > > >> >
>>> > > > > > > > >> >> > > > >> > I just copy/pastes here your feedback on
>>> the
>>> > > > > timestamp
>>> > > > > > > > >> proposal
>>> > > > > > > > >> >> > that
>>> > > > > > > > >> >> > > > was
>>> > > > > > > > >> >> > > > >> > in the discussion thread of KIP-31.
>>> Please see
>>> > > the
>>> > > > > > > replies
>>> > > > > > > > >> >> inline.
>>> > > > > > > > >> >> > > > >> > The main change I made compared with
>>> previous
>>> > > > > proposal
>>> > > > > > > is
>>> > > > > > > > to
>>> > > > > > > > >> >> add
>>> > > > > > > > >> >> > > both
>>> > > > > > > > >> >> > > > >> > CreateTime and LogAppendTime to the
>>> message.
>>> > > > > > > > >> >> > > > >> >
>>> > > > > > > > >> >> > > > >> > On Tue, Sep 8, 2015 at 10:57 AM, Jay
>>> Kreps <
>>> > > > > > > > j...@confluent.io
>>> > > > > > > > >> >
>>> > > > > > > > >> >> > > wrote:
>>> > > > > > > > >> >> > > > >> >
>>> > > > > > > > >> >> > > > >> > > Hey Beckett,
>>> > > > > > > > >> >> > > > >> > >
>>> > > > > > > > >> >> > > > >> > > I was proposing splitting up the KIP
>>> just
>>> > for
>>> > > > > > > > simplicity of
>>> > > > > > > > >> >> > > > >> discussion.
>>> > > > > > > > >> >> > > > >> > You
>>> > > > > > > > >> >> > > > >> > > can still implement them in one patch. I
>>> > think
>>> > > > > > > > otherwise it
>>> > > > > > > > >> >> will
>>> > > > > > > > >> >> > > be
>>> > > > > > > > >> >> > > > >> hard
>>> > > > > > > > >> >> > > > >> > to
>>> > > > > > > > >> >> > > > >> > > discuss/vote on them since if you like
>>> the
>>> > > > offset
>>> > > > > > > > proposal
>>> > > > > > > > >> >> but
>>> > > > > > > > >> >> > not
>>> > > > > > > > >> >> > > > the
>>> > > > > > > > >> >> > > > >> > time
>>> > > > > > > > >> >> > > > >> > > proposal what do you do?
>>> > > > > > > > >> >> > > > >> > >
>>> > > > > > > > >> >> > > > >> > > Introducing a second notion of time into
>>> > Kafka
>>> > > > is
>>> > > > > a
>>> > > > > > > > pretty
>>> > > > > > > > >> >> > massive
>>> > > > > > > > >> >> > > > >> > > philosophical change so it kind of
>>> warrants
>>> > > it's
>>> > > > > own
>>> > > > > > > > KIP I
>>> > > > > > > > >> >> think
>>> > > > > > > > >> >> > > it
>>> > > > > > > > >> >> > > > >> > isn't
>>> > > > > > > > >> >> > > > >> > > just "Change message format".
>>> > > > > > > > >> >> > > > >> > >
>>> > > > > > > > >> >> > > > >> > > WRT time I think one thing to clarify
>>> in the
>>> > > > > > proposal
>>> > > > > > > is
>>> > > > > > > > >> how
>>> > > > > > > > >> >> MM
>>> > > > > > > > >> >> > > will
>>> > > > > > > > >> >> > > > >> have
>>> > > > > > > > >> >> > > > >> > > access to set the timestamp? Presumably
>>> this
>>> > > > will
>>> > > > > > be a
>>> > > > > > > > new
>>> > > > > > > > >> >> field
>>> > > > > > > > >> >> > > in
>>> > > > > > > > >> >> > > > >> > > ProducerRecord, right? If so then any
>>> user
>>> > can
>>> > > > set
>>> > > > > > the
>>> > > > > > > > >> >> > timestamp,
>>> > > > > > > > >> >> > > > >> right?
>>> > > > > > > > >> >> > > > >> > > I'm not sure you answered the questions
>>> > around
>>> > > > how
>>> > > > > > > this
>>> > > > > > > > >> will
>>> > > > > > > > >> >> > work
>>> > > > > > > > >> >> > > > for
>>> > > > > > > > >> >> > > > >> MM
>>> > > > > > > > >> >> > > > >> > > since when MM retains timestamps from
>>> > multiple
>>> > > > > > > > partitions
>>> > > > > > > > >> >> they
>>> > > > > > > > >> >> > > will
>>> > > > > > > > >> >> > > > >> then
>>> > > > > > > > >> >> > > > >> > be
>>> > > > > > > > >> >> > > > >> > > out of order and in the past (so the
>>> > > > > > > > >> >> max(lastAppendedTimestamp,
>>> > > > > > > > >> >> > > > >> > > currentTimeMillis) override you proposed
>>> > will
>>> > > > not
>>> > > > > > > work,
>>> > > > > > > > >> >> right?).
>>> > > > > > > > >> >> > > If
>>> > > > > > > > >> >> > > > we
>>> > > > > > > > >> >> > > > >> > > don't do this then when you set up
>>> mirroring
>>> > > the
>>> > > > > > data
>>> > > > > > > > will
>>> > > > > > > > >> >> all
>>> > > > > > > > >> >> > be
>>> > > > > > > > >> >> > > > new
>>> > > > > > > > >> >> > > > >> and
>>> > > > > > > > >> >> > > > >> > > you have the same retention problem you
>>> > > > described.
>>> > > > > > > > Maybe I
>>> > > > > > > > >> >> > missed
>>> > > > > > > > >> >> > > > >> > > something...?
>>> > > > > > > > >> >> > > > >> > lastAppendedTimestamp means the timestamp
>>> of
>>> > the
>>> > > > > > message
>>> > > > > > > > that
>>> > > > > > > > >> >> last
>>> > > > > > > > >> >> > > > >> > appended to the log.
>>> > > > > > > > >> >> > > > >> > If a broker is a leader, since it will
>>> assign
>>> > > the
>>> > > > > > > > timestamp
>>> > > > > > > > >> by
>>> > > > > > > > >> >> > > itself,
>>> > > > > > > > >> >> > > > >> the
>>> > > > > > > > >> >> > > > >> > lastAppenedTimestamp will be its local
>>> clock
>>> > > when
>>> > > > > > append
>>> > > > > > > > the
>>> > > > > > > > >> >> last
>>> > > > > > > > >> >> > > > >> message.
>>> > > > > > > > >> >> > > > >> > So if there is no leader migration,
>>> > > > > > > > >> max(lastAppendedTimestamp,
>>> > > > > > > > >> >> > > > >> > currentTimeMillis) = currentTimeMillis.
>>> > > > > > > > >> >> > > > >> > If a broker is a follower, because it will
>>> > keep
>>> > > > the
>>> > > > > > > > leader's
>>> > > > > > > > >> >> > > timestamp
>>> > > > > > > > >> >> > > > >> > unchanged, the lastAppendedTime would be
>>> the
>>> > > > > leader's
>>> > > > > > > > clock
>>> > > > > > > > >> >> when
>>> > > > > > > > >> >> > it
>>> > > > > > > > >> >> > > > >> appends
>>> > > > > > > > >> >> > > > >> > that message message. It keeps track of
>>> the
>>> > > > > > > > lastAppendedTime
>>> > > > > > > > >> >> only
>>> > > > > > > > >> >> > in
>>> > > > > > > > >> >> > > > >> case
>>> > > > > > > > >> >> > > > >> > it becomes leader later on. At that
>>> point, it
>>> > is
>>> > > > > > > possible
>>> > > > > > > > >> that
>>> > > > > > > > >> >> the
>>> > > > > > > > >> >> > > > >> > timestamp of the last appended message was
>>> > > stamped
>>> > > > > by
>>> > > > > > > old
>>> > > > > > > > >> >> leader,
>>> > > > > > > > >> >> > > but
>>> > > > > > > > >> >> > > > >> the
>>> > > > > > > > >> >> > > > >> > new leader's currentTimeMillis <
>>> > > lastAppendedTime.
>>> > > > > If
>>> > > > > > a
>>> > > > > > > > new
>>> > > > > > > > >> >> > message
>>> > > > > > > > >> >> > > > >> comes,
>>> > > > > > > > >> >> > > > >> > instead of stamp it with new leader's
>>> > > > > > currentTimeMillis,
>>> > > > > > > > we
>>> > > > > > > > >> >> have
>>> > > > > > > > >> >> > to
>>> > > > > > > > >> >> > > > >> stamp
>>> > > > > > > > >> >> > > > >> > it to lastAppendedTime to avoid the
>>> timestamp
>>> > in
>>> > > > the
>>> > > > > > log
>>> > > > > > > > >> going
>>> > > > > > > > >> >> > > > backward.
>>> > > > > > > > >> >> > > > >> > The max(lastAppendedTimestamp,
>>> > > currentTimeMillis)
>>> > > > is
>>> > > > > > > > purely
>>> > > > > > > > >> >> based
>>> > > > > > > > >> >> > on
>>> > > > > > > > >> >> > > > the
>>> > > > > > > > >> >> > > > >> > broker side clock. If MM produces message
>>> with
>>> > > > > > different
>>> > > > > > > > >> >> > > LogAppendTime
>>> > > > > > > > >> >> > > > >> in
>>> > > > > > > > >> >> > > > >> > source clusters to the same target
>>> cluster,
>>> > the
>>> > > > > > > > LogAppendTime
>>> > > > > > > > >> >> will
>>> > > > > > > > >> >> > > be
>>> > > > > > > > >> >> > > > >> > ignored re-stamped by target cluster.
>>> > > > > > > > >> >> > > > >> > I added a use case example for mirror
>>> maker in
>>> > > > > KIP-32.
>>> > > > > > > > Also
>>> > > > > > > > >> >> there
>>> > > > > > > > >> >> > > is a
>>> > > > > > > > >> >> > > > >> > corner case discussion about when we need
>>> the
>>> > > > > > > > >> >> > max(lastAppendedTime,
>>> > > > > > > > >> >> > > > >> > currentTimeMillis) trick. Could you take a
>>> > look
>>> > > > and
>>> > > > > > see
>>> > > > > > > if
>>> > > > > > > > >> that
>>> > > > > > > > >> >> > > > answers
>>> > > > > > > > >> >> > > > >> > your question?
>>> > > > > > > > >> >> > > > >> >
>>> > > > > > > > >> >> > > > >> > >
>>> > > > > > > > >> >> > > > >> > > My main motivation is that given that
>>> both
>>> > > Samza
>>> > > > > and
>>> > > > > > > > Kafka
>>> > > > > > > > >> >> > streams
>>> > > > > > > > >> >> > > > are
>>> > > > > > > > >> >> > > > >> > > doing work that implies a mandatory
>>> > > > client-defined
>>> > > > > > > > notion
>>> > > > > > > > >> of
>>> > > > > > > > >> >> > > time, I
>>> > > > > > > > >> >> > > > >> > really
>>> > > > > > > > >> >> > > > >> > > think introducing a different mandatory
>>> > notion
>>> > > > of
>>> > > > > > time
>>> > > > > > > > in
>>> > > > > > > > >> >> Kafka
>>> > > > > > > > >> >> > is
>>> > > > > > > > >> >> > > > >> going
>>> > > > > > > > >> >> > > > >> > to
>>> > > > > > > > >> >> > > > >> > > be quite odd. We should think hard
>>> about how
>>> > > > > > > > client-defined
>>> > > > > > > > >> >> time
>>> > > > > > > > >> >> > > > could
>>> > > > > > > > >> >> > > > >> > > work. I'm not sure if it can, but I'm
>>> also
>>> > not
>>> > > > > sure
>>> > > > > > > > that it
>>> > > > > > > > >> >> > can't.
>>> > > > > > > > >> >> > > > >> Having
>>> > > > > > > > >> >> > > > >> > > both will be odd. Did you chat about
>>> this
>>> > with
>>> > > > > > > > Yi/Kartik on
>>> > > > > > > > >> >> the
>>> > > > > > > > >> >> > > > Samza
>>> > > > > > > > >> >> > > > >> > side?
>>> > > > > > > > >> >> > > > >> > I talked with Kartik and realized that it
>>> > would
>>> > > be
>>> > > > > > > useful
>>> > > > > > > > to
>>> > > > > > > > >> >> have
>>> > > > > > > > >> >> > a
>>> > > > > > > > >> >> > > > >> client
>>> > > > > > > > >> >> > > > >> > timestamp to facilitate use cases like
>>> stream
>>> > > > > > > processing.
>>> > > > > > > > >> >> > > > >> > I was trying to figure out if we can
>>> simply
>>> > use
>>> > > > > client
>>> > > > > > > > >> >> timestamp
>>> > > > > > > > >> >> > > > without
>>> > > > > > > > >> >> > > > >> > introducing the server time. There are
>>> some
>>> > > > > discussion
>>> > > > > > > in
>>> > > > > > > > the
>>> > > > > > > > >> >> KIP.
>>> > > > > > > > >> >> > > > >> > The key problem we want to solve here is
>>> > > > > > > > >> >> > > > >> > 1. We want log retention and rolling to
>>> depend
>>> > > on
>>> > > > > > server
>>> > > > > > > > >> clock.
>>> > > > > > > > >> >> > > > >> > 2. We want to make sure the log-assiciated
>>> > > > timestamp
>>> > > > > > to
>>> > > > > > > be
>>> > > > > > > > >> >> > retained
>>> > > > > > > > >> >> > > > when
>>> > > > > > > > >> >> > > > >> > replicas moves.
>>> > > > > > > > >> >> > > > >> > 3. We want to use the timestamp in some
>>> way
>>> > that
>>> > > > can
>>> > > > > > > allow
>>> > > > > > > > >> >> > searching
>>> > > > > > > > >> >> > > > by
>>> > > > > > > > >> >> > > > >> > timestamp.
>>> > > > > > > > >> >> > > > >> > For 1 and 2, an alternative is to pass the
>>> > > > > > > log-associated
>>> > > > > > > > >> >> > timestamp
>>> > > > > > > > >> >> > > > >> > through replication, that means we need to
>>> > have
>>> > > a
>>> > > > > > > > different
>>> > > > > > > > >> >> > protocol
>>> > > > > > > > >> >> > > > for
>>> > > > > > > > >> >> > > > >> > replica fetching to pass log-associated
>>> > > timestamp.
>>> > > > > It
>>> > > > > > is
>>> > > > > > > > >> >> actually
>>> > > > > > > > >> >> > > > >> > complicated and there could be a lot of
>>> corner
>>> > > > cases
>>> > > > > > to
>>> > > > > > > > >> handle.
>>> > > > > > > > >> >> > e.g.
>>> > > > > > > > >> >> > > > >> what
>>> > > > > > > > >> >> > > > >> > if an old leader started to fetch from
>>> the new
>>> > > > > leader,
>>> > > > > > > > should
>>> > > > > > > > >> >> it
>>> > > > > > > > >> >> > > also
>>> > > > > > > > >> >> > > > >> > update all of its old log segment
>>> timestamp?
>>> > > > > > > > >> >> > > > >> > I think actually client side timestamp
>>> would
>>> > be
>>> > > > > better
>>> > > > > > > > for 3
>>> > > > > > > > >> >> if we
>>> > > > > > > > >> >> > > can
>>> > > > > > > > >> >> > > > >> > find a way to make it work.
>>> > > > > > > > >> >> > > > >> > So far I am not able to convince myself
>>> that
>>> > > only
>>> > > > > > having
>>> > > > > > > > >> client
>>> > > > > > > > >> >> > side
>>> > > > > > > > >> >> > > > >> > timestamp would work mainly because 1 and
>>> 2.
>>> > > There
>>> > > > > > are a
>>> > > > > > > > few
>>> > > > > > > > >> >> > > > situations
>>> > > > > > > > >> >> > > > >> I
>>> > > > > > > > >> >> > > > >> > mentioned in the KIP.
>>> > > > > > > > >> >> > > > >> > >
>>> > > > > > > > >> >> > > > >> > > When you are saying it won't work you
>>> are
>>> > > > assuming
>>> > > > > > > some
>>> > > > > > > > >> >> > particular
>>> > > > > > > > >> >> > > > >> > > implementation? Maybe that the index is
>>> a
>>> > > > > > > monotonically
>>> > > > > > > > >> >> > increasing
>>> > > > > > > > >> >> > > > >> set of
>>> > > > > > > > >> >> > > > >> > > pointers to the least record with a
>>> > timestamp
>>> > > > > larger
>>> > > > > > > > than
>>> > > > > > > > >> the
>>> > > > > > > > >> >> > > index
>>> > > > > > > > >> >> > > > >> time?
>>> > > > > > > > >> >> > > > >> > > In other words a search for time X
>>> gives the
>>> > > > > largest
>>> > > > > > > > offset
>>> > > > > > > > >> >> at
>>> > > > > > > > >> >> > > which
>>> > > > > > > > >> >> > > > >> all
>>> > > > > > > > >> >> > > > >> > > records are <= X?
>>> > > > > > > > >> >> > > > >> > It is a promising idea. We probably can
>>> have
>>> > an
>>> > > > > > > in-memory
>>> > > > > > > > >> index
>>> > > > > > > > >> >> > like
>>> > > > > > > > >> >> > > > >> that,
>>> > > > > > > > >> >> > > > >> > but might be complicated to have a file on
>>> > disk
>>> > > > like
>>> > > > > > > that.
>>> > > > > > > > >> >> Imagine
>>> > > > > > > > >> >> > > > there
>>> > > > > > > > >> >> > > > >> > are two timestamps T0 < T1. We see
>>> message Y
>>> > > > created
>>> > > > > > at
>>> > > > > > > T1
>>> > > > > > > > >> and
>>> > > > > > > > >> >> > > created
>>> > > > > > > > >> >> > > > >> > index like [T1->Y], then we see message
>>> > created
>>> > > at
>>> > > > > T1,
>>> > > > > > > > >> >> supposedly
>>> > > > > > > > >> >> > we
>>> > > > > > > > >> >> > > > >> should
>>> > > > > > > > >> >> > > > >> > have index look like [T0->X, T1->Y], it is
>>> > easy
>>> > > to
>>> > > > > do
>>> > > > > > in
>>> > > > > > > > >> >> memory,
>>> > > > > > > > >> >> > but
>>> > > > > > > > >> >> > > > we
>>> > > > > > > > >> >> > > > >> > might have to rewrite the index file
>>> > completely.
>>> > > > > Maybe
>>> > > > > > > we
>>> > > > > > > > can
>>> > > > > > > > >> >> have
>>> > > > > > > > >> >> > > the
>>> > > > > > > > >> >> > > > >> > first entry with timestamp to 0, and only
>>> > update
>>> > > > the
>>> > > > > > > first
>>> > > > > > > > >> >> pointer
>>> > > > > > > > >> >> > > for
>>> > > > > > > > >> >> > > > >> any
>>> > > > > > > > >> >> > > > >> > out of range timestamp, so the index will
>>> be
>>> > > > [0->X,
>>> > > > > > > > T1->Y].
>>> > > > > > > > >> >> Also,
>>> > > > > > > > >> >> > > the
>>> > > > > > > > >> >> > > > >> range
>>> > > > > > > > >> >> > > > >> > of timestamps in the log segments can
>>> overlap
>>> > > with
>>> > > > > > each
>>> > > > > > > > >> other.
>>> > > > > > > > >> >> > That
>>> > > > > > > > >> >> > > > >> means
>>> > > > > > > > >> >> > > > >> > we either need to keep a cross segments
>>> index
>>> > > file
>>> > > > > or
>>> > > > > > we
>>> > > > > > > > need
>>> > > > > > > > >> >> to
>>> > > > > > > > >> >> > > check
>>> > > > > > > > >> >> > > > >> all
>>> > > > > > > > >> >> > > > >> > the index file for each log segment.
>>> > > > > > > > >> >> > > > >> > I separated out the time based log index
>>> to
>>> > > KIP-33
>>> > > > > > > > because it
>>> > > > > > > > >> >> can
>>> > > > > > > > >> >> > be
>>> > > > > > > > >> >> > > > an
>>> > > > > > > > >> >> > > > >> > independent follow up feature as Neha
>>> > > suggested. I
>>> > > > > > will
>>> > > > > > > > try
>>> > > > > > > > >> to
>>> > > > > > > > >> >> > make
>>> > > > > > > > >> >> > > > the
>>> > > > > > > > >> >> > > > >> > time based index work with client side
>>> > > timestamp.
>>> > > > > > > > >> >> > > > >> > >
>>> > > > > > > > >> >> > > > >> > > For retention, I agree with the problem
>>> you
>>> > > > point
>>> > > > > > out,
>>> > > > > > > > but
>>> > > > > > > > >> I
>>> > > > > > > > >> >> > think
>>> > > > > > > > >> >> > > > >> what
>>> > > > > > > > >> >> > > > >> > you
>>> > > > > > > > >> >> > > > >> > > are saying in that case is that you
>>> want a
>>> > > size
>>> > > > > > limit
>>> > > > > > > > too.
>>> > > > > > > > >> If
>>> > > > > > > > >> >> > you
>>> > > > > > > > >> >> > > > use
>>> > > > > > > > >> >> > > > >> > > system time you actually hit the same
>>> > problem:
>>> > > > say
>>> > > > > > you
>>> > > > > > > > do a
>>> > > > > > > > >> >> full
>>> > > > > > > > >> >> > > > dump
>>> > > > > > > > >> >> > > > >> of
>>> > > > > > > > >> >> > > > >> > a
>>> > > > > > > > >> >> > > > >> > > DB table with a setting of 7 days
>>> retention,
>>> > > > your
>>> > > > > > > > retention
>>> > > > > > > > >> >> will
>>> > > > > > > > >> >> > > > >> actually
>>> > > > > > > > >> >> > > > >> > > not get enforced for the first 7 days
>>> > because
>>> > > > the
>>> > > > > > data
>>> > > > > > > > is
>>> > > > > > > > >> >> "new
>>> > > > > > > > >> >> > to
>>> > > > > > > > >> >> > > > >> Kafka".
>>> > > > > > > > >> >> > > > >> > I kind of think the size limit here is
>>> > > orthogonal.
>>> > > > > It
>>> > > > > > > is a
>>> > > > > > > > >> >> valid
>>> > > > > > > > >> >> > use
>>> > > > > > > > >> >> > > > >> case
>>> > > > > > > > >> >> > > > >> > where people only want to use time based
>>> > > retention
>>> > > > > > only.
>>> > > > > > > > In
>>> > > > > > > > >> >> your
>>> > > > > > > > >> >> > > > >> example,
>>> > > > > > > > >> >> > > > >> > depending on client timestamp might break
>>> the
>>> > > > > > > > functionality -
>>> > > > > > > > >> >> say
>>> > > > > > > > >> >> > it
>>> > > > > > > > >> >> > > > is
>>> > > > > > > > >> >> > > > >> a
>>> > > > > > > > >> >> > > > >> > bootstrap case people actually need to
>>> read
>>> > all
>>> > > > the
>>> > > > > > > data.
>>> > > > > > > > If
>>> > > > > > > > >> we
>>> > > > > > > > >> >> > > depend
>>> > > > > > > > >> >> > > > >> on
>>> > > > > > > > >> >> > > > >> > the client timestamp, the data might be
>>> > deleted
>>> > > > > > > instantly
>>> > > > > > > > >> when
>>> > > > > > > > >> >> > they
>>> > > > > > > > >> >> > > > >> come to
>>> > > > > > > > >> >> > > > >> > the broker. It might be too demanding to
>>> > expect
>>> > > > the
>>> > > > > > > > broker to
>>> > > > > > > > >> >> > > > understand
>>> > > > > > > > >> >> > > > >> > what people actually want to do with the
>>> data
>>> > > > coming
>>> > > > > > in.
>>> > > > > > > > So
>>> > > > > > > > >> the
>>> > > > > > > > >> >> > > > >> guarantee
>>> > > > > > > > >> >> > > > >> > of using server side timestamp is that
>>> "after
>>> > > > > appended
>>> > > > > > > to
>>> > > > > > > > the
>>> > > > > > > > >> >> log,
>>> > > > > > > > >> >> > > all
>>> > > > > > > > >> >> > > > >> > messages will be available on broker for
>>> > > retention
>>> > > > > > > time",
>>> > > > > > > > >> >> which is
>>> > > > > > > > >> >> > > not
>>> > > > > > > > >> >> > > > >> > changeable by clients.
>>> > > > > > > > >> >> > > > >> > >
>>> > > > > > > > >> >> > > > >> > > -Jay
>>> > > > > > > > >> >> > > > >> >
>>> > > > > > > > >> >> > > > >> > On Thu, Sep 10, 2015 at 12:55 PM, Jiangjie
>>> > Qin <
>>> > > > > > > > >> >> j...@linkedin.com
>>> > > > > > > > >> >> > >
>>> > > > > > > > >> >> > > > >> wrote:
>>> > > > > > > > >> >> > > > >> >
>>> > > > > > > > >> >> > > > >> >> Hi folks,
>>> > > > > > > > >> >> > > > >> >>
>>> > > > > > > > >> >> > > > >> >> This proposal was previously in KIP-31
>>> and we
>>> > > > > > separated
>>> > > > > > > > it
>>> > > > > > > > >> to
>>> > > > > > > > >> >> > > KIP-32
>>> > > > > > > > >> >> > > > >> per
>>> > > > > > > > >> >> > > > >> >> Neha and Jay's suggestion.
>>> > > > > > > > >> >> > > > >> >>
>>> > > > > > > > >> >> > > > >> >> The proposal is to add the following two
>>> > > > timestamps
>>> > > > > > to
>>> > > > > > > > Kafka
>>> > > > > > > > >> >> > > message.
>>> > > > > > > > >> >> > > > >> >> - CreateTime
>>> > > > > > > > >> >> > > > >> >> - LogAppendTime
>>> > > > > > > > >> >> > > > >> >>
>>> > > > > > > > >> >> > > > >> >> The CreateTime will be set by the
>>> producer
>>> > and
>>> > > > will
>>> > > > > > > > change
>>> > > > > > > > >> >> after
>>> > > > > > > > >> >> > > > that.
>>> > > > > > > > >> >> > > > >> >> The LogAppendTime will be set by broker
>>> for
>>> > > > purpose
>>> > > > > > > such
>>> > > > > > > > as
>>> > > > > > > > >> >> > enforce
>>> > > > > > > > >> >> > > > log
>>> > > > > > > > >> >> > > > >> >> retention and log rolling.
>>> > > > > > > > >> >> > > > >> >>
>>> > > > > > > > >> >> > > > >> >> Thanks,
>>> > > > > > > > >> >> > > > >> >>
>>> > > > > > > > >> >> > > > >> >> Jiangjie (Becket) Qin
>>> > > > > > > > >> >> > > > >> >>
>>> > > > > > > > >> >> > > > >> >>
>>> > > > > > > > >> >> > > > >> >
>>> > > > > > > > >> >> > > > >>
>>> > > > > > > > >> >> > > > >
>>> > > > > > > > >> >> > > > >
>>> > > > > > > > >> >> > > >
>>> > > > > > > > >> >> > >
>>> > > > > > > > >> >> > >
>>> > > > > > > > >> >> > >
>>> > > > > > > > >> >> > > --
>>> > > > > > > > >> >> > > -- Guozhang
>>> > > > > > > > >> >> > >
>>> > > > > > > > >> >> >
>>> > > > > > > > >> >>
>>> > > > > > > > >> >
>>> > > > > > > > >> >
>>> > > > > > > > >>
>>> > > > > > > >
>>> > > > > > > >
>>> > > > > > >
>>> > > > > >
>>> > > > >
>>> > > >
>>> > >
>>> >
>>>
>>>
>>>
>>> --
>>> -- Guozhang
>>>
>>
>>
>
