public class StateStoreMigratedException extends InvalidStateStoreException
public class StateStoreRetryableException extends InvalidStateStoreException
public class StateStoreFailException extends InvalidStateStoreException
public class StateStoreClosedException extends InvalidStateStoreException

public class KafkaStreams {
    public <T> T store(final String storeName, final QueryableStoreType<T> queryableStoreType);
}
 
public class CompositeReadOnlyKeyValueStore<K, V> implements ReadOnlyKeyValueStore<K, V> {
    public V get(final K key);
    public KeyValueIterator<K, V> range(final K from, final K to);
    public KeyValueIterator<K, V> all();
    public long approximateNumEntries();
}
 
public class CompositeReadOnlySessionStore<K, V> implements ReadOnlySessionStore<K, V> {
    public KeyValueIterator<Windowed<K>, V> fetch(final K key);
    public KeyValueIterator<Windowed<K>, V> fetch(final K from, final K to);
}
 
public class CompositeReadOnlyWindowStore<K, V> implements ReadOnlyWindowStore<K, V> {
    public WindowStoreIterator<V> fetch(final K key, final long timeFrom, final long timeTo);
    public KeyValueIterator<Windowed<K>, V> fetch(final K from, final K to, final long timeFrom, final long timeTo);
    public KeyValueIterator<Windowed<K>, V> all();
    public KeyValueIterator<Windowed<K>, V> fetchAll(final long timeFrom, final long timeTo);
}
 
class DelegatingPeekingKeyValueIterator<K, V> implements KeyValueIterator<K, V>, PeekingKeyValueIterator<K, V> {
    public synchronized boolean hasNext();
    public synchronized KeyValue<K, V> next();
    public KeyValue<K, V> peekNext();
}

class MeteredWindowStoreIterator<V> implements WindowStoreIterator<V> {
    public boolean hasNext();
    public KeyValue<Long, V> next();
    public Long peekNextKey()
}

public interface QueryableStoreType<T> {
    // TODO: pass stream instance parameter
    T create(final KafkaStreams streams, final StateStoreProvider storeProvider, final String storeName);
}

public <T> T getStore(final String storeName, final QueryableStoreType<T> queryableStoreType) {
    final List<T> globalStore = globalStoreProvider.stores(storeName, queryableStoreType);
    if (!globalStore.isEmpty()) {
        return queryableStoreType.create(new WrappingStoreProvider(Collections.<StateStoreProvider>singletonList(globalStoreProvider)), storeName);
    }
    final List<T> allStores = new ArrayList<>();
    for (StateStoreProvider storeProvider : storeProviders) {
        allStores.addAll(storeProvider.stores(storeName, queryableStoreType));
    }
    if (allStores.isEmpty()) {
        // TODO: Replace with StateStoreMigratedException
        throw new InvalidStateStoreException("The state store, " + storeName + ", may have migrated to another instance.");
    }
    return queryableStoreType.create(
            new WrappingStoreProvider(storeProviders),
            storeName);
}

public <T> List<T> stores(final String storeName, final QueryableStoreType<T> queryableStoreType) {
    final StateStore store = globalStateStores.get(storeName);
    if (store == null || !queryableStoreType.accepts(store)) {
        return Collections.emptyList();
    }
    if (!store.isOpen()) {
        // TODO: Replace with StateStoreClosedException
        throw new InvalidStateStoreException("the state store, " + storeName + ", is not open.");
    }
    return (List<T>) Collections.singletonList(store);
}

 public <T> List<T> stores(final String storeName, final QueryableStoreType<T> queryableStoreType) {
    if (streamThread.state() == StreamThread.State.DEAD) {
        return Collections.emptyList();
    }
    if (!streamThread.isRunningAndNotRebalancing()) {
        // TODO: Replace with StateStoreRetryableException
        throw new InvalidStateStoreException("Cannot get state store " + storeName + " because the stream thread is " +
                streamThread.state() + ", not RUNNING");
    }
    final List<T> stores = new ArrayList<>();
    for (Task streamTask : streamThread.tasks().values()) {
        final StateStore store = streamTask.getStore(storeName);
        if (store != null && queryableStoreType.accepts(store)) {
            if (!store.isOpen()) {
                // TODO: Replace with StateStoreClosedException
                throw new InvalidStateStoreException("Cannot get state store " + storeName + " for task " + streamTask +
                        " because the store is not open. The state store may have migrated to another instances.");
            }
            stores.add((T) store);
        }
    }
    return stores;
}

public <T> List<T> stores(final String storeName, QueryableStoreType<T> type) {
    final List<T> allStores = new ArrayList<>();
    for (StateStoreProvider provider : storeProviders) {
        final List<T> stores =
            provider.stores(storeName, type);
        allStores.addAll(stores);
    }
    if (allStores.isEmpty()) {
        // TODO: Replace with StateStoreMigratedException
        throw new InvalidStateStoreException("The state store, " + storeName + ", may have migrated to another instance.");
    }
    return allStores;
}

void validateStoreOpen() {
    if (!innerState.isOpen()) {
        // TODO: Replace with StateStoreClosedException
        throw new InvalidStateStoreException("Store " + innerState.name() + " is currently closed.");
    }
}

private void validateStoreOpen() {
    if (!open) {
        // TODO: Replace with StateStoreClosedException
        throw new InvalidStateStoreException("Store " + this.name + " is currently closed");
    }
}

public synchronized boolean hasNext() {
    if (!open) {
        // TODO: Replace with StateStoreClosedException
        throw new InvalidStateStoreException(String.format("RocksDB store %s has closed", storeName));
    }

    return iter.isValid();
}

public synchronized boolean hasNext() {
    if (!open) {
        // TODO: Replace with StateStoreClosedException
        throw new InvalidStateStoreException(String.format("Store %s has closed", storeName));
    }
    if (next != null) {
        return true;
    }

    if (!underlying.hasNext()) {
        return false;
    }

    next = underlying.next();
    return true;
}

List<Segment> segments(final long timeFrom, final long timeTo) {
    final long segFrom = Math.max(minSegmentId, segmentId(Math.max(0L, timeFrom)));
    final long segTo = Math.min(maxSegmentId, segmentId(Math.min(maxSegmentId * segmentInterval, Math.max(0, timeTo))));

    final List<Segment> segments = new ArrayList<>();
    for (long segmentId = segFrom; segmentId <= segTo; segmentId++) {
        Segment segment = getSegment(segmentId);
        if (segment != null && segment.isOpen()) {
            try {
                segments.add(segment);
            } catch (InvalidStateStoreException ise) {  // TODO: Replace with StateStoreClosedException
                // segment may have been closed by streams thread;
            }
        }
    }
    return segments;
}

List<Segment> allSegments() {
    final List<Segment> segments = new ArrayList<>();
    for (Segment segment : this.segments.values()) {
        if (segment.isOpen()) {
            try {
                segments.add(segment);
            } catch (InvalidStateStoreException ise) {  // TODO: Replace with StateStoreClosedException
                // segment may have been closed by streams thread;
            }
        }
    }
    Collections.sort(segments);
    return segments;
}

public boolean hasNext() {
    boolean hasNext = false;
    while ((currentIterator == null || !(hasNext = hasNextCondition.hasNext(currentIterator)) || !currentSegment.isOpen())
            && segments.hasNext()) {
        close();
        currentSegment = segments.next();
        try {
            if (from == null || to == null) {
                currentIterator = currentSegment.all();
            } else {
                currentIterator = currentSegment.range(from, to);
            }
        } catch (InvalidStateStoreException e) {  // TODO: Replace with StateStoreClosedException
            // segment may have been closed so we ignore it.
        }
    }
    return currentIterator != null && hasNext;
}