Great, can we change the name to something related to the change--"KIP-31: Move to relative offsets in compressed message sets".
Also you had mentioned before you were going to expand on the mechanics of handling these log format changes, right? -Jay On Thu, Sep 10, 2015 at 12:42 PM, Jiangjie Qin <j...@linkedin.com.invalid> wrote: > Neha and Jay, > > Thanks a lot for the feedback. Good point about splitting the discussion. I > have split the proposal to three KIPs and it does make each discussion more > clear: > KIP-31 - Message format change (Use relative offset) > KIP-32 - Add CreateTime and LogAppendTime to Kafka message > KIP-33 - Build a time-based log index > > KIP-33 can be a follow up KIP for KIP-32, so we can discuss about KIP-31 > and KIP-32 first for now. I will create a separate discussion thread for > KIP-32 and reply the concerns you raised regarding the timestamp. > > So far it looks there is no objection to KIP-31. Since I removed a few part > from previous KIP and only left the relative offset proposal, it would be > great if people can take another look to see if there is any concerns. > > Thanks, > > Jiangjie (Becket) Qin > > > On Tue, Sep 8, 2015 at 1:28 PM, Neha Narkhede <n...@confluent.io> wrote: > > > Becket, > > > > Nice write-up. Few thoughts - > > > > I'd split up the discussion for simplicity. Note that you can always > group > > several of these in one patch to reduce the protocol changes people have > to > > deal with.This is just a suggestion, but I think the following split > might > > make it easier to tackle the changes being proposed - > > > > - Relative offsets > > - Introducing the concept of time > > - Time-based indexing (separate the usage of the timestamp field from > > how/whether we want to include a timestamp in the message) > > > > I'm a +1 on relative offsets, we should've done it back when we > introduced > > it. Other than reducing the CPU overhead, this will also reduce the > garbage > > collection overhead on the brokers. > > > > On the timestamp field, I generally agree that we should add a timestamp > to > > a Kafka message but I'm not quite sold on how this KIP suggests the > > timestamp be set. Will avoid repeating the downsides of a broker side > > timestamp mentioned previously in this thread. I think the topic of > > including a timestamp in a Kafka message requires a lot more thought and > > details than what's in this KIP. I'd suggest we make it a separate KIP > that > > includes a list of all the different use cases for the timestamp (beyond > > log retention) including stream processing and discuss tradeoffs of > > including client and broker side timestamps. > > > > Agree with the benefit of time-based indexing, but haven't had a chance > to > > dive into the design details yet. > > > > Thanks, > > Neha > > > > 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...? > > > > > > 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? > > > > > > 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? > > > > > > 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". > > > > > > -Jay > > > > > > > > > On Mon, Sep 7, 2015 at 10:44 AM, Jiangjie Qin > <j...@linkedin.com.invalid > > > > > > wrote: > > > > > > > Jay, > > > > > > > > Thanks for the comments. Yes, there are actually three proposals as > you > > > > pointed out. > > > > > > > > We will have a separate proposal for (1) - version control mechanism. > > We > > > > actually thought about whether we want to separate 2 and 3 internally > > > > before creating the KIP. The reason we put 2 and 3 together is it > will > > > > saves us another cross board wire protocol change. Like you said, we > > have > > > > to migrate all the clients in all languages. To some extent, the > effort > > > to > > > > spend on upgrading the clients can be even bigger than implementing > the > > > new > > > > feature itself. So there are some attractions if we can do 2 and 3 > > > together > > > > instead of separately. Maybe after (1) is done it will be easier to > do > > > > protocol migration. But if we are able to come to an agreement on the > > > > timestamp solution, I would prefer to have it together with relative > > > offset > > > > in the interest of avoiding another wire protocol change (the process > > to > > > > migrate to relative offset is exactly the same as migrate to message > > with > > > > timestamp). > > > > > > > > In terms of timestamp. I completely agree that having client > timestamp > > is > > > > more useful if we can make sure the timestamp is good. But in reality > > > that > > > > can be a really big *IF*. I think the problem is exactly as Ewen > > > mentioned, > > > > if we let the client to set the timestamp, it would be very hard for > > the > > > > broker to utilize it. If broker apply retention policy based on the > > > client > > > > timestamp. One misbehave producer can potentially completely mess up > > the > > > > retention policy on the broker. Although people don't care about > server > > > > side timestamp. People do care a lot when timestamp breaks. Searching > > by > > > > timestamp is a really important use case even though it is not used > as > > > > often as searching by offset. It has significant direct impact on RTO > > > when > > > > there is a cross cluster failover as Todd mentioned. > > > > > > > > The trick using max(lastAppendedTimestamp, currentTimeMillis) is to > > > > guarantee monotonic increase of the timestamp. Many commercial system > > > > actually do something similar to this to solve the time skew. About > > > > changing the time, I am not sure if people use NTP like using a watch > > to > > > > just set it forward/backward by an hour or so. The time adjustment I > > used > > > > to do is typically to adjust something like a minute / week. So for > > each > > > > second, there might be a few mircoseconds slower/faster but should > not > > > > break the clock completely to make sure all the time-based > transactions > > > are > > > > not affected. The one minute change will be done within a week but > not > > > > instantly. > > > > > > > > Personally, I think having client side timestamp will be useful if we > > > don't > > > > need to put the broker and data integrity under risk. If we have to > > > choose > > > > from one of them but not both. I would prefer server side timestamp > > > because > > > > for client side timestamp there is always a plan B which is putting > the > > > > timestamp into payload. > > > > > > > > Another reason I am reluctant to use the client side timestamp is > that > > it > > > > is always dangerous to mix the control plane with data plane. IP did > > this > > > > and it has caused so many different breaches so people are migrating > to > > > > something like MPLS. An example in Kafka is that any client can > > > construct a > > > > LeaderAndIsrRequest/UpdateMetadataRequest/ContorlledShutdownRequest > > (you > > > > name it) and send it to the broker to mess up the entire cluster, > also > > as > > > > we already noticed a busy cluster can respond quite slow to > controller > > > > messages. So it would really be nice if we can avoid giving the power > > to > > > > clients to control the log retention. > > > > > > > > Thanks, > > > > > > > > Jiangjie (Becket) Qin > > > > > > > > > > > > On Sun, Sep 6, 2015 at 9:54 PM, Todd Palino <tpal...@gmail.com> > wrote: > > > > > > > > > So, with regards to why you want to search by timestamp, the > biggest > > > > > problem I've seen is with consumers who want to reset their > > timestamps > > > > to a > > > > > specific point, whether it is to replay a certain amount of > messages, > > > or > > > > to > > > > > rewind to before some problem state existed. This happens more > often > > > than > > > > > anyone would like. > > > > > > > > > > To handle this now we need to constantly export the broker's offset > > for > > > > > every partition to a time-series database and then use external > > > processes > > > > > to query this. I know we're not the only ones doing this. The way > the > > > > > broker handles requests for offsets by timestamp is a little obtuse > > > > > (explain it to anyone without intimate knowledge of the internal > > > workings > > > > > of the broker - every time I do I see this). In addition, as Becket > > > > pointed > > > > > out, it causes problems specifically with retention of messages by > > time > > > > > when you move partitions around. > > > > > > > > > > I'm deliberately avoiding the discussion of what timestamp to use. > I > > > can > > > > > see the argument either way, though I tend to lean towards the idea > > > that > > > > > the broker timestamp is the only viable source of truth in this > > > > situation. > > > > > > > > > > -Todd > > > > > > > > > > > > > > > On Sun, Sep 6, 2015 at 7:08 PM, Ewen Cheslack-Postava < > > > e...@confluent.io > > > > > > > > > > wrote: > > > > > > > > > > > On Sun, Sep 6, 2015 at 4:57 PM, Jay Kreps <j...@confluent.io> > > wrote: > > > > > > > > > > > > > > > > > > > > 2. Nobody cares what time it is on the server. > > > > > > > > > > > > > > > > > > > This is a good way of summarizing the issue I was trying to get > at, > > > > from > > > > > an > > > > > > app's perspective. Of the 3 stated goals of the KIP, #2 (lot > > > retention) > > > > > is > > > > > > reasonably handled by a server-side timestamp. I really just care > > > that > > > > a > > > > > > message is there long enough that I have a chance to process it. > #3 > > > > > > (searching by timestamp) only seems useful if we can guarantee > the > > > > > > server-side timestamp is close enough to the original client-side > > > > > > timestamp, and any mirror maker step seems to break that (even > > > ignoring > > > > > any > > > > > > issues with broker availability). > > > > > > > > > > > > I'm also wondering whether optimizing for search-by-timestamp on > > the > > > > > broker > > > > > > is really something we want to do given that messages aren't > really > > > > > > guaranteed to be ordered by application-level timestamps on the > > > broker. > > > > > Is > > > > > > part of the need for this just due to the current consumer APIs > > being > > > > > > difficult to work with? For example, could you implement this > > pretty > > > > > easily > > > > > > client side just the way you would broker-side? I'd imagine a > > couple > > > of > > > > > > random seeks + reads during very rare occasions (i.e. when the > app > > > > starts > > > > > > up) wouldn't be a problem performance-wise. Or is it also that > you > > > need > > > > > the > > > > > > broker to enforce things like monotonically increasing timestamps > > > since > > > > > you > > > > > > can't do the query properly and efficiently without that > guarantee, > > > and > > > > > > therefore what applications are actually looking for *is* > > broker-side > > > > > > timestamps? > > > > > > > > > > > > -Ewen > > > > > > > > > > > > > > > > > > > > > > > > > Consider cases where data is being copied from a database or > from > > > log > > > > > > > files. In steady-state the server time is very close to the > > client > > > > time > > > > > > if > > > > > > > their clocks are sync'd (see 1) but there will be times of > large > > > > > > divergence > > > > > > > when the copying process is stopped or falls behind. When this > > > occurs > > > > > it > > > > > > is > > > > > > > clear that the time the data arrived on the server is > irrelevant, > > > it > > > > is > > > > > > the > > > > > > > source timestamp that matters. This is the problem you are > trying > > > to > > > > > fix > > > > > > by > > > > > > > retaining the mm timestamp but really the client should always > > set > > > > the > > > > > > time > > > > > > > with the use of server-side time as a fallback. It would be > worth > > > > > talking > > > > > > > to the Samza folks and reading through this blog post ( > > > > > > > > > > > > > > > > > > > > > > > > > > > > http://radar.oreilly.com/2015/08/the-world-beyond-batch-streaming-101.html > > > > > > > ) > > > > > > > on this subject since we went through similar learnings on the > > > stream > > > > > > > processing side. > > > > > > > > > > > > > > I think the implication of these two is that we need a proposal > > > that > > > > > > > handles potentially very out-of-order timestamps in some kind > of > > > > sanish > > > > > > way > > > > > > > (buggy clients will set something totally wrong as the time). > > > > > > > > > > > > > > -Jay > > > > > > > > > > > > > > On Sun, Sep 6, 2015 at 4:22 PM, Jay Kreps <j...@confluent.io> > > > wrote: > > > > > > > > > > > > > > > The magic byte is used to version message format so we'll > need > > to > > > > > make > > > > > > > > sure that check is in place--I actually don't see it in the > > > current > > > > > > > > consumer code which I think is a bug we should fix for the > next > > > > > release > > > > > > > > (filed KAFKA-2523). The purpose of that field is so there is > a > > > > clear > > > > > > > check > > > > > > > > on the format rather than the scrambled scenarios Becket > > > describes. > > > > > > > > > > > > > > > > Also, Becket, I don't think just fixing the java client is > > > > sufficient > > > > > > as > > > > > > > > that would break other clients--i.e. if anyone writes a v1 > > > > messages, > > > > > > even > > > > > > > > by accident, any non-v1-capable consumer will break. I think > we > > > > > > probably > > > > > > > > need a way to have the server ensure a particular message > > format > > > > > either > > > > > > > at > > > > > > > > read or write time. > > > > > > > > > > > > > > > > -Jay > > > > > > > > > > > > > > > > On Thu, Sep 3, 2015 at 3:47 PM, Jiangjie Qin > > > > > <j...@linkedin.com.invalid > > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > >> Hi Guozhang, > > > > > > > >> > > > > > > > >> I checked the code again. Actually CRC check probably won't > > > fail. > > > > > The > > > > > > > >> newly > > > > > > > >> added timestamp field might be treated as keyLength instead, > > so > > > we > > > > > are > > > > > > > >> likely to receive an IllegalArgumentException when try to > read > > > the > > > > > > key. > > > > > > > >> I'll update the KIP. > > > > > > > >> > > > > > > > >> Thanks, > > > > > > > >> > > > > > > > >> Jiangjie (Becket) Qin > > > > > > > >> > > > > > > > >> On Thu, Sep 3, 2015 at 12:48 PM, Jiangjie Qin < > > > j...@linkedin.com> > > > > > > > wrote: > > > > > > > >> > > > > > > > >> > Hi, Guozhang, > > > > > > > >> > > > > > > > > >> > Thanks for reading the KIP. By "old consumer", I meant the > > > > > > > >> > ZookeeperConsumerConnector in trunk now, i.e. without this > > bug > > > > > > fixed. > > > > > > > >> If we > > > > > > > >> > fix the ZookeeperConsumerConnector then it will throw > > > exception > > > > > > > >> complaining > > > > > > > >> > about the unsupported version when it sees message format > > V1. > > > > > What I > > > > > > > was > > > > > > > >> > trying to say is that if we have some > > > ZookeeperConsumerConnector > > > > > > > running > > > > > > > >> > without the fix, the consumer will complain about CRC > > mismatch > > > > > > instead > > > > > > > >> of > > > > > > > >> > unsupported version. > > > > > > > >> > > > > > > > > >> > Thanks, > > > > > > > >> > > > > > > > > >> > Jiangjie (Becket) Qin > > > > > > > >> > > > > > > > > >> > On Thu, Sep 3, 2015 at 12:15 PM, Guozhang Wang < > > > > > wangg...@gmail.com> > > > > > > > >> wrote: > > > > > > > >> > > > > > > > > >> >> Thanks for the write-up Jiangjie. > > > > > > > >> >> > > > > > > > >> >> One comment about migration plan: "For old consumers, if > > they > > > > see > > > > > > the > > > > > > > >> new > > > > > > > >> >> protocol the CRC check will fail".. > > > > > > > >> >> > > > > > > > >> >> Do you mean this bug in the old consumer cannot be fixed > > in a > > > > > > > >> >> backward-compatible way? > > > > > > > >> >> > > > > > > > >> >> Guozhang > > > > > > > >> >> > > > > > > > >> >> > > > > > > > >> >> On Thu, Sep 3, 2015 at 8:35 AM, Jiangjie Qin > > > > > > > <j...@linkedin.com.invalid > > > > > > > >> > > > > > > > > >> >> wrote: > > > > > > > >> >> > > > > > > > >> >> > Hi, > > > > > > > >> >> > > > > > > > > >> >> > We just created KIP-31 to propose a message format > change > > > in > > > > > > Kafka. > > > > > > > >> >> > > > > > > > > >> >> > > > > > > > > >> >> > > > > > > > > >> >> > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-31+-+Message+format+change+proposal > > > > > > > >> >> > > > > > > > > >> >> > As a summary, the motivations are: > > > > > > > >> >> > 1. Avoid server side message re-compression > > > > > > > >> >> > 2. Honor time-based log roll and retention > > > > > > > >> >> > 3. Enable offset search by timestamp at a finer > > > granularity. > > > > > > > >> >> > > > > > > > > >> >> > Feedback and comments are welcome! > > > > > > > >> >> > > > > > > > > >> >> > Thanks, > > > > > > > >> >> > > > > > > > > >> >> > Jiangjie (Becket) Qin > > > > > > > >> >> > > > > > > > > >> >> > > > > > > > >> >> > > > > > > > >> >> > > > > > > > >> >> -- > > > > > > > >> >> -- Guozhang > > > > > > > >> >> > > > > > > > >> > > > > > > > > >> > > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > -- > > > > > > Thanks, > > > > > > Ewen > > > > > > > > > > > > > > > > > > > > > > > > > > -- > > Thanks, > > Neha > > >
