Currently, IQ throws InvalidStateStoreException for any types of error, that means a user cannot handle different types of error.

Because of that, we should throw different exceptions for each type.

Proposed Changes

There append three are add four new exceptions:

Code Block

language	java

public class StateStoreClosedExceptionStateStoreMigratedException extends InvalidStateStoreException
public class StateStoreMigratedExceptionStateStoreRetryableException extends InvalidStateStoreException
public class StateStoreRetryableExceptionStateStoreFailException extends InvalidStateStoreException
public class StateStoreFailExceptionStateStoreClosedException extends InvalidStateStoreException

...

StateStoreRetriableException: The application instance in the state of rebalancing, the user just need retry and wait until rebalance finished.
StateStoreMigratedException: The store got migrated and not hosted in application instance, the users need to rediscover the store.
StateStoreFailException: Fatal error when access state store, the user cannot retry or rediscover.

One internal exception: StateStoreClosedException.

The following is the public method that users will call:

...

Expand

title	Call trace 3: ReadOnlyKeyValueStore#range()

CompositeReadOnlyKeyValueStore#range() (v)
- WrappingStoreProvider#stores() (v)
  - StreamThreadStateStoreProvider#stores() (v)
- return new DelegatingPeekingKeyValueIterator<>()
DelegatingPeekingKeyValueIterator#hasNext() (v)
- CompositKeyValueIterator#hasNext()
  - NextIteratorFunction#apply() (v)
    - MeteredKeyValueBytesStore#range()
      - InnerMeteredKeyValueStore#range()
        CachingKeyValueStore#range()
        AbstractStateStore#validateStoreOpen() (v)
        ChangeLoggingKeyValueBytesStore#range()
        RocksDBStore#range()
        RocksDBStore#validateStoreOpen() (v)
        return new RocksDBRangeIterator()
        return new MergedSortedCacheKeyValueBytesStoreIterator()
        return new MeteredKeyValueIterator()
      - return
    - return
  - return
- CompositKeyValueIterator#next()
  - MeteredKeyValueIterator#next()
    - MergedSortedCacheKeyValueBytesStoreIterator#next()
      AbstractMergedSortedCacheStoreIterator#next()

- - - - RocksDBRangeIterator#hasNext()
        RocksDBIterator#hasNext() (v)
        RocksIterator#isValid()
      - AbstractMergedStortedCacheStoreIterator#nextSrtoreValue()
        RocksDBRangeIterator#next()
        RocksDBIterator#next()
        RocksDbIterator#hasNext() (v)
        RocksDbIterator#getKeyValue()
        RocksIterator#next()
        return keyvalue entry
        return
      - return
    - return outerkeyvalue
  - return
- return
DelegatingPeekingKeyValueIterator#next()
- return keyvalue

Expand

title	Call trace 4: ReadOnlyKeyValueStore#all()

CompositeReadOnlyKeyValueStore#all() (v)
- WrappingStoreProvider#stores() (v)
  - StreamThreadStateStoreProvider#stores() (v)
- return new DelegatingPeekingKeyValueIterator<>()
DelegatingPeekingKeyValueIterator#hasNext() (v)
- CompositKeyValueIterator#hasNext()
  - NextIteratorFunction#apply() (v)
    - MeteredKeyValueBytesStore#all()
      - InnerMeteredKeyValueStore#all()
        CachingKeyValueStore#all()
        AbstractStateStore#validateStoreOpen() (v)
        ChangeLoggingKeyValueBytesStore#all()
        RocksDBStore#all()
        RocksDBStore#validateStoreOpen() (v)
        return new RocksDBIterator()
        return
        return new MergedSortedCacheKeyValueBytesStoreIterator()
        return new MeteredKeyValueIterator()
      - return
    - return
  - return
- CompositKeyValueIterator#next()
  - MeteredKeyValueIterator#next()
    - MergedSortedCacheKeyValueBytesStoreIterator#next()
      AbstractMergedSortedCacheStoreIterator#next()
      - MemoryLRUCacheBytesIterator.hasNext()
      - DelegatingPeekingKeyValueIterator.hasNext() (v)
      - AbstractMergedSortedCacheStoreIterator#nextStoreValue()
        DelegatingPeekingKeyValueIterator#next()
        DelegatingPeekingKeyValueIterator#hasNext() (v)
        return
      - return
    - return outerkeyvalue
  - return
- return
DelegatingPeekingKeyValueIterator#next()
- return keyvalue

Expand

title	Call trace 5: ReadOnlyKeyValueStore#approximateNumEntries()

CompositeReadOnlyKeyValueStore#approximateNumEntries() (v)
- WrappingStoreProvider#stores() (v)
  - StreamThreadStateStoreProvider#stores() (v)
- MeteredKeyValueBytesStore#approximateNumEntries()
  - InnerMeteredKeyValueStore#approximateNumEntries()
    - CachingKeyValueStore#approximateNumEntries()
      - AbstractStateStore#validateStoreOpen() (v)
        RocksDBStore#isOpen()
      - RocksDBStore#approximateNumEntries()
        RocksDBStore#validateStoreOpen() (v)
        return value
      - return
    - return
  - return
- return total

Expand

title	Call trace 6: ReadOnlySessionStore#fetch(key)

CompositeReadOnlySessionStore#fetch(key) (v)
- WrappingStoreProvider#stores() (v)
  - StreamThreadStateStoreProvider#stores() (v)
- MeteredSessionStore#fetch(key)
  - MeteredSessionStore#findSessions()
    - CachingSessionStore#findSessions()
      - AbstractStateStore#validateStoreOpen() (v)
      - ChangeLoggingSessionBytesStore#findSessions()
        RocksDBSessionStore.findSessions(k)
        RocksDBSessionStore.findSessions(from, to)
        RocksDBSegmentedBytesStore#fetch()
        SessionKeySchema#segmentsToSearch()
        Segments#segments() (v)
        Segments#getSegment()
        ConcurrentHashMap#get()
        Segments#isSegment()
        return segment
        retiurn segments
        return
        return new SegmentIterator()
        return new WrappedSessionStoreIterator()
        return
        return
      - return new MergedSortedCacheSessionStoreIterator()
    - return new MeteredWindowedKeyValueIterator()
  - return
- MeteredWindowedKeyValueIterator#hasNext()
  - MergedSortedCacheSessionStoreIterator#hasNext()
    AbstraceMergedSortedCacheStoreIterator#hasNext()
    - FilteredCacheIterator#hasNext()
    - WrappedSessionStoreIterator#hasNext()
      - SegmentIterator#hasNext()
        Segment.range(from, to)
        RocksDBStore.range(from, to)
        RocksDBStore.validateStoreOpen() (v)
        return new RocksDBRangeIterator()
        return
      - return
    - return
  - return iterator(MeteredWindowedKeyValueIterator)
MeteredWindowedKeyValueIterator#next()
- MergedSortedCacheSessionStoreIterator#next()
  AbstractMergedSortedCacheStoreIterator#next()
  - AbstractMergedSortedCacheStoreIterator#hasNext()
  - FilteredCacheIterator#hasNext()
  - WrappedSessionStoreIterator#hasNext()
    - SegmentIterator#hasNext()
      - Segment.range(from, to)
        RocksDBStore.range(from, to)
        RocksDBStore.validateStoreOpen() (v)
        return new RocksDBRangeIterator()
      - return
    - return
  - MergedSortedCacheSessionStoreIterator#nextStoreValue()
    AbstractMergedSortedCacheStoreIterator#nextStoreValue()
    - WrappedSessionStoreIterator#next()
      - SegmentIterator#next()
        RocksDBRangeIterator#next()
        RocksDbIterator#next()
        RocksDbIterator#getKeyValue()
        RocksIterator#next()
        return entry
        return
      - return
    - return
  - return
- return

Expand

title	Call trace 7: ReadOnlySessionStore#fetch(from, to)

CompositeReadOnlySessionStore#fetch(from, to) (v)
- WrappingStoreProvider#stores() (v)
  - StreamThreadStateStoreProvider#stores() (v)
- return new DelegatingPeekingKeyValueIterator<>()
DelegatingPeekingKeyValueIterator#hasNext() (v)
- CompositKeyValueIterator#hasNext()
  - NextIteratorFunction#apply(store)
    - MeteredSessionStore#fetch(from, to)
      - MeteredSessionStore#findSession(from, to, 0, t)
        CachingSessionStore#findSession(from, to, 0, t)
        CachingSessionStore#validStoreOpen()
        AbstractStateStore#validStoreOpen() (v)
        ChangeLoggingSessionBytesStore#isOpen()
        RocksDBSessionStore#isOpen()
        AbstractStateStore#isOpen()
        RocksDBSegmentedBytesStore#isOpen()
        return
        return
        return
        ChangeLoggingSessionBytesStore#findSesisons()
        RocksDBSessionStore#findSessions()
        RocksDBSegmentedBytesStore#fetch()
        SessionKeySchema#segmentsToSearch()
        Segments#segments() (v)
        RocksDBStore#isOpen()
        return new SegmentIterator()
        return new WrappedSessionStoreIterator()
        return
        return
        return new MergedSortedCacheSessionStoreIterator()
        return MeteredWindowedKeyValueIterator()
      - return
    - return
  - MeteredWindowedKeyValueIterator#hasNext()
    - MergedSortedCacheSessionStoreIterator#hasNext()
      AbstractMergedSortedCacheStoreIterator#hasNext()
      - FilteredCacheIterator#hasNext()
      - WrappedSessionStoreIterator#hasNext()
        SegmentIterator#hasNext()
        Segment.range(from, to)
        RocksDBStore.range(from, to)

- - - - RocksDBStore.validateStoreOpen() (v)
        return new RocksDBRangeIterator()
        return
        return
      - return
    - return
  - return
- CompositeKeyValueIterator#next()
- return
DelegatingPeekingKeyValueIterator#next()
- return keyvalue

...

Code Block

language	java
title	QueryableStoreProvider#getStore()	collapse	trueclass QueryableStoreProvider

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);
}

Code Block

language	java
title	GlobalStateStoreProvider#stores()
collapse	true
class GlobalStateStoreProvider

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);
}

Code Block

language	java
title	StreamThreadStateStoreProvider#stores()	collapse	trueclass StreamThreadStateStoreProvider

 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 StateStoreMigratedException
        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;
}

Code Block

language	java
title	WrappingStoreProvider#stores()
collapse	true
class WrappingStoreProvider

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()) {
        // Replace with StateStoreMigratedException
        throw new InvalidStateStoreException("the state store, " + storeName + ", may have migrated to another instance.");
    }
    return allStores;
}

Code Block

language	java
title	AbstractStateStore#validateStoreOpen()
collapse	true
class AbstractStateStore

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

Code Block

language	java
title	RocksDBStore#validateStoreOpen()
collapse	true
class RocksDBStore

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

Code Block

language	java
title	RocksDBIterator#hasNext()
collapse	true
class RocksDBIterator

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

    return iter.isValid();
}

Code Block

language	java
title	DelegatingPeekingKeyValueIterator#hasNext()
collapse	true
class DelegatingPeekingKeyValueIterator

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;
}

Code Block

language	java
title	class Segments

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;
}

Code Block

language	java
title	class SegmentIterator

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;
}

Compatibility, Deprecation, and Migration Plan

All new exceptions extend from InvalidStateStoreException, this change will be fully backward compatible.

Rejected Alternatives

None.

...

Space shortcuts

Child pages

Versions Compared

Old Version 25

New Version 26

Key

Proposed Changes

Compatibility, Deprecation, and Migration Plan

Rejected Alternatives

Space shortcuts

Child pages

Page History

Versions Compared

Old Version 25

New Version 26

Key

Proposed Changes

Compatibility, Deprecation, and Migration Plan

Rejected Alternatives