Re: Correct prefetching of data to KTable-like structure on application startup

[Jan Lukavský]

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