kafka-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Matthias J. Sax" <matth...@confluent.io>
Subject Re: Correct prefetching of data to KTable-like structure on application startup
Date Mon, 13 Feb 2017 17:37:33 GMT
Jan,

brokers with version 0.10.1 or higher allow to set both topic cleanup
policies in combination:

https://cwiki.apache.org/confluence/display/KAFKA/KIP-71%3A+Enable+log+compaction+and+deletion+to+co-exist

However, this will only delete data in you changelog topic but not in
your RocksDB -- if you want to get data delete in RocksDB, you would
need to send tombstone messages for those keys. It's kinda tricky to get
this done.

An "brute force" alternative would be, stop the application, delete the
local state directory, and restart. This will force Streams to recreate
the RocksDB files from the changelog and thus only loading keys that got
not deleted. But this is of course a quite expensive approach and you
should be very careful about using it.


-Matthias


On 2/13/17 12:25 AM, Jan Lukavský wrote:
> Hi Michael,
> 
> sorry for my late answer. Configuring the topic as you suggest is one
> option (and I will configure it that way), but I wanted to combine the
> two data retention mechanisms (if possible). I would like to use log
> compaction, so that I will always get at least the last message for
> given key, but I would also like to use the classical temporal data
> retention, which would function as a sort of TTL for the keys - if a key
> doesn't get an update for the configured period of time, if could be
> removed. That way I could ensure that out-dated keys could be removed.
> 
> Is there any other option for this? And can kafka be configured this way?
> 
> Best,
> 
>  Jan
> 
> On 02/09/2017 12:08 PM, Michael Noll wrote:
>> Jan,
>>
>>>   - if I don't send any data to a kafka partition for a period longer
>>> then
>> the data retention interval, then all data from the partition is wiped
>> out
>>
>> If I interpret your first and second message in this email thread
>> correctly, then you are talking only about your "state topic" here, i.e.
>> the topic that you read into a KTable.  You should configure this
>> topic to
>> use log compaction, which will ensure that the latest value for a
>> given key
>> will never be wiped.  So even if you don't send any data to a Kafka
>> partition of this (now log-compacted) "state topic" for a long period of
>> time, you'd always have access to (at least) the latest value for
>> every key.
>>
>> Would that help?
>>
>> -Michael
>>
>>
>>
>>
>>
>> On Thu, Feb 9, 2017 at 10:16 AM, Jan Lukavský <je.ik@seznam.cz> wrote:
>>
>>> Hi Matthias,
>>>
>>> first of all, thanks for your answer. Sorry if I didn't explain the
>>> problem well, I didn't want to dig too much into detail to focus on the
>>> important and maybe the result was not clear.
>>>
>>> My fault, I will try to explain in again. I have two KafkaConsumers
>>> in two
>>> separate threads consuming from two topics - let's call the first one
>>> "stream topic" (processed like KStream)
>>>
>>> and the second one "state topic" (processed like KTable). The state
>>> topic
>>> carries a persistent data that I need in order to process the stream
>>> topic,
>>> so I need to cache the state topic
>>>
>>> locally before starting consumption of the stream topic. When the
>>> application is running normally, there seems to be no issue with this,
>>>
>>> because the state topic is updated asynchronously and I use internal
>>> locks
>>> to synchronize the processing inside the application. So far,
>>> everything is
>>> fine.
>>>
>>>
>>> The problem might arise when the application starts - then I do the
>>> following:
>>>
>>>   - lock processing of the stream topic (because I don't have the state
>>> topic cached)
>>>
>>>   - read the current offset N from the state topic (which gives me
>>> offsets
>>> of a message that should be expected next, that is message that has
>>> not yet
>>> been written)
>>>
>>>   - reset offset of the state topic to beginning and read it until I
>>> read
>>> offset N - 1, which tells me that I have cached all the data I need to
>>> process the stream topic, so I unlock the stream processing and continue
>>>
>>> All this works well, except for some very rare situation, when the
>>> following happens (as I understand it, maybe here I am making some
>>> mistake):
>>>
>>>   - for a long period of time there is no update to (at least single
>>> partition) of the state topic
>>>
>>>   - when I try to cache the state topic during startup as explained
>>> above,
>>> it might never finish, because I will never get a message with offset
>>> N - 1
>>> - that is because I will not get any message at all, because all of the
>>> data has been wiped out
>>>
>>>   - because I don't know if I get all the data from the state topic, I
>>> cannot start processing the stream topic and the whole application is
>>> stuck, until first message arrives into all partition of the state topic
>>> (which might even never happen)
>>>
>>>   - I might use some sort of timeout to handle this, but this could be
>>> dangerous, relying on KafkaConsumer.poll() returning empty records
>>> sounds
>>> to me a little fragile too (because this might also indicate that no
>>> records could have been fetched within the timeout, am I right?), what
>>> would totally solve my issue would be that during data retention, the
>>> last
>>> message would always be kept, and therefore I will always get the
>>> message
>>> with offset N - 1, and the whole issue would vanish.
>>>
>>> The situation when a partition on the state topic gets no updates during
>>> long time happens mostly in development environment (where there is
>>> little
>>> to no traffic), but I sense that this could be an issue in production
>>> too,
>>> for example due to some repartitioning of topics.
>>>
>>> Does that make any sense to you now?
>>>
>>> Thanks again for your response,
>>>
>>>   Jan
>>>
>>>
>>>
>>> On 02/09/2017 08:00 AM, Matthias J. Sax wrote:
>>>
>>>> Jan,
>>>>
>>>> you scenario is quite complex and I am not sure if I understood every
>>>> part of it. I try to break it down:
>>>>
>>>> In my scenario on startup, I want to read all data from a topic (or a
>>>>> subset of its partitions),
>>>>> wait until all the old data has been cached and then start
>>>>> processing of
>>>>> a different stream
>>>>>
>>>> That is hard to accomplish in general. Kafka Streams internally uses
>>>> KafkaConsumer (one instance per StreamThread) and thus, does rely on
>>>> the
>>>> consumer's behavior with regard to poll(). Hence, Streams cannot
>>>> control
>>>> in detail, what data will be fetched from the brokers.
>>>>
>>>> Furthermore, Streams follow its own internal strategy to pick a record
>>>> (from the available ones returned from poll()) and you cannot
>>>> control in
>>>> your code (at least not directly) what record will be picked.
>>>>
>>>> Basically, Streams tried to process records in "timestamp order", ie,
>>>> based an the timestamp returned from TimestampExtractor. So you can
>>>> "influence" the processing order by record timestamps (as far as you
>>>> can
>>>> influence them) and/or by providing a custom TimestampExtractor.
>>>>
>>>> In your example, you might want the records you want to process first
>>>> (KTable), to have smaller timestamps (ie, be earlier) than the records
>>>> from your KStream. But even this will only give you "best effort"
>>>> behavior, and it can happen that a KStream record is processed before
>>>> all KTable records to processed. It's a know issues but hard to
>>>> resolve.
>>>>
>>>> when the specific partition doesn't get any message within the
>>>> retention
>>>>> period,
>>>>> then I end up stuck trying to prefetch data to the "KTable" - this is
>>>>> because I get
>>>>> the offset of the last message (plus 1) from the broker, but I
>>>>> don't get
>>>>> any data
>>>>> ever (until I send a message to the partition)
>>>>>
>>>> Cannot follow here: if there is no data, than you can of course not
>>>> process any data -- so why do you end up being stuck?
>>>>
>>>> The problem I see here is that kafka tells me what the last offset in a
>>>>> partition is,
>>>>> but there is no upper bound on when a first message will arrive,
>>>>>
>>>> In general, the latency between data append at the broker and data
>>>> receive at a consumer is rather small. So even if there is strictly no
>>>> upper bound until a message gets delivered, this should not be an issue
>>>> in practice. Or do I miss understand something?
>>>>
>>>> even though I reset the offset and start reading from the beginning
>>>> of a
>>>>> partition.
>>>>>
>>>> How does this relate? Cannot follow.
>>>>
>>>> My question is, is it a possibility not to clear the whole
>>>> partition, but
>>>>> to always keep at least the last message?
>>>>>
>>>> Not with regular retention policy -- not sure if log compaction can
>>>> help
>>>> here.
>>>>
>>>> That way, the client would always get at least the last message, can
>>>>> therefore figure out
>>>>> it is at the end of the partition (reading the old data) and start
>>>>> processing.
>>>>>
>>>> Why is this required? If the client's offset is the same as "endOfLog"
>>>> for each partition, you can figure out that there is nothing to
>>>> read. So
>>>> why would you need the last old message to figure this out?
>>>>
>>>>
>>>> -Matthias
>>>>
>>>>
>>>>
>>>> On 2/7/17 3:46 AM, Jan Lukavský wrote:
>>>>
>>>>> Hi all,
>>>>>
>>>>> I have a question how to do a correct caching in KTable-like structure
>>>>> on application startup. I'm not sure if this belongs to user or dev
>>>>> maillist, so sorry if I've chosen the bad one. What is my
>>>>> observation so
>>>>> far:
>>>>>
>>>>>    - if I don't send any data to a kafka partition for a period longer
>>>>> then the data retention interval, then all data from the partition is
>>>>> wiped out
>>>>>
>>>>>    - the index file is not cleared (which is obvious, it has to
>>>>> keep track
>>>>> of the next offset to assign to a new message)
>>>>>
>>>>> In my scenario on startup, I want to read all data from a topic (or a
>>>>> subset of its partitions), wait until all the old data has been cached
>>>>> and then start processing of a different stream (basically I'm doing
a
>>>>> join of KStream and KTable, but I have implemented it manually due to
>>>>> some special behavior). Now, what is the issue here - when the
>>>>> specific
>>>>> partition doesn't get any message within the retention period, then I
>>>>> end up stuck trying to prefetch data to the "KTable" - this is
>>>>> because I
>>>>> get the offset of the last message (plus 1) from the broker, but I
>>>>> don't
>>>>> get any data ever (until I send a message to the partition). The
>>>>> problem
>>>>> I see here is that kafka tells me what the last offset in a partition
>>>>> is, but there is no upper bound on when a first message will arrive,
>>>>> even though I reset the offset and start reading from the beginning
>>>>> of a
>>>>> partition. My question is, is it a possibility not to clear the whole
>>>>> partition, but to always keep at least the last message? That way, the
>>>>> client would always get at least the last message, can therefore
>>>>> figure
>>>>> out it is at the end of the partition (reading the old data) and start
>>>>> processing. I believe that KTable implementation could have a very
>>>>> similar issue. Or is there any other way around? I could add a
>>>>> timeout,
>>>>> but this seems a little fragile.
>>>>>
>>>>> Thanks in advance for any suggestions and opinions,
>>>>>
>>>>>    Jan
>>>>>
>>>>>
> 


Mime
View raw message