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
