KAFKA-6878 NPE when querying global state store not in READY state

[tedyu]

Check whether cache is null before retrieving from cache.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[mjsax]

Call for review @bbejeck @vvcephei

[guozhangwang]

Should we consider two other CachingWindowed and CachingSessioned store classes as well?

A meta comment: instead of checking on cache object not null, should we consider including the init function to be protected by the write lock as well (@tedyu you improved on the locking mechanism here so I'd like to have your feedback)

[tedyu]

I can apply similar change to CachingWindowStore and CachingSessionStore if people think the check is beneficial.

w.r.t. additional locking, I tend to leave that to future JIRA. Additional locking increases the complexity of the code.
Also, with current change, client code can detect the completion of initialization, making the additional locking unnecessary (at least for now).

[tedyu]

With additional locking, client query may take arbitrarily long time.
This is change in behavior which I hesitate in introducing.

[bbejeck]

Java 8 error caused by out of memory.

retest this please.

[bbejeck]

Thanks @tedyu, overall LGTM
+1 for adding CachingWindowStore and CachingSessionStore
Also, can we add a test to verify behavior?

[tedyu]

Looking at CachingKeyValueStoreTest, store.init is called in setUp method.

[guozhangwang]

Also, with current change, client code can detect the completion of initialization, making the additional locking unnecessary (at least for now).

I'm a bit unclear how the client can detect if the initialization is completed? If get returns null it could mean 1) init has done, but the key / window / session does not exist, 2) init is not done. How to distinguish these two cases?

[guozhangwang]

In CachingSessionStore#findSessions, we could also check if cache is null.

[tedyu]

Looking at CachingSessionStore#findSessions, an Iterator is returned.

        final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(cacheName, cacheKeyFrom, cacheKeyTo);

When cache is null, should a special MemoryLRUCacheBytesIterator be returned ?

[tedyu]

Updated change in CachingKeyValueStore so that we rely on underlying for query when cache is null.

[tedyu]

With current change, client is not concerned with completion of initialization.
If initialization hasn't completed, the query would take a bit longer.

[guozhangwang]

Thanks @tedyu , the change lgtm. Merging to trunk.

[guozhangwang]

Also cherry-picked to 1.1

@tedyu I realized that in CachingWindowStore there are a couple of other fetch functions that need this safety check as well:

fetch(final Bytes key, final long timeFrom, final long timeTo)

fetch(final Bytes from, final Bytes to, final long timeFrom, final long timeTo)

If the cache is not available, we could skip returning a wrapper iterator that merges the two iterators from underlying and the cache, but just return the underlying iterator directly. Could you submit another PR for this?

[mjsax]

Does it make sense to delegate the query to the underlying store? The issue about the non-initialed cache is, that the store itself was not initialized yet. Thus, the underlying store is not initialized either.

From my point of view, as long as as a store is not initialized and fully recovered, it should not be possible to query the store. We have some check in place that validate that stores are opened but we are hitting a race condition for this case. Overall, I think that we should throw an InvalidStateStoreException -- or something more precise (cf. https://cwiki.apache.org/confluence/display/KAFKA/KIP-216%3A+IQ+should+throw+different+exceptions+for+different+errors)

[tedyu]

Since underlying is passed to CachingKeyValueStore ctor, it is usable before initInternal finishes.

[guozhangwang]

The ordering of initialization is the following:

1. underlying.init where underlying.isOpen is set to true;
2. assign cache

The ordering of get is the following:

3. check underlying.isOpen;
4. try to access cache.

So I think the race condition that @mjsax was talking about is around 1 -> 3 -> 2 -> 4.

I think to fix this issue, in init we could consider switching the steps of 1 and 2:

initInternal(context);
underlying.init(context, root);

since

volatile boolean open = false;

it should be sufficient. In this case the check on step 3) will fail if underlying.init is not completed and we will throw InvalidStateStoreException.

[tedyu]

Opened #4988