One motivation of my proposal is actually to avoid any clients trying to
read the timestamp type from the topic metadata response and behave
differently since:
1) topic metadata response is not always in-sync with the source-of-truth
(ZK), hence when the clients realized that the config has changed it may
already be too late (i.e. for consumer the records with the wrong timestamp
could already be returned to user).
2) the client logic could be a bit simpler, and this will benefit non-Java
development a lot. Also we can avoid adding this into the topic metadata
response.
Guozhang
On Tue, Jan 26, 2016 at 3:20 PM, Becket Qin <becket....@gmail.com> wrote:
> 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
> >>>
> >>
> >>
> >
>
--
-- Guozhang
