[jira] [Created] (KAFKA-7158) Duplicates when searching kafka stream state store with caching

["Christian Henry (JIRA)" <ji...@apache.org>]

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:
{noformat}
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
and 
-- 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:
{code:java}
IntStream.rangeClosed(1, 100).forEach(i -> store.put("key" + i, "value" + i));
{code}
then calling
{code:java}
store.fetchAll(start, end)
{code}
would return an iterator with MORE than 100 items, whereas if you explicitly did
{code:java}
store.fetch("key" + i)
{code}
for i = 1 to 100, each fetch would only return a single item in the iterator. 



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)