aliehsaeedii commented on code in PR #21906:
URL: https://github.com/apache/kafka/pull/21906#discussion_r3461017276
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java:
##########
@@ -1293,4 +1301,864 @@ public void run() {
}
}
+ @Test
+ public void shouldQueryHeadersAwareKeyValueStoreBasicOperations(final
TestInfo testInfo) throws Exception {
+ final String uniqueTestName = safeUniqueTestName(testInfo);
+ final String inputTopic = uniqueTestName + "-input";
+ final String storeName = uniqueTestName + "-store";
+
+ CLUSTER.createTopic(inputTopic);
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.table(
+ inputTopic,
+ Consumed.with(Serdes.String(), Serdes.String()),
+ Materialized.<String, String>as(
+
Stores.persistentTimestampedKeyValueStoreWithHeaders(storeName))
+ .withKeySerde(Serdes.String())
+ .withValueSerde(Serdes.String())
+ );
+
+ final Properties properties = mkProperties(mkMap(
+ mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, uniqueTestName +
"-app"),
+ mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers())
+ ));
+
+ try (final KafkaStreams streams = getRunningStreams(properties,
builder, true)) {
+ IntegrationTestUtils.produceKeyValuesSynchronously(
+ inputTopic,
+ Arrays.asList(
+ new KeyValue<>("alpha", "a1"),
+ new KeyValue<>("beta", "b1"),
+ new KeyValue<>("gamma", "g1"),
+ new KeyValue<>("delta", "d1"),
+ new KeyValue<>("epsilon", "e1")
+ ),
+ TestUtils.producerConfig(
+ CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class,
+ new Properties()),
+ CLUSTER.time
+ );
+
+ retryOnExceptionWithTimeout(30_000, () -> {
+ final ReadOnlyKeyValueStore<String,
ValueTimestampHeaders<String>> store =
+ streams.store(fromNameAndType(storeName,
QueryableStoreTypes.timestampedKeyValueStoreWithHeaders()));
+ assertThat(store, notNullValue());
+
+ final ValueTimestampHeaders<String> result =
store.get("alpha");
+ assertThat(result, notNullValue());
+ assertThat(result.value(), is("a1"));
+ assertThat(result.timestamp() > 0, is(true));
+ assertThat(result.headers(), notNullValue());
Review Comment:
The records here are produced with plain `KeyValue` (no headers attached),
so the stored `Headers` are always empty and `assertThat(result.headers(),
notNullValue())` passes trivially — it never verifies that headers survive the
query path, which is the whole point of KAFKA-20259. Please produce records
that carry headers (e.g. a `ProducerRecord` with `RecordHeaders`) and assert
the queried headers match the produced ones (keys, values, and order). This
applies to every headers assertion in these new tests.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java:
##########
@@ -1293,4 +1301,864 @@ public void run() {
}
}
+ @Test
+ public void shouldQueryHeadersAwareKeyValueStoreBasicOperations(final
TestInfo testInfo) throws Exception {
+ final String uniqueTestName = safeUniqueTestName(testInfo);
+ final String inputTopic = uniqueTestName + "-input";
+ final String storeName = uniqueTestName + "-store";
+
+ CLUSTER.createTopic(inputTopic);
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.table(
+ inputTopic,
+ Consumed.with(Serdes.String(), Serdes.String()),
+ Materialized.<String, String>as(
+
Stores.persistentTimestampedKeyValueStoreWithHeaders(storeName))
+ .withKeySerde(Serdes.String())
+ .withValueSerde(Serdes.String())
+ );
+
+ final Properties properties = mkProperties(mkMap(
+ mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, uniqueTestName +
"-app"),
+ mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers())
+ ));
+
+ try (final KafkaStreams streams = getRunningStreams(properties,
builder, true)) {
+ IntegrationTestUtils.produceKeyValuesSynchronously(
+ inputTopic,
+ Arrays.asList(
+ new KeyValue<>("alpha", "a1"),
+ new KeyValue<>("beta", "b1"),
+ new KeyValue<>("gamma", "g1"),
+ new KeyValue<>("delta", "d1"),
+ new KeyValue<>("epsilon", "e1")
+ ),
+ TestUtils.producerConfig(
+ CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class,
+ new Properties()),
+ CLUSTER.time
+ );
+
+ retryOnExceptionWithTimeout(30_000, () -> {
+ final ReadOnlyKeyValueStore<String,
ValueTimestampHeaders<String>> store =
+ streams.store(fromNameAndType(storeName,
QueryableStoreTypes.timestampedKeyValueStoreWithHeaders()));
+ assertThat(store, notNullValue());
+
+ final ValueTimestampHeaders<String> result =
store.get("alpha");
+ assertThat(result, notNullValue());
+ assertThat(result.value(), is("a1"));
+ assertThat(result.timestamp() > 0, is(true));
+ assertThat(result.headers(), notNullValue());
+
+ assertThat(store.get("nonexistent"),
org.hamcrest.CoreMatchers.nullValue());
Review Comment:
Minor style nit: the new code uses fully-qualified
`org.hamcrest.CoreMatchers.nullValue()` (and `java.util.function.Function`
elsewhere) instead of imports. `notNullValue` is already statically imported
here — please add a `import static org.hamcrest.Matchers.nullValue;` (and an
import for `Function`) and use the short names for consistency with the rest of
the file.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java:
##########
@@ -1293,4 +1301,864 @@ public void run() {
}
}
+ @Test
+ public void shouldQueryHeadersAwareKeyValueStoreBasicOperations(final
TestInfo testInfo) throws Exception {
+ final String uniqueTestName = safeUniqueTestName(testInfo);
+ final String inputTopic = uniqueTestName + "-input";
+ final String storeName = uniqueTestName + "-store";
+
+ CLUSTER.createTopic(inputTopic);
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.table(
+ inputTopic,
+ Consumed.with(Serdes.String(), Serdes.String()),
+ Materialized.<String, String>as(
+
Stores.persistentTimestampedKeyValueStoreWithHeaders(storeName))
+ .withKeySerde(Serdes.String())
+ .withValueSerde(Serdes.String())
+ );
+
+ final Properties properties = mkProperties(mkMap(
+ mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, uniqueTestName +
"-app"),
+ mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers())
+ ));
+
+ try (final KafkaStreams streams = getRunningStreams(properties,
builder, true)) {
+ IntegrationTestUtils.produceKeyValuesSynchronously(
+ inputTopic,
+ Arrays.asList(
+ new KeyValue<>("alpha", "a1"),
+ new KeyValue<>("beta", "b1"),
+ new KeyValue<>("gamma", "g1"),
+ new KeyValue<>("delta", "d1"),
+ new KeyValue<>("epsilon", "e1")
+ ),
+ TestUtils.producerConfig(
+ CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class,
+ new Properties()),
+ CLUSTER.time
+ );
+
+ retryOnExceptionWithTimeout(30_000, () -> {
+ final ReadOnlyKeyValueStore<String,
ValueTimestampHeaders<String>> store =
+ streams.store(fromNameAndType(storeName,
QueryableStoreTypes.timestampedKeyValueStoreWithHeaders()));
+ assertThat(store, notNullValue());
+
+ final ValueTimestampHeaders<String> result =
store.get("alpha");
+ assertThat(result, notNullValue());
+ assertThat(result.value(), is("a1"));
+ assertThat(result.timestamp() > 0, is(true));
+ assertThat(result.headers(), notNullValue());
+
+ assertThat(store.get("nonexistent"),
org.hamcrest.CoreMatchers.nullValue());
+
+ assertThat(store.approximateNumEntries(), is(5L));
+
+ final ReadOnlyKeyValueStore<String, ValueAndTimestamp<String>>
tsStore =
+ streams.store(fromNameAndType(storeName,
QueryableStoreTypes.timestampedKeyValueStore()));
+ assertThat(tsStore, notNullValue());
+ final ValueAndTimestamp<String> tsResult = tsStore.get("beta");
+ assertThat(tsResult, notNullValue());
+ assertThat(tsResult.value(), is("b1"));
+ });
+ }
+ }
+
+ @Test
+ public void shouldQueryHeadersAwareKeyValueStoreIterationOperations(final
TestInfo testInfo) throws Exception {
+ final String uniqueTestName = safeUniqueTestName(testInfo);
+ final String inputTopic = uniqueTestName + "-input";
+ final String storeName = uniqueTestName + "-store";
+
+ CLUSTER.createTopic(inputTopic);
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.table(
+ inputTopic,
+ Consumed.with(Serdes.String(), Serdes.String()),
+ Materialized.<String, String>as(
+
Stores.persistentTimestampedKeyValueStoreWithHeaders(storeName))
+ .withKeySerde(Serdes.String())
+ .withValueSerde(Serdes.String())
+ );
+
+ final Properties properties = mkProperties(mkMap(
+ mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, uniqueTestName +
"-app"),
+ mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers())
+ ));
+
+ try (final KafkaStreams streams = getRunningStreams(properties,
builder, true)) {
+ IntegrationTestUtils.produceKeyValuesSynchronously(
+ inputTopic,
+ Arrays.asList(
+ new KeyValue<>("apple", "fruit"),
+ new KeyValue<>("banana", "fruit"),
+ new KeyValue<>("carrot", "vegetable"),
+ new KeyValue<>("date", "fruit"),
+ new KeyValue<>("eggplant", "vegetable")
+ ),
+ TestUtils.producerConfig(
+ CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class,
+ new Properties()),
+ CLUSTER.time
+ );
+
+ retryOnExceptionWithTimeout(30_000, () -> {
+ final ReadOnlyKeyValueStore<String,
ValueTimestampHeaders<String>> store =
+ streams.store(fromNameAndType(storeName,
QueryableStoreTypes.timestampedKeyValueStoreWithHeaders()));
+ assertThat(store, notNullValue());
+ assertThat(store.get("apple"), notNullValue());
+
+ int rangeCount = 0;
+ try (final KeyValueIterator<String,
ValueTimestampHeaders<String>> range = store.range("banana", "date")) {
+ while (range.hasNext()) {
+ final KeyValue<String, ValueTimestampHeaders<String>>
entry = range.next();
+ assertThat(entry.value, notNullValue());
+ assertThat(entry.value.headers(), notNullValue());
+ rangeCount++;
+ }
+ }
+ assertThat(rangeCount, is(3));
+
+ int allCount = 0;
+ try (final KeyValueIterator<String,
ValueTimestampHeaders<String>> all = store.all()) {
+ while (all.hasNext()) {
+ final KeyValue<String, ValueTimestampHeaders<String>>
entry = all.next();
+ assertThat(entry.value, notNullValue());
+ assertThat(entry.value.headers(), notNullValue());
+ allCount++;
+ }
+ }
+ assertThat(allCount, is(5));
+
+ int prefixCount = 0;
+ try (final KeyValueIterator<String,
ValueTimestampHeaders<String>> prefix =
+ store.prefixScan("ba", new StringSerializer())) {
+ while (prefix.hasNext()) {
+ final KeyValue<String, ValueTimestampHeaders<String>>
entry = prefix.next();
+ assertThat(entry.key, is("banana"));
+ assertThat(entry.value.value(), is("fruit"));
+ assertThat(entry.value.headers(), notNullValue());
+ prefixCount++;
+ }
+ }
+ assertThat(prefixCount, is(1));
+ });
+ }
+ }
+
+ @Test
+ public void shouldQueryHeadersAwareKeyValueStoreViaDslConfig(final
TestInfo testInfo) throws Exception {
+ final String uniqueTestName = safeUniqueTestName(testInfo);
+ final String inputTopic = uniqueTestName + "-input";
+ final String storeName = uniqueTestName + "-store";
+
+ CLUSTER.createTopic(inputTopic);
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.table(
+ inputTopic,
+ Consumed.with(Serdes.String(), Serdes.String()),
+ Materialized.<String, String, KeyValueStore<Bytes,
byte[]>>as(storeName)
+ .withKeySerde(Serdes.String())
+ .withValueSerde(Serdes.String())
+ );
+
+ final Properties properties = mkProperties(mkMap(
+ mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, uniqueTestName +
"-app"),
+ mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers()),
+ mkEntry(StreamsConfig.DSL_STORE_FORMAT_CONFIG, "headers")
+ ));
+
+ try (final KafkaStreams streams = getRunningStreams(properties,
builder, true)) {
+ IntegrationTestUtils.produceKeyValuesSynchronously(
+ inputTopic,
+ Collections.singletonList(new KeyValue<>("key1", "val1")),
+ TestUtils.producerConfig(
+ CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class,
+ new Properties()),
+ CLUSTER.time
+ );
+
+ retryOnExceptionWithTimeout(30_000, () -> {
+ final ReadOnlyKeyValueStore<String,
ValueTimestampHeaders<String>> store =
+ streams.store(fromNameAndType(storeName,
QueryableStoreTypes.timestampedKeyValueStoreWithHeaders()));
+ assertThat(store, notNullValue());
+ final ValueTimestampHeaders<String> result = store.get("key1");
+ assertThat(result, notNullValue());
+ assertThat(result.value(), is("val1"));
+ assertThat(result.headers(), notNullValue());
+
+ final ReadOnlyKeyValueStore<String, ValueAndTimestamp<String>>
tsStore =
+ streams.store(fromNameAndType(storeName,
QueryableStoreTypes.timestampedKeyValueStore()));
+ assertThat(tsStore, notNullValue());
+ final ValueAndTimestamp<String> tsResult = tsStore.get("key1");
+ assertThat(tsResult, notNullValue());
+ assertThat(tsResult.value(), is("val1"));
+ });
+ }
+ }
+
+ @Test
+ public void shouldQueryHeadersAwareWindowStoreOperations(final TestInfo
testInfo) throws Exception {
+ final String uniqueTestName = safeUniqueTestName(testInfo);
+ final String inputTopic = uniqueTestName + "-input";
+ final String storeName = uniqueTestName + "-window-store";
+
+ CLUSTER.createTopic(inputTopic);
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.stream(inputTopic, Consumed.with(Serdes.String(),
Serdes.String()))
+ .groupByKey()
+ .windowedBy(TimeWindows.ofSizeWithNoGrace(ofMillis(WINDOW_SIZE)))
+ .count(Materialized.<String, Long>as(
+ Stores.persistentTimestampedWindowStoreWithHeaders(storeName,
ofMillis(WINDOW_SIZE), ofMillis(WINDOW_SIZE), false))
+ .withKeySerde(Serdes.String())
+ .withValueSerde(Serdes.Long()));
+
+ final Properties properties = mkProperties(mkMap(
+ mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, uniqueTestName +
"-app"),
+ mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers())
+ ));
+
+ try (final KafkaStreams streams = getRunningStreams(properties,
builder, true)) {
+ IntegrationTestUtils.produceKeyValuesSynchronously(
+ inputTopic,
+ Arrays.asList(
+ new KeyValue<>("w-key1", "v1"),
+ new KeyValue<>("w-key1", "v2"),
+ new KeyValue<>("w-key2", "v3")
+ ),
+ TestUtils.producerConfig(
+ CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class,
+ new Properties()),
+ CLUSTER.time
+ );
+
+ retryOnExceptionWithTimeout(30_000, () -> {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<Long>>
store =
+ streams.store(fromNameAndType(storeName,
QueryableStoreTypes.timestampedWindowStoreWithHeaders()));
+ assertThat(store, notNullValue());
+
+ int fetchCount = 0;
+ try (final WindowStoreIterator<ValueTimestampHeaders<Long>>
iter =
+ store.fetch("w-key1", ofEpochMilli(0),
ofEpochMilli(System.currentTimeMillis() + WINDOW_SIZE))) {
+ while (iter.hasNext()) {
+ final KeyValue<Long, ValueTimestampHeaders<Long>>
entry = iter.next();
+ assertThat(entry.value, notNullValue());
+ assertThat(entry.value.value(), is(2L));
+ assertThat(entry.value.headers(), notNullValue());
+ fetchCount++;
+ }
+ }
+ assertThat(fetchCount > 0, is(true));
Review Comment:
`assertThat(fetchCount > 0, is(true))` (and similar `timestamp() > 0` checks
elsewhere) would pass even on partially-wrong behavior. Prefer asserting the
exact expected count/value so the test fails if the result regresses.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java:
##########
@@ -848,7 +848,6 @@ public void verifyStore(final boolean cache, final boolean
log, final StoresToTe
}
}
-
private <T> void shouldHandleRangeQueries() {
Review Comment:
This blank-line deletion is the only remaining change in this file and is
unrelated to the PR. As discussed earlier, please revert
`IQv2StoreIntegrationTest.java` entirely so the PR touches only
`QueryableStateIntegrationTest.java`.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java:
##########
@@ -1293,4 +1301,864 @@ public void run() {
}
}
+ @Test
+ public void shouldQueryHeadersAwareKeyValueStoreBasicOperations(final
TestInfo testInfo) throws Exception {
+ final String uniqueTestName = safeUniqueTestName(testInfo);
+ final String inputTopic = uniqueTestName + "-input";
+ final String storeName = uniqueTestName + "-store";
+
+ CLUSTER.createTopic(inputTopic);
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.table(
+ inputTopic,
+ Consumed.with(Serdes.String(), Serdes.String()),
+ Materialized.<String, String>as(
+
Stores.persistentTimestampedKeyValueStoreWithHeaders(storeName))
+ .withKeySerde(Serdes.String())
+ .withValueSerde(Serdes.String())
+ );
+
+ final Properties properties = mkProperties(mkMap(
+ mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, uniqueTestName +
"-app"),
+ mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers())
+ ));
+
+ try (final KafkaStreams streams = getRunningStreams(properties,
builder, true)) {
+ IntegrationTestUtils.produceKeyValuesSynchronously(
+ inputTopic,
+ Arrays.asList(
+ new KeyValue<>("alpha", "a1"),
+ new KeyValue<>("beta", "b1"),
+ new KeyValue<>("gamma", "g1"),
+ new KeyValue<>("delta", "d1"),
+ new KeyValue<>("epsilon", "e1")
+ ),
+ TestUtils.producerConfig(
+ CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class,
+ new Properties()),
+ CLUSTER.time
+ );
+
+ retryOnExceptionWithTimeout(30_000, () -> {
+ final ReadOnlyKeyValueStore<String,
ValueTimestampHeaders<String>> store =
+ streams.store(fromNameAndType(storeName,
QueryableStoreTypes.timestampedKeyValueStoreWithHeaders()));
+ assertThat(store, notNullValue());
+
+ final ValueTimestampHeaders<String> result =
store.get("alpha");
+ assertThat(result, notNullValue());
+ assertThat(result.value(), is("a1"));
+ assertThat(result.timestamp() > 0, is(true));
+ assertThat(result.headers(), notNullValue());
+
+ assertThat(store.get("nonexistent"),
org.hamcrest.CoreMatchers.nullValue());
+
+ assertThat(store.approximateNumEntries(), is(5L));
+
+ final ReadOnlyKeyValueStore<String, ValueAndTimestamp<String>>
tsStore =
+ streams.store(fromNameAndType(storeName,
QueryableStoreTypes.timestampedKeyValueStore()));
+ assertThat(tsStore, notNullValue());
+ final ValueAndTimestamp<String> tsResult = tsStore.get("beta");
+ assertThat(tsResult, notNullValue());
+ assertThat(tsResult.value(), is("b1"));
+ });
+ }
+ }
+
+ @Test
+ public void shouldQueryHeadersAwareKeyValueStoreIterationOperations(final
TestInfo testInfo) throws Exception {
+ final String uniqueTestName = safeUniqueTestName(testInfo);
+ final String inputTopic = uniqueTestName + "-input";
+ final String storeName = uniqueTestName + "-store";
+
+ CLUSTER.createTopic(inputTopic);
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.table(
+ inputTopic,
+ Consumed.with(Serdes.String(), Serdes.String()),
+ Materialized.<String, String>as(
+
Stores.persistentTimestampedKeyValueStoreWithHeaders(storeName))
+ .withKeySerde(Serdes.String())
+ .withValueSerde(Serdes.String())
+ );
+
+ final Properties properties = mkProperties(mkMap(
+ mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, uniqueTestName +
"-app"),
+ mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers())
+ ));
+
+ try (final KafkaStreams streams = getRunningStreams(properties,
builder, true)) {
+ IntegrationTestUtils.produceKeyValuesSynchronously(
+ inputTopic,
+ Arrays.asList(
+ new KeyValue<>("apple", "fruit"),
+ new KeyValue<>("banana", "fruit"),
+ new KeyValue<>("carrot", "vegetable"),
+ new KeyValue<>("date", "fruit"),
+ new KeyValue<>("eggplant", "vegetable")
+ ),
+ TestUtils.producerConfig(
+ CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class,
+ new Properties()),
+ CLUSTER.time
+ );
+
+ retryOnExceptionWithTimeout(30_000, () -> {
+ final ReadOnlyKeyValueStore<String,
ValueTimestampHeaders<String>> store =
+ streams.store(fromNameAndType(storeName,
QueryableStoreTypes.timestampedKeyValueStoreWithHeaders()));
+ assertThat(store, notNullValue());
+ assertThat(store.get("apple"), notNullValue());
+
+ int rangeCount = 0;
+ try (final KeyValueIterator<String,
ValueTimestampHeaders<String>> range = store.range("banana", "date")) {
+ while (range.hasNext()) {
+ final KeyValue<String, ValueTimestampHeaders<String>>
entry = range.next();
+ assertThat(entry.value, notNullValue());
+ assertThat(entry.value.headers(), notNullValue());
+ rangeCount++;
+ }
+ }
+ assertThat(rangeCount, is(3));
+
+ int allCount = 0;
+ try (final KeyValueIterator<String,
ValueTimestampHeaders<String>> all = store.all()) {
+ while (all.hasNext()) {
+ final KeyValue<String, ValueTimestampHeaders<String>>
entry = all.next();
+ assertThat(entry.value, notNullValue());
+ assertThat(entry.value.headers(), notNullValue());
+ allCount++;
+ }
+ }
+ assertThat(allCount, is(5));
+
+ int prefixCount = 0;
+ try (final KeyValueIterator<String,
ValueTimestampHeaders<String>> prefix =
+ store.prefixScan("ba", new StringSerializer())) {
+ while (prefix.hasNext()) {
+ final KeyValue<String, ValueTimestampHeaders<String>>
entry = prefix.next();
+ assertThat(entry.key, is("banana"));
+ assertThat(entry.value.value(), is("fruit"));
+ assertThat(entry.value.headers(), notNullValue());
+ prefixCount++;
+ }
+ }
+ assertThat(prefixCount, is(1));
+ });
+ }
+ }
+
+ @Test
+ public void shouldQueryHeadersAwareKeyValueStoreViaDslConfig(final
TestInfo testInfo) throws Exception {
+ final String uniqueTestName = safeUniqueTestName(testInfo);
+ final String inputTopic = uniqueTestName + "-input";
+ final String storeName = uniqueTestName + "-store";
+
+ CLUSTER.createTopic(inputTopic);
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.table(
+ inputTopic,
+ Consumed.with(Serdes.String(), Serdes.String()),
+ Materialized.<String, String, KeyValueStore<Bytes,
byte[]>>as(storeName)
+ .withKeySerde(Serdes.String())
+ .withValueSerde(Serdes.String())
+ );
+
+ final Properties properties = mkProperties(mkMap(
+ mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, uniqueTestName +
"-app"),
+ mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers()),
+ mkEntry(StreamsConfig.DSL_STORE_FORMAT_CONFIG, "headers")
+ ));
+
+ try (final KafkaStreams streams = getRunningStreams(properties,
builder, true)) {
+ IntegrationTestUtils.produceKeyValuesSynchronously(
+ inputTopic,
+ Collections.singletonList(new KeyValue<>("key1", "val1")),
+ TestUtils.producerConfig(
+ CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class,
+ new Properties()),
+ CLUSTER.time
+ );
+
+ retryOnExceptionWithTimeout(30_000, () -> {
+ final ReadOnlyKeyValueStore<String,
ValueTimestampHeaders<String>> store =
+ streams.store(fromNameAndType(storeName,
QueryableStoreTypes.timestampedKeyValueStoreWithHeaders()));
+ assertThat(store, notNullValue());
+ final ValueTimestampHeaders<String> result = store.get("key1");
+ assertThat(result, notNullValue());
+ assertThat(result.value(), is("val1"));
+ assertThat(result.headers(), notNullValue());
+
+ final ReadOnlyKeyValueStore<String, ValueAndTimestamp<String>>
tsStore =
+ streams.store(fromNameAndType(storeName,
QueryableStoreTypes.timestampedKeyValueStore()));
+ assertThat(tsStore, notNullValue());
+ final ValueAndTimestamp<String> tsResult = tsStore.get("key1");
+ assertThat(tsResult, notNullValue());
+ assertThat(tsResult.value(), is("val1"));
+ });
+ }
+ }
+
+ @Test
+ public void shouldQueryHeadersAwareWindowStoreOperations(final TestInfo
testInfo) throws Exception {
+ final String uniqueTestName = safeUniqueTestName(testInfo);
+ final String inputTopic = uniqueTestName + "-input";
+ final String storeName = uniqueTestName + "-window-store";
+
+ CLUSTER.createTopic(inputTopic);
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.stream(inputTopic, Consumed.with(Serdes.String(),
Serdes.String()))
+ .groupByKey()
+ .windowedBy(TimeWindows.ofSizeWithNoGrace(ofMillis(WINDOW_SIZE)))
+ .count(Materialized.<String, Long>as(
+ Stores.persistentTimestampedWindowStoreWithHeaders(storeName,
ofMillis(WINDOW_SIZE), ofMillis(WINDOW_SIZE), false))
+ .withKeySerde(Serdes.String())
+ .withValueSerde(Serdes.Long()));
+
+ final Properties properties = mkProperties(mkMap(
+ mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, uniqueTestName +
"-app"),
+ mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers())
+ ));
+
+ try (final KafkaStreams streams = getRunningStreams(properties,
builder, true)) {
+ IntegrationTestUtils.produceKeyValuesSynchronously(
+ inputTopic,
+ Arrays.asList(
+ new KeyValue<>("w-key1", "v1"),
+ new KeyValue<>("w-key1", "v2"),
+ new KeyValue<>("w-key2", "v3")
+ ),
+ TestUtils.producerConfig(
+ CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class,
+ new Properties()),
+ CLUSTER.time
+ );
+
+ retryOnExceptionWithTimeout(30_000, () -> {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<Long>>
store =
+ streams.store(fromNameAndType(storeName,
QueryableStoreTypes.timestampedWindowStoreWithHeaders()));
+ assertThat(store, notNullValue());
+
+ int fetchCount = 0;
+ try (final WindowStoreIterator<ValueTimestampHeaders<Long>>
iter =
+ store.fetch("w-key1", ofEpochMilli(0),
ofEpochMilli(System.currentTimeMillis() + WINDOW_SIZE))) {
+ while (iter.hasNext()) {
+ final KeyValue<Long, ValueTimestampHeaders<Long>>
entry = iter.next();
+ assertThat(entry.value, notNullValue());
+ assertThat(entry.value.value(), is(2L));
+ assertThat(entry.value.headers(), notNullValue());
+ fetchCount++;
+ }
+ }
+ assertThat(fetchCount > 0, is(true));
+
+ int allCount = 0;
+ try (final KeyValueIterator<Windowed<String>,
ValueTimestampHeaders<Long>> all = store.all()) {
+ while (all.hasNext()) {
+ final KeyValue<Windowed<String>,
ValueTimestampHeaders<Long>> entry = all.next();
+ assertThat(entry.value, notNullValue());
+ assertThat(entry.value.headers(), notNullValue());
+ allCount++;
+ }
+ }
+ assertThat(allCount, is(2));
+
+ final ReadOnlyWindowStore<String, ValueAndTimestamp<Long>>
tsStore =
+ streams.store(fromNameAndType(storeName,
QueryableStoreTypes.timestampedWindowStore()));
+ assertThat(tsStore, notNullValue());
+ try (final WindowStoreIterator<ValueAndTimestamp<Long>> tsIter
=
+ tsStore.fetch("w-key2", ofEpochMilli(0),
ofEpochMilli(System.currentTimeMillis() + WINDOW_SIZE))) {
+ assertThat(tsIter.hasNext(), is(true));
+ assertThat(tsIter.next().value.value(), is(1L));
+ }
+ });
+ }
+ }
+
+ @Test
+ public void shouldQueryHeadersAwareSessionStoreOperations(final TestInfo
testInfo) throws Exception {
+ final String uniqueTestName = safeUniqueTestName(testInfo);
+ final String inputTopic = uniqueTestName + "-input";
+ final String storeName = uniqueTestName + "-session-store";
+
+ CLUSTER.createTopic(inputTopic);
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.stream(inputTopic, Consumed.with(Serdes.String(),
Serdes.String()))
+ .groupByKey()
+
.windowedBy(SessionWindows.ofInactivityGapWithNoGrace(ofMillis(WINDOW_SIZE)))
+ .count(Materialized.<String, Long>as(
+ Stores.persistentSessionStoreWithHeaders(storeName,
ofMillis(WINDOW_SIZE)))
+ .withKeySerde(Serdes.String())
+ .withValueSerde(Serdes.Long()));
+
+ final Properties properties = mkProperties(mkMap(
+ mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, uniqueTestName +
"-app"),
+ mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers())
+ ));
+
+ try (final KafkaStreams streams = getRunningStreams(properties,
builder, true)) {
+ IntegrationTestUtils.produceKeyValuesSynchronously(
+ inputTopic,
+ Arrays.asList(
+ new KeyValue<>("s-key1", "v1"),
+ new KeyValue<>("s-key1", "v2"),
+ new KeyValue<>("s-key2", "v3")
+ ),
+ TestUtils.producerConfig(
+ CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class,
+ new Properties()),
+ CLUSTER.time
+ );
+
+ retryOnExceptionWithTimeout(30_000, () -> {
+ final ReadOnlySessionStore<String,
AggregationWithHeaders<Long>> store =
+ streams.store(fromNameAndType(storeName,
QueryableStoreTypes.sessionStoreWithHeaders()));
+ assertThat(store, notNullValue());
+
+ int sessionCount = 0;
+ try (final KeyValueIterator<Windowed<String>,
AggregationWithHeaders<Long>> iter = store.fetch("s-key1")) {
+ while (iter.hasNext()) {
+ final KeyValue<Windowed<String>,
AggregationWithHeaders<Long>> entry = iter.next();
+ assertThat(entry.value, notNullValue());
+ assertThat(entry.value.aggregation(), is(2L));
+ assertThat(entry.value.headers(), notNullValue());
+ sessionCount++;
+ }
+ }
+ assertThat(sessionCount, is(1));
+
+ final ReadOnlySessionStore<String, Long> plainStore =
+ streams.store(fromNameAndType(storeName, sessionStore()));
+ assertThat(plainStore, notNullValue());
+ try (final KeyValueIterator<Windowed<String>, Long> plainIter
= plainStore.fetch("s-key2")) {
+ assertThat(plainIter.hasNext(), is(true));
+ assertThat(plainIter.next().value, is(1L));
+ }
+ });
+ }
+ }
+
+ @Test
+ public void shouldQueryHeadersAwareKeyValueStoreAfterRestart(final
TestInfo testInfo) throws Exception {
+ final String uniqueTestName = safeUniqueTestName(testInfo);
+ final String inputTopic = uniqueTestName + "-input";
+ final String storeName = uniqueTestName + "-store";
+ final String stateDir =
TestUtils.tempDirectory("headers-restart-test").getPath();
+
+ CLUSTER.createTopic(inputTopic);
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.table(
+ inputTopic,
+ Consumed.with(Serdes.String(), Serdes.String()),
+ Materialized.<String, String>as(
+
Stores.persistentTimestampedKeyValueStoreWithHeaders(storeName))
+ .withKeySerde(Serdes.String())
+ .withValueSerde(Serdes.String())
+ );
+
+ final Properties properties = mkProperties(mkMap(
+ mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, uniqueTestName +
"-app"),
+ mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers()),
+ mkEntry(StreamsConfig.STATE_DIR_CONFIG, stateDir)
+ ));
+
+ try (final KafkaStreams streams = getRunningStreams(properties,
builder, true)) {
+ IntegrationTestUtils.produceKeyValuesSynchronously(
+ inputTopic,
+ Arrays.asList(
+ new KeyValue<>("r-key1", "val1"),
+ new KeyValue<>("r-key2", "val2")
+ ),
+ TestUtils.producerConfig(
+ CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class,
+ new Properties()),
+ CLUSTER.time
+ );
+
+ retryOnExceptionWithTimeout(30_000, () -> {
+ final ReadOnlyKeyValueStore<String,
ValueTimestampHeaders<String>> store =
+ streams.store(fromNameAndType(storeName,
QueryableStoreTypes.timestampedKeyValueStoreWithHeaders()));
+ assertThat(store.get("r-key1"), notNullValue());
+ assertThat(store.get("r-key1").value(), is("val1"));
+ });
+ }
+
+ final StreamsBuilder builder2 = new StreamsBuilder();
+ builder2.table(
+ inputTopic,
+ Consumed.with(Serdes.String(), Serdes.String()),
+ Materialized.<String, String>as(
+
Stores.persistentTimestampedKeyValueStoreWithHeaders(storeName))
+ .withKeySerde(Serdes.String())
+ .withValueSerde(Serdes.String())
+ );
+
+ try (final KafkaStreams streams2 = getRunningStreams(properties,
builder2, true)) {
+ retryOnExceptionWithTimeout(30_000, () -> {
+ final ReadOnlyKeyValueStore<String,
ValueTimestampHeaders<String>> store =
+ streams2.store(fromNameAndType(storeName,
QueryableStoreTypes.timestampedKeyValueStoreWithHeaders()));
+ assertThat(store, notNullValue());
+ final ValueTimestampHeaders<String> result =
store.get("r-key1");
+ assertThat(result, notNullValue());
+ assertThat(result.value(), is("val1"));
+ assertThat(result.headers(), notNullValue());
+
+ final ValueTimestampHeaders<String> result2 =
store.get("r-key2");
+ assertThat(result2, notNullValue());
+ assertThat(result2.value(), is("val2"));
+ });
+ }
+ }
+
+ @Test
+ public void shouldQueryHeadersAwareKeyValueStoreReverseOperations(final
TestInfo testInfo) throws Exception {
+ final String uniqueTestName = safeUniqueTestName(testInfo);
+ final String inputTopic = uniqueTestName + "-input";
+ final String storeName = uniqueTestName + "-store";
+
+ CLUSTER.createTopic(inputTopic);
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.table(
+ inputTopic,
+ Consumed.with(Serdes.String(), Serdes.String()),
+ Materialized.<String, String>as(
+
Stores.persistentTimestampedKeyValueStoreWithHeaders(storeName))
+ .withKeySerde(Serdes.String())
+ .withValueSerde(Serdes.String())
+ );
+
+ final Properties properties = mkProperties(mkMap(
+ mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, uniqueTestName +
"-app"),
+ mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers())
+ ));
+
+ try (final KafkaStreams streams = getRunningStreams(properties,
builder, true)) {
+ IntegrationTestUtils.produceKeyValuesSynchronously(
+ inputTopic,
+ Arrays.asList(
+ new KeyValue<>("alpha", "a1"),
+ new KeyValue<>("beta", "b1"),
+ new KeyValue<>("gamma", "g1"),
+ new KeyValue<>("delta", "d1"),
+ new KeyValue<>("epsilon", "e1")
+ ),
+ TestUtils.producerConfig(
+ CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class,
+ new Properties()),
+ CLUSTER.time
+ );
+
+ retryOnExceptionWithTimeout(30_000, () -> {
+ final ReadOnlyKeyValueStore<String,
ValueTimestampHeaders<String>> store =
+ streams.store(fromNameAndType(storeName,
QueryableStoreTypes.timestampedKeyValueStoreWithHeaders()));
+ assertThat(store.get("alpha"), notNullValue());
+
+ final List<String> reverseRangeKeys = new ArrayList<>();
+ try (final KeyValueIterator<String,
ValueTimestampHeaders<String>> iter = store.reverseRange("beta", "epsilon")) {
+ while (iter.hasNext()) {
+ final KeyValue<String, ValueTimestampHeaders<String>>
entry = iter.next();
+ assertThat(entry.value.headers(), notNullValue());
+ reverseRangeKeys.add(entry.key);
+ }
+ }
+ assertThat(reverseRangeKeys, is(Arrays.asList("epsilon",
"delta", "beta")));
+
+ final List<String> reverseAllKeys = new ArrayList<>();
+ try (final KeyValueIterator<String,
ValueTimestampHeaders<String>> iter = store.reverseAll()) {
+ while (iter.hasNext()) {
+ final KeyValue<String, ValueTimestampHeaders<String>>
entry = iter.next();
+ assertThat(entry.value.headers(), notNullValue());
+ reverseAllKeys.add(entry.key);
+ }
+ }
+ assertThat(reverseAllKeys, is(Arrays.asList("gamma",
"epsilon", "delta", "beta", "alpha")));
+ });
+ }
+ }
+
+ private <K, V> int countAndVerifyHeaders(final KeyValueIterator<K, V> iter,
Review Comment:
`countAndVerifyHeaders` / `countAndVerifyWindowHeaders` are only used in one
test, while the KV-iteration and session tests duplicate the same `while
(hasNext)` loop inline. Either route all the iteration tests through these
helpers, or drop them. They'd also read better grouped with the other private
helpers rather than sitting between test methods.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
