Jan, If I understand you problem correctly, you do something like this on startup (I simplify to single partition)
endOffset = consumer.endOffset(...)
while (!done) {
for (ConsumerRecord r : consumer.poll()) {
// do some processing
if (r.offset == endOffset) {
done = true;
break;
}
}
}
If your partitions is empty, poll() never returns anything and thus you
loop forever.
However, to solve this problem, you can simple check the "start offset"
of the partitions before the loop. If start and end offset are the same,
the partitions is empty and you never call poll.
startOffset = consumer.beginningOffset(...)
endOffset = consumer.endOffset(...)
if(startOffset < endOffset) {
while (!done) {
for (ConsumerRecord r : consumer.poll()) {
// do some processing
if (r.offset == endOffset) {
done = true;
break;
}
}
}
}
Does this help?
-Matthias
On 2/14/17 12:31 AM, Jan Lukavský wrote:
> Hi Matthias,
>
> I understand that the local cache will not be automatically cleared, but
> that is not an issue for me now.
>
> The problem I see is still the same as at the beginning - even caching
> data to RocksDB in KafkaStreams implementation might (I would say)
> suffer from this issue. When using time based data retention (for
> whatever reason, maybe in combination with the log compation, but the
> the issue is there irrespective to whether the log compation is used or
> not), it is possible that some partition will report nonzero "next"
> offset, but will not be able to deliver any message to the KafkaConsumer
> (because the partition is emptied by the data retention) and therefore
> the consumer will not be able to finish the materialization of the topic
> to local store (either RocksDB or any other cache) and therefore will
> not be able to start processing the KStream. If I understand the problem
> right, then using timestamp will not help either, because there must be
> some sort of vector clock with a time dimension for each input
> partition, and the empty partition will not be able to move the
> timestamp any further, and therefore the whole system will remain
> blocked at timestamp 0, because the vector clock usually calculates
> minimum from all time dimensions.
>
> Does that make any sense, or I am doing something fundamentally wrong? :)
>
> Thanks again for any thoughts,
>
> Jan
>
>
> On 02/13/2017 06:37 PM, Matthias J. Sax wrote:
>> 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...@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
>>>>>>>
>>>>>>>
>
signature.asc
Description: OpenPGP digital signature
