Kafka Streams Issue

[Daan Gertis <dg...@korfinancial.com>]

Hi All,

We are experiencing some weird behaviour with our interactive query service implementation.
This is the flow we’ve implemented:


  1.  kafkaStreams.queryMetadataForKey(store, key, serializer) returns for activeHost HostInfo{host='localhost', port=8562}, and standbyHosts [] for the store and partition where the key would reside. We are not interested in standby hosts. Luckily, we have an active host which we can call.
  2.  We make an HTTP call to host localhost:8562, asking for the key there.
  3.  Inside the 8562 host, we retrieve the store by calling kafkaStreams.store(parameters), using parameters with staleStores set to false.
  4.  We call kafkaStreams.state().equals(RUNNING) to make sure we’re in the RUNNING state.
  5.  Now we call store.get(key) in order to retrieve the key from the store, if it has been stored there.
  6.  The get method on our store implementation calls the storeProvider.stores(storeName, storeType) method to iterate over all the stores available on the host.
  7.  The storeProvider is a WrappingStoreProvider, which calls storeProvider.stores(storeQueryParameters) for each StreamThreadStateStoreProvider it wraps (just one in our case).
  8.  As the logic inside that stores method finds that the StreamThread is in the RUNNING state, it retrieves the tasks based on storeQueryParams.staleStoresEnabled() ? streamThread.allTasks().values() : streamThread.activeTasks(), which evaluates to false since we set staleStores to false in the params.
  9.  To our surprise, the streamThread.activeTasks() method returns an empty ArrayList, while the streamThread.allTasks().values() returns one StandbyTask for the store we’re looking for.
  10. As there appear to be no active tasks on this host for this store, we return the fabled “The state store, " + storeName + ", may have migrated to another instance.” InvalidStateStoreException.

This flow is quite tricky as the queryMetadataForKey returns an active host, which turns out to only have a standby task once queried.
I have executed the queryMetadataForKey method on the active host as well, once before calling kafkaStreams.store in step 3, and another time between step 4 and 5. Each time the metadata returns the same, the host we’re on at that moment is the active host.

Could it be there is a difference between activeHost and activeTask?

For those also on the confluent community slack might recognize this message as it has been posted there by our CTO as well.

Cheers,
D.

Re: Kafka Streams Issue

[John Roesler <vv...@apache.org>]

Hi Daan,

First of all, it does sound like that is a correct
implementation of QueryableStoreProvider. Kudos for taking
that on; the complexity of that API was one of my top
motivations for replacing it with IQv2!
(https://cwiki.apache.org/confluence/display/KAFKA/KIP-796%3A+Interactive+Query+v2
)

To answer your question directly, no "activeHost" just means
the host that currently has the "activeTask" for the desired
store.

I suspect that this is either a subtle and rare edge case in
how the metadata gets updated, or it's just a simple race
condition between the query and a rebalance in the cluster,
which is a fact of life in any distributed database.

If you are able to reproduce it and send us the logs, we
should be able to tell which is which.

In particular, we'd need to see thee things in the logs:
1. The logs for the rebalances and assignments (which are on
by default)
2. The log of when you check the metadata and what the
result it
3. The log of when the query tries to run on the
"activeHost" and what it finds there (that the task is only
a standby)

One other possibility worth considering is whether the
queryMetadataForKey is producing the correct partition. What
it does is run the provided key through the provided
serializer and then run the serialized key though the
default partitioner. If your actual data isn't partitioned
the same way, then queryMetadataForKey might be effectively
selecting a random host, which sometimes happens to host the
active task and other times does not? Kind of a long shot,
but I just wanted to put it out there.

Thanks,
-John


On Mon, 2022-03-28 at 13:48 +0000, Daan Gertis wrote:
> Hi All,
> 
> We are experiencing some weird behaviour with our interactive query service implementation.
> This is the flow we’ve implemented:
> 
> 
>   1.  kafkaStreams.queryMetadataForKey(store, key, serializer) returns for activeHost HostInfo{host='localhost', port=8562}, and standbyHosts [] for the store and partition where the key would reside. We are not interested in standby hosts. Luckily, we have an active host which we can call.
>   2.  We make an HTTP call to host localhost:8562, asking for the key there.
>   3.  Inside the 8562 host, we retrieve the store by calling kafkaStreams.store(parameters), using parameters with staleStores set to false.
>   4.  We call kafkaStreams.state().equals(RUNNING) to make sure we’re in the RUNNING state.
>   5.  Now we call store.get(key) in order to retrieve the key from the store, if it has been stored there.
>   6.  The get method on our store implementation calls the storeProvider.stores(storeName, storeType) method to iterate over all the stores available on the host.
>   7.  The storeProvider is a WrappingStoreProvider, which calls storeProvider.stores(storeQueryParameters) for each StreamThreadStateStoreProvider it wraps (just one in our case).
>   8.  As the logic inside that stores method finds that the StreamThread is in the RUNNING state, it retrieves the tasks based on storeQueryParams.staleStoresEnabled() ? streamThread.allTasks().values() : streamThread.activeTasks(), which evaluates to false since we set staleStores to false in the params.
>   9.  To our surprise, the streamThread.activeTasks() method returns an empty ArrayList, while the streamThread.allTasks().values() returns one StandbyTask for the store we’re looking for.
>   10. As there appear to be no active tasks on this host for this store, we return the fabled “The state store, " + storeName + ", may have migrated to another instance.” InvalidStateStoreException.
> 
> This flow is quite tricky as the queryMetadataForKey returns an active host, which turns out to only have a standby task once queried.
> I have executed the queryMetadataForKey method on the active host as well, once before calling kafkaStreams.store in step 3, and another time between step 4 and 5. Each time the metadata returns the same, the host we’re on at that moment is the active host.
> 
> Could it be there is a difference between activeHost and activeTask?
> 
> For those also on the confluent community slack might recognize this message as it has been posted there by our CTO as well.
> 
> Cheers,
> D.