Versions Compared

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

JIRA:  
Jira
serverASF JIRA
columnskey,summary,type,created,updated,due,assignee,reporter,priority,status,resolution
serverId5aa69414-a9e9-3523-82ec-879b028fb15b
keyKAFKA-9445

Discussion:

...

  https://www.mail-archive.com/dev@kafka.apache.org/msg104287.html

Motivation:

Whenever a call is made to get a particular key from a Kafka Streams instance, currently it returns a Queryable store wrapper that contains a list of the stores for all the running and restoring/replica(with KIP-535: Allow state stores to serve stale reads during rebalance) on the instance via StreamThreadStateStoreProvider#stores(). This list of stores is then provided to CompositeReadOnlyKeyValueStore#get() which looks into each store one by one. With the changes that went in as a part of KIP-535 since we have access to the information that a key belongs to which partition, we should have the capability to fetch store for that particular partition and look for the key in store for that partition only. It would be a good improvement for improving latencies for applications that contain multiple partitions on a single instance and don't have bloom filters enabled internally for RocksDB.

When serving queries (like `get(key)` or `range(from, to)`), the wrapper actually iterates over all the underlying stores and issues the same query on each one. This is quite inefficient, and more importantly, it disallows some capabilities that KIP-535 intended to provide.

KIP-535 introduced two discovery mechanism so that users could implement a query routing layer, the ability to find out the partition for a specific key, and the ability to find out the locations and freshness of each replica of each partition of a store. Further, it introduced one key mechanism of a resilient query fetch layer, the ability to serve queries from hot-standby replicas and not just running active ones.

What is implicit is that the query routing layer would select an instance from which to fetch each partition of a store that the query spans, and then fan out to execute sub-queries against each such partition on the selected instances. However, the current store() API disallows this last step. Callers are only able to query all partitions on the local instance, not one specific partition.

Here's an example of how this is a drawback:

Imagine we have a cluster with two instances (A and B), and a store S with two partitions (0 and 1). Imagine further that store S has one active and one standby replica configured. Say, instance A hosts (0-active and 1-standby) and instance B hosts (1-active and 0-standby). Now, suppose the query routing layer wants to query the standby replica (so as not to compete with active processing). This arrangement is currently impossible. What would happen instead is that both instance A and B would return results from both partition 0 and 1, and the query router would have to de-duplicate the results. Plus, it would not achieve the objective to avoid competing with active processing.

To fill this gap, this KIP proposes to allow querying a specific partition of a store, while still preserving the ability to query all local partitions. This would also reduce latencies while querying a particular key from an instance, as it will fetch the key only from the specific store partition where it belongs which would be very helpful in instances containing multiple partitionsSo, while adding the new functionality to query a particular partition of a store, it makes sense to combine all the input parameters to KafkaStreams#store() under a new public class StoreQueryParams for ease of use. This requires changes in the public API, namely KafkaStreams.

Public Interfaces:

  • Adding new Class StoreQueryParams.java class StoreQueryParameters to provide user options to the QueryableStoreProvider layer  layer to understand what kind of stores a user wants. It would currently include whether a user is okay with serving stale data and if user already knows what is the partition of the store a user is looking at. Since store name and partition would be a unique combination, a taskId can be generated from this information to return the store for that particular task.

...

Code Block
languagejava
titleStoreQueryParams.java
collapsetrue
StoreQueryParameters.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.kafka.streams;

import org.apache.kafka.streams.state.QueryableStoreType;

import java.util.Objects;

/**
 *// Represents all the query options that a user can provide to state what kind of stores it is expecting. The options would be whether a user would want to enable/disable stale stores* or whether it knows the list of partitions that it specifically wants to fetch. If this information is not provided the default behavior is to fetch the stores for all the partitions available on that instance* for that particular store name.
 * It contains a partition, which for a point queries can be populated from the  KeyQueryMetadata.
 */
public class StoreQueryParams<T>StoreQueryParameters<T> {

    privatepublic Integerstatic partition;
<T>    private boolean includeStaleStores;
    private final String storeName;
    private final QueryableStoreType<T> queryableStoreType;

    public StoreQueryParamsStoreQueryParameters<T> fromNameAndType(final String storeName, final QueryableStoreType<T>  queryableStoreType) {
        this.storeName = storeName;
        this.queryableStoreType = queryableStoreType;
    }

    /**
     * Get the {@link StoreQueryParams} with stale(standby, restoring) stores added via fetching the stores. 
     *
     * @param partition   The specific integer partition to be fetched from the stores list by using {@link StoreQueryParams}.
     * 
     * @return String storeName
     */
    public StoreQueryParams<T>public StoreQueryParameters<T> withPartition(final Integer partition) {
        this.partition = partition;
        return this;
    }

    /**
     * Get the {@link StoreQueryParams} with stale(standby, restoring) stores added via fetching the stores. 
     *                      
     * @return String storeName
     */
    public StoreQueryParams<T> withIncludeStaleStores() {
        this.includeStaleStores = true;
        return this;
    }

    /**
     * Get the store name for which key is queried by the user.
     *
     * @return String storeName
     */
    public String getStoreName() {
        return storeName;
    }

    /**
     * Get the queryable store type for which key is queried by the user.
     *
     * @return QueryableStoreType queryableStoreType
     */
    public QueryableStoreType<T> getQueryableStoreType() {
        return queryableStoreType;
    }StoreQueryParameters<T> enableStaleStores();

    public Integer partition();

    @Override
    public boolean equals(final Object obj) {
        if (!(obj instanceof StoreQueryParams)) {
            return false;
        }staleStoresEnabled();

    public    final StoreQueryParams storeQueryParams = (StoreQueryParams) obj;
        return Objects.equals(storeQueryParams.partition, partition)
                && Objects.equals(storeQueryParams.includeStaleStores, includeStaleStores)
                && Objects.equals(storeQueryParams.storeName, storeName)
                && Objects.equals(storeQueryParams.queryableStoreType, queryableStoreType);
    }

    @OverrideString storeName();

    public String toString() {
        return "StoreQueryParams {" +
                "partition=" + partition +
                ", includeStaleStores=" + includeStaleStores +
                ", storeName=" + storeName +
                ", queryableStoreType=" + queryableStoreType +
                '}';
    }

    @Override
    public int hashCode() {
        return Objects.hash(partition, includeStaleStores, storeName, queryableStoreType);
    }
}
QueryableStoreType<T> queryableStoreType();
}



  • Changing Deprecating the KafkaStreams#store(final String storeName, final QueryableStoreType<T> queryableStoreType, final boolean includeStaleStoresstaleStores) in  in favour of the funtion function mentioned below as this one hasn't been released yet.


Code Block
languagejava
titleKafkaStreams.java
collapsetrue
public class KafkaStreams   {

    /**@Deprecated
  public <T> T *store(final Get a facade wrapping the local {@link StateStore} instances with the provided {@link StoreQueryParams}.
     * StoreQueryParams need required parameters to be set, which are {@code storeName} and if
     * type is accepted by the provided {@link QueryableStoreType#accepts(StateStore) queryableStoreType}.
     * The optional parameters to the StoreQueryParams include {@code partition} and {@code includeStaleStores}.
     * The returned object can be used to query the {@link StateStore} instances.
     *
     * @param storeQueryParams    If storeQueryParams.withPartition(int partition) is used, it allow queries on the specific partition irrespective if it is a standby
     *                            or a restoring replicas in addition to active ones.
     *                            If storeQueryParams.withIncludeStaleStores() is used, it allow queries on standbys and restoring replicas in addition to active ones for all the local partitions on the instance.
     *                            If StoreQueryParams.withIncludeStaleStores().withPartition(int partition), it allow queries on the specific partition irrespective if it is a standby
     *                            or a restoring replicas in addition to active ones..
     *                            By default, if just storeQueryParams is used, it returns all the local partitions for the store which are in running state.
     * @param <T>                 return type
     * @return A facade wrapping the local {@link StateStore} instances
     * @throws InvalidStateStoreException if Kafka Streams is (re-)initializing or a store with {@code storeName} and
     * {@code queryableStoreType} doesn't exist which are provided in {@code storeQueryParams}
     */
    String storeName, final QueryableStoreType<T> queryableStoreType);

  // remove (was added via KIP-535 and was never released)
  public <T> T store(final String storeName, final QueryableStoreType<T> queryableStoreType, final boolean staleStores);

  // newly added 
  public <T> T store(final StoreQueryParams<T> storeQueryParams) {
        validateIsRunningOrRebalancing();
        return queryableStoreProvider.getStore(storeQueryParamsStoreQueryParameters<T> storeQueryParameters);
    }


Proposed Changes:

  • Add a new public class StoreQueryParams.java to class StoreQueryParameters to set options for what kind of stores a user wants.
  • Create a taskId from the combination of store name and partition provided by the user.
  • In StreamThreadStateStoreProvider.java return   return only the stores for the task requested by the user and also check the condition to return only running stores or standby/recovering stores as well.

...

  • KafkaStreams#store(final String storeName, final QueryableStoreType<T> queryableStoreType, final boolean includeStaleStores) will be changed to the one mentioned in the Public Interfaces changes. Since the mentioned function is not released yet in any version, no deprecation is required.
  • Deprecating store(final String storeName, final QueryableStoreType<T> queryableStoreType) method in favour of  public <T> T store(final StoreQueryParameters<T> storeQueryParameters) as both store name and queryableStoreType have been added to StoreQueryParameters. 

Rejected Alternatives:

  • Overload the QueryableStoreProvider#getStore() and StreamThreadStateStoreProvider#stores() with new parameters to pass a list of partitions along with the currently passed flag includeStaleStores.