kafka-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Christian Henry (JIRA)" <j...@apache.org>
Subject [jira] [Created] (KAFKA-7158) Duplicates when searching kafka stream state store with caching
Date Thu, 12 Jul 2018 17:42:00 GMT
Christian Henry created KAFKA-7158:

             Summary: Duplicates when searching kafka stream state store with caching
                 Key: KAFKA-7158
                 URL: https://issues.apache.org/jira/browse/KAFKA-7158
             Project: Kafka
          Issue Type: Bug
          Components: streams
    Affects Versions: 1.1.0
            Reporter: Christian Henry

See mailing list email with same name for initial discussion, reposting my initial email here
for convenience:
We have a kafka stream application, and one of our transform steps keeps a state store to
filter out messages with a previously seen GUID. That is, our transform looks like:
public KeyValue<byte[], String> transform(byte[] key, String guid) {
    try (WindowStoreIterator<DuplicateMessageMetadata> iterator = duplicateStore.fetch(correlationId,
start, now)) {
        if (iterator.hasNext()) {
            return null;
        } else {
            duplicateStore.put(correlationId, some metadata);
            return new KeyValue<>(key, message);
where the duplicateStore is a persistent windowed store with caching enabled. 

I was debugging some tests and found that sometimes when calling all() or fetchAll() on
the duplicate store and stepping through the iterator, it would return the same guid more
than once, even if it was only inserted into the store once. More specifically, if I had the
following guids sent to the stream: [11111, 22222, ... 99999] (for 9 values total), sometimes
it would return 10 values, with one (or more) of the values being returned twice by the iterator.
However, this would not show up with a fetch(guid) on that specific guid. For instance,
if 11111 was being returned twice by fetchAll(), calling duplicateStore.fetch("11111", start,
end) will still return an iterator with size of 1. 

I dug into this a bit more by setting a breakpoint in SegmentedCacheFunction#compareSegmentedKeys(cacheKey,
storeKey) and watching the two input values as I looped through the iterator using "while(iterator.hasNext())
{ print(iterator.next()) }". In one test, the duplicate value was 66666, and saw the following
behavior (trimming off the segment values from the byte input):
-- compareSegmentedKeys(cacheKey = 66666, storeKey = 22222)
-- next() returns 66666
-- compareSegmentedKeys(cacheKey = 77777, storeKey = 66666)
-- next() returns 66666
Besides those, the input values are the same and the output is as expected. Additionally,
a coworker noted that the number of duplicates always matches the number of times Long.compare(cacheSegmentId,
storeSegmentId) returns a non-zero value, indicating that duplicates are likely arising due
to the segment comparison. {noformat}

Basically, what we're seeing is that if you have a persistent store with caching enabled,
you will sometimes get duplicate keys when querying for all keys (using all() or fetchAll())
even though fetch(key) will only return one result. That is, if you had a fresh store with
nothing in it and did something like:
IntStream.rangeClosed(1, 100).forEach(i -> store.put("key" + i, "value" + i));
then calling
store.fetchAll(start, end)
would return an iterator with MORE than 100 items, whereas if you explicitly did
store.fetch("key" + i)
for i = 1 to 100, each fetch would only return a single item in the iterator. 

This message was sent by Atlassian JIRA

View raw message