KAFKA-20307: Add Interactive Queries (IQv1) support for headers-aware stores#21744
KAFKA-20307: Add Interactive Queries (IQv1) support for headers-aware stores#21744aliehsaeedii wants to merge 3 commits intoapache:trunkfrom
Conversation
aliehsaeedii
force-pushed
the
iqv1_updates
branch
from
1ac1b6d to
e2b2fd9
Compare
mjsax
added
ci-approved
kip
mjsax
left a comment
mjsax
left a comment
There was a problem hiding this comment.
Overall LGTM -- basically only comments about tests
|
|
||
| static RecordConverter converterForStore(final StateStore store) { | ||
| // First check if the top-level store implements HeadersBytesStore or TimestampedBytesStore | ||
| // This handles in-memory stores with marker wrappers |
There was a problem hiding this comment.
It also handler persistent stores, if there are not adaptors?
| if (queryableStoreType instanceof QueryableStoreTypes.KeyValueStoreType) { | ||
| return (List<T>) Collections.singletonList(new GenericReadOnlyKeyValueStoreFacade<>((TimestampedKeyValueStoreWithHeaders<Object, Object>) store, ValueConverters.extractValueFromHeaders())); | ||
| } else if (queryableStoreType instanceof QueryableStoreTypes.TimestampedKeyValueStoreType) { | ||
| return (List<T>) Collections.singletonList(new GenericReadOnlyKeyValueStoreFacade<>((TimestampedKeyValueStoreWithHeaders<Object, Object>) store, ValueConverters.headersToValueAndTimestamp())); |
There was a problem hiding this comment.
nit: might be good to rename from headersToValueAndTimestamp to extractValueAndTimestampFromHeaders to align naming to extractValueFromHeaders ?
| @Test | ||
| @SuppressWarnings("unchecked") | ||
| public void shouldReturnTimestampedConverterForTimestampedToHeadersStoreAdapter() { | ||
| final WrappedStateStore<?, ?, ?> mockWrapper = mock(WrappedStateStore.class); |
There was a problem hiding this comment.
Why are we using mocks here, but not above? Could we simplify the tests above using mocks, too?
|
|
||
| @Test | ||
| public void shouldReturnIdentityConverterForPlainKeyValueStore() { | ||
| final StateStore store = Stores.keyValueStoreBuilder( |
There was a problem hiding this comment.
Same. Should we just use a mock instead?
There was a problem hiding this comment.
Actually, this test seems to be duplicated vi a shouldReturnIdentityConverterForMockKeyValueStore below? Both test the KeyValueStore case?
| } | ||
|
|
||
| @Test | ||
| public void shouldReturnHeadersConverterForHeadersAwareWindowStore() { |
There was a problem hiding this comment.
Should we add a test for session-header store?
| assertEquals(1, stores.size()); | ||
| for (final ReadOnlyWindowStore<String, String> store : stores) { | ||
| assertThat(store, instanceOf(ReadOnlyWindowStore.class)); | ||
| assertThat(store, instanceOf(GenericReadOnlyWindowStoreFacade.class)); |
| } | ||
|
|
||
| @Test | ||
| public void shouldFindTimestampedKeyValueStoresWithHeaders() { |
There was a problem hiding this comment.
Similar comments as for global-store-provide-test
| .build(); | ||
|
|
||
| final QueryableStoreType<ReadOnlyKeyValueStore<String, ValueAndTimestamp<String>>> storeType = | ||
| QueryableStoreTypes.timestampedKeyValueStore(); |
There was a problem hiding this comment.
Do we need similar test for keyValuStore()?
| .build(); | ||
|
|
||
| final QueryableStoreType<ReadOnlyWindowStore<String, ValueAndTimestamp<String>>> storeType = | ||
| QueryableStoreTypes.timestampedWindowStore(); |
There was a problem hiding this comment.
Do we need similar test for windowStore()?
| import static org.junit.jupiter.api.Assertions.assertFalse; | ||
| import static org.junit.jupiter.api.Assertions.assertTrue; | ||
|
|
||
| public class QueryableStoreTypesWithHeadersTest { |
There was a problem hiding this comment.
Should we add tests for session-store cases, too?
2 participants
This PR adds backward compatibility support for the new headers-aware
state stores (TimestampedKeyValueStoreWithHeaders and
TimestampedWindowStoreWithHeaders) when queried through the
Interactive Queries v1 (IQv1) API. Users can now query headers-aware
stores using the existing QueryableStoreTypes API without any code
changes.