aliehsaeedii commented on code in PR #22901:
URL: https://github.com/apache/kafka/pull/22901#discussion_r3669063294
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -135,6 +137,117 @@ private void buildAndStart(final StoreBuilder<?>
storeBuilder,
IntegrationTestUtils.startApplicationAndWaitUntilRunning(kafkaStreams);
}
+ /**
+ * Produces a single record (no explicit timestamp) into the input stream.
+ */
+ private <K, V> void produce(final K key, final V value) {
+ IntegrationTestUtils.produceKeyValuesSynchronously(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
StringSerializer.class, StringSerializer.class),
+ CLUSTER.time,
+ false);
+ }
+
+ /**
+ * Produces a single record with an explicit timestamp into the input
stream.
+ */
+ private <K, V> void produce(final K key, final V value, final long
timestamp) {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
StringSerializer.class, StringSerializer.class),
+ timestamp,
+ false);
+ }
+
+ /**
+ * Produces a single record with an explicit timestamp and headers into
the input stream.
+ */
+ private <K, V> void produce(final K key, final V value, final long
timestamp, final Headers headers) {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
StringSerializer.class, StringSerializer.class),
+ headers,
+ timestamp,
+ false);
+ }
+
+ /**
+ * Shared skeleton for the process-and-verify / verify helpers: waits
until the named store is
+ * queryable and the supplied condition holds. The store lookup is retried
until the condition
+ * passes or the timeout elapses; transient query {@link Exception}s (e.g.
store not yet ready)
+ * are swallowed and treated as "not ready". {@link AssertionError} thrown
by a condition that
+ * uses {@code assertX(...)} is intentionally NOT caught here, so those
helpers still fail fast.
+ */
+ private <S> void awaitStore(final String storeName,
+ final QueryableStoreType<S> storeType,
+ final Predicate<S> condition,
+ final String message) throws Exception {
+ TestUtils.waitForCondition(() -> {
+ try {
+ final S store = IntegrationTestUtils.getStore(storeName,
kafkaStreams, storeType);
+ if (store == null) {
+ return false;
+ }
+ return condition.test(store);
+ } catch (final Exception swallow) {
+ LOG.error("Error while checking store result", swallow);
+ return false;
+ }
+ }, DEFAULT_STORE_TIMEOUT_MS, message);
+ }
+
+ /**
+ * Computes the start of the window that {@code timestamp} falls into for
the fixed
+ * {@link #WINDOW_SIZE_MS} window size.
+ */
+ private static long windowStart(final long timestamp) {
+ return timestamp - (timestamp % WINDOW_SIZE_MS);
+ }
+
+ /**
+ * Finds the value stored for {@code key} in the window that {@code
timestamp} falls into,
+ * by scanning {@link ReadOnlyWindowStore#all()} and matching on key and
window start.
+ */
+ private static Optional<ValueTimestampHeaders<String>> findWindowedValue(
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store,
+ final String key,
+ final long timestamp) {
+ final long start = windowStart(timestamp);
+ try (final KeyValueIterator<Windowed<String>,
ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>,
ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() ==
start) {
+ return Optional.ofNullable(kv.value);
Review Comment:
`Optional.ofNullable` maps a null stored value to `empty`, so callers can no
longer tell "no record yet" from "record with a null value". A null value used
to fail immediately on `assertNotNull`; now it just retries until the 60s
timeout and reports a generic message.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -814,261 +822,115 @@ public void
shouldProxyTimestampedWindowStoreToTimestampedWindowStoreWithHeaders
private void processPlainWindowedKeyValueAndVerify(final String key,
final String value,
final long timestamp)
throws Exception {
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
- inputStream,
- List.of(KeyValue.pair(key, value)),
- TestUtils.producerConfig(CLUSTER.bootstrapServers(),
- StringSerializer.class,
- StringSerializer.class),
- timestamp,
- false);
-
- TestUtils.waitForCondition(() -> {
- try {
- final ReadOnlyWindowStore<String, String> store =
- IntegrationTestUtils.getStore(WINDOW_STORE_NAME,
kafkaStreams, QueryableStoreTypes.windowStore());
-
- if (store == null) {
- return false;
- }
-
- final long windowStart = timestamp - (timestamp %
WINDOW_SIZE_MS);
- final String result = store.fetch(key, windowStart);
+ produce(key, value, timestamp);
+ awaitStore(WINDOW_STORE_NAME, QueryableStoreTypes.<String,
String>windowStore(),
+ store -> {
+ final String result = store.fetch(key, windowStart(timestamp));
return result != null && result.equals(value);
- } catch (final Exception e) {
- return false;
- }
- }, 60_000L, "Could not verify plain window value in time.");
+ },
+ "Could not verify plain window value in time.");
}
private void verifyPlainWindowValueWithEmptyHeadersAndTimestamp(final
String key,
final
String value,
final long
windowTimestamp,
final long
expectedTimestamp) throws Exception {
- TestUtils.waitForCondition(() -> {
- try {
- final ReadOnlyWindowStore<String,
ValueTimestampHeaders<String>> store =
- IntegrationTestUtils.getStore(WINDOW_STORE_NAME,
kafkaStreams, QueryableStoreTypes.timestampedWindowStoreWithHeaders());
-
- if (store == null) {
- return false;
- }
-
- final long windowStart = windowTimestamp - (windowTimestamp %
WINDOW_SIZE_MS);
-
- final List<KeyValue<Windowed<String>,
ValueTimestampHeaders<String>>> results = new LinkedList<>();
- try (final KeyValueIterator<Windowed<String>,
ValueTimestampHeaders<String>> iterator = store.all()) {
- while (iterator.hasNext()) {
- final KeyValue<Windowed<String>,
ValueTimestampHeaders<String>> kv = iterator.next();
- if (kv.key.key().equals(key) &&
kv.key.window().start() == windowStart) {
- results.add(kv);
- }
- }
- }
-
- if (results.isEmpty()) {
+ awaitStore(WINDOW_STORE_NAME, QueryableStoreTypes.<String,
String>timestampedWindowStoreWithHeaders(),
+ store -> {
+ final Optional<ValueTimestampHeaders<String>> result =
findWindowedValue(store, key, windowTimestamp);
+ if (result.isEmpty()) {
return false;
}
- final ValueTimestampHeaders<String> result =
results.get(0).value;
- assertNotNull(result, "Result should not be null");
- assertEquals(value, result.value(), "Value should match");
- assertEquals(expectedTimestamp, result.timestamp(), "Timestamp
should be " + expectedTimestamp + " for plain store migration");
+ final ValueTimestampHeaders<String> value0 = result.get();
+ assertNotNull(value0, "Result should not be null");
Review Comment:
This can never fire: `value0` is `result.get()` after the `isEmpty()` guard,
so it is never null. Same dead assert in `verifyWindowValueWithEmptyHeaders`
and `verifySessionValueWithEmptyHeaders` — worth dropping all three.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -135,6 +137,117 @@ private void buildAndStart(final StoreBuilder<?>
storeBuilder,
IntegrationTestUtils.startApplicationAndWaitUntilRunning(kafkaStreams);
}
+ /**
+ * Produces a single record (no explicit timestamp) into the input stream.
+ */
+ private <K, V> void produce(final K key, final V value) {
+ IntegrationTestUtils.produceKeyValuesSynchronously(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
StringSerializer.class, StringSerializer.class),
+ CLUSTER.time,
+ false);
+ }
+
+ /**
+ * Produces a single record with an explicit timestamp into the input
stream.
+ */
+ private <K, V> void produce(final K key, final V value, final long
timestamp) {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
StringSerializer.class, StringSerializer.class),
+ timestamp,
+ false);
+ }
+
+ /**
+ * Produces a single record with an explicit timestamp and headers into
the input stream.
+ */
+ private <K, V> void produce(final K key, final V value, final long
timestamp, final Headers headers) {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
StringSerializer.class, StringSerializer.class),
+ headers,
+ timestamp,
+ false);
+ }
+
+ /**
+ * Shared skeleton for the process-and-verify / verify helpers: waits
until the named store is
+ * queryable and the supplied condition holds. The store lookup is retried
until the condition
+ * passes or the timeout elapses; transient query {@link Exception}s (e.g.
store not yet ready)
+ * are swallowed and treated as "not ready". {@link AssertionError} thrown
by a condition that
+ * uses {@code assertX(...)} is intentionally NOT caught here, so those
helpers still fail fast.
+ */
+ private <S> void awaitStore(final String storeName,
+ final QueryableStoreType<S> storeType,
+ final Predicate<S> condition,
+ final String message) throws Exception {
+ TestUtils.waitForCondition(() -> {
+ try {
+ final S store = IntegrationTestUtils.getStore(storeName,
kafkaStreams, storeType);
+ if (store == null) {
Review Comment:
`IntegrationTestUtils.getStore` never returns null — it throws
`InvalidStateStoreException` once its own timeout elapses — so this branch is
unreachable and can go.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -371,67 +484,27 @@ private <K, V> void
processKeyValueAndVerifyTimestampedValue(final K key,
final long
timestamp)
throws Exception {
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
- inputStream,
- singletonList(KeyValue.pair(key, value)),
- TestUtils.producerConfig(CLUSTER.bootstrapServers(),
- StringSerializer.class,
- StringSerializer.class),
- timestamp,
- false);
-
- TestUtils.waitForCondition(
- () -> {
- try {
- final ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>> store
=
- IntegrationTestUtils.getStore(STORE_NAME,
kafkaStreams, QueryableStoreTypes.timestampedKeyValueStore());
+ produce(key, value, timestamp);
- if (store == null) {
- return false;
- }
-
- final ValueAndTimestamp<V> result = store.get(key);
- return result != null && result.value().equals(value) &&
result.timestamp() == timestamp;
- } catch (final Exception swallow) {
- LOG.error("Error while checking store result", swallow);
- return false;
- }
+ awaitStore(STORE_NAME, QueryableStoreTypes.<K,
V>timestampedKeyValueStore(),
Review Comment:
This `awaitStore` block is now identical to `verifyLegacyTimestampedValue` —
make this method `produce(key, value, timestamp);
verifyLegacyTimestampedValue(key, value, timestamp);`.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -1295,20 +1157,7 @@ private boolean windowStoreContainsKey(final String key,
final long timestamp) {
final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>>
store =
IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams,
QueryableStoreTypes.timestampedWindowStoreWithHeaders());
- if (store == null) {
- return false;
- }
-
- final long expectedWindowStart = timestamp - (timestamp %
WINDOW_SIZE_MS);
- try (final KeyValueIterator<Windowed<String>,
ValueTimestampHeaders<String>> iterator = store.all()) {
- while (iterator.hasNext()) {
- final KeyValue<Windowed<String>,
ValueTimestampHeaders<String>> kv = iterator.next();
- if (kv.key.key().equals(key) && kv.key.window().start() ==
expectedWindowStart) {
- return true;
- }
- }
- }
- return false;
+ return store != null && findWindowedValue(store, key,
timestamp).isPresent();
Review Comment:
`windowStoreContainsKey` and `sessionStoreContainsKey` still hand-roll the
get-store-and-swallow pattern, and their two call sites still wrap them in a
raw `TestUtils.waitForCondition(..., 30_000L, ...)`. Folding those into
`awaitStore` would finish the wrapper cleanup.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -135,6 +137,117 @@ private void buildAndStart(final StoreBuilder<?>
storeBuilder,
IntegrationTestUtils.startApplicationAndWaitUntilRunning(kafkaStreams);
}
+ /**
+ * Produces a single record (no explicit timestamp) into the input stream.
+ */
+ private <K, V> void produce(final K key, final V value) {
+ IntegrationTestUtils.produceKeyValuesSynchronously(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
StringSerializer.class, StringSerializer.class),
+ CLUSTER.time,
+ false);
+ }
+
+ /**
+ * Produces a single record with an explicit timestamp into the input
stream.
+ */
+ private <K, V> void produce(final K key, final V value, final long
timestamp) {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
StringSerializer.class, StringSerializer.class),
+ timestamp,
+ false);
+ }
+
+ /**
+ * Produces a single record with an explicit timestamp and headers into
the input stream.
+ */
+ private <K, V> void produce(final K key, final V value, final long
timestamp, final Headers headers) {
Review Comment:
`setupAndPopulateSessionStoreWithHeaders` still calls
`IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp` inline with
exactly these arguments — switch it to this helper too.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -814,261 +822,115 @@ public void
shouldProxyTimestampedWindowStoreToTimestampedWindowStoreWithHeaders
private void processPlainWindowedKeyValueAndVerify(final String key,
final String value,
final long timestamp)
throws Exception {
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
- inputStream,
- List.of(KeyValue.pair(key, value)),
- TestUtils.producerConfig(CLUSTER.bootstrapServers(),
- StringSerializer.class,
- StringSerializer.class),
- timestamp,
- false);
-
- TestUtils.waitForCondition(() -> {
- try {
- final ReadOnlyWindowStore<String, String> store =
- IntegrationTestUtils.getStore(WINDOW_STORE_NAME,
kafkaStreams, QueryableStoreTypes.windowStore());
-
- if (store == null) {
- return false;
- }
-
- final long windowStart = timestamp - (timestamp %
WINDOW_SIZE_MS);
- final String result = store.fetch(key, windowStart);
+ produce(key, value, timestamp);
+ awaitStore(WINDOW_STORE_NAME, QueryableStoreTypes.<String,
String>windowStore(),
+ store -> {
+ final String result = store.fetch(key, windowStart(timestamp));
return result != null && result.equals(value);
- } catch (final Exception e) {
- return false;
- }
- }, 60_000L, "Could not verify plain window value in time.");
+ },
+ "Could not verify plain window value in time.");
}
private void verifyPlainWindowValueWithEmptyHeadersAndTimestamp(final
String key,
final
String value,
final long
windowTimestamp,
final long
expectedTimestamp) throws Exception {
- TestUtils.waitForCondition(() -> {
- try {
- final ReadOnlyWindowStore<String,
ValueTimestampHeaders<String>> store =
- IntegrationTestUtils.getStore(WINDOW_STORE_NAME,
kafkaStreams, QueryableStoreTypes.timestampedWindowStoreWithHeaders());
-
- if (store == null) {
- return false;
- }
-
- final long windowStart = windowTimestamp - (windowTimestamp %
WINDOW_SIZE_MS);
-
- final List<KeyValue<Windowed<String>,
ValueTimestampHeaders<String>>> results = new LinkedList<>();
- try (final KeyValueIterator<Windowed<String>,
ValueTimestampHeaders<String>> iterator = store.all()) {
- while (iterator.hasNext()) {
- final KeyValue<Windowed<String>,
ValueTimestampHeaders<String>> kv = iterator.next();
- if (kv.key.key().equals(key) &&
kv.key.window().start() == windowStart) {
- results.add(kv);
- }
- }
- }
-
- if (results.isEmpty()) {
+ awaitStore(WINDOW_STORE_NAME, QueryableStoreTypes.<String,
String>timestampedWindowStoreWithHeaders(),
+ store -> {
+ final Optional<ValueTimestampHeaders<String>> result =
findWindowedValue(store, key, windowTimestamp);
+ if (result.isEmpty()) {
return false;
}
- final ValueTimestampHeaders<String> result =
results.get(0).value;
- assertNotNull(result, "Result should not be null");
- assertEquals(value, result.value(), "Value should match");
- assertEquals(expectedTimestamp, result.timestamp(), "Timestamp
should be " + expectedTimestamp + " for plain store migration");
+ final ValueTimestampHeaders<String> value0 = result.get();
+ assertNotNull(value0, "Result should not be null");
+ assertEquals(value, value0.value(), "Value should match");
+ assertEquals(expectedTimestamp, value0.timestamp(), "Timestamp
should be " + expectedTimestamp + " for plain store migration");
// Verify headers exist but are empty (migrated from plain
store without headers or timestamps)
- assertNotNull(result.headers(), "Headers should not be null
for migrated data");
- assertEquals(0, result.headers().toArray().length, "Headers
should be empty for migrated data");
+ assertNotNull(value0.headers(), "Headers should not be null
for migrated data");
+ assertEquals(0, value0.headers().toArray().length, "Headers
should be empty for migrated data");
return true;
- } catch (final Exception e) {
- LOG.error("Error while verifying plain window value with empty
headers and timestamp", e);
- return false;
- }
- }, 60_000L, "Could not verify plain window value with empty headers
and timestamp in time.");
+ },
+ "Could not verify plain window value with empty headers and
timestamp in time.");
}
private void processPlainWindowedKeyValueWithHeadersAndVerify(final String
key,
final String
value,
final long
timestamp,
final
Headers headers,
final
Headers expectedHeaders) throws Exception {
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
- inputStream,
- List.of(KeyValue.pair(key, value)),
- TestUtils.producerConfig(CLUSTER.bootstrapServers(),
- StringSerializer.class,
- StringSerializer.class),
- headers,
- timestamp,
- false);
-
- TestUtils.waitForCondition(() -> {
- try {
- final ReadOnlyWindowStore<String,
ValueTimestampHeaders<String>> store =
- IntegrationTestUtils.getStore(WINDOW_STORE_NAME,
kafkaStreams, QueryableStoreTypes.timestampedWindowStoreWithHeaders());
-
- if (store == null) {
- return false;
- }
-
- final long windowStart = timestamp - (timestamp %
WINDOW_SIZE_MS);
-
- final List<KeyValue<Windowed<String>,
ValueTimestampHeaders<String>>> results = new LinkedList<>();
- try (final KeyValueIterator<Windowed<String>,
ValueTimestampHeaders<String>> iterator = store.all()) {
- while (iterator.hasNext()) {
- final KeyValue<Windowed<String>,
ValueTimestampHeaders<String>> kv = iterator.next();
- if (kv.key.key().equals(key) &&
kv.key.window().start() == windowStart) {
- results.add(kv);
- }
- }
- }
+ produce(key, value, timestamp, headers);
- if (results.isEmpty()) {
- return false;
- }
-
- final ValueTimestampHeaders<String> result =
results.get(0).value;
+ awaitStore(WINDOW_STORE_NAME, QueryableStoreTypes.<String,
String>timestampedWindowStoreWithHeaders(),
Review Comment:
This differs from `processWindowedKeyValueWithHeadersAndVerify` only in the
expected timestamp (`-1L` vs `timestamp`). One method with an
`expectedTimestamp` parameter would cover both, like the key-value helpers
already do.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -135,6 +137,117 @@ private void buildAndStart(final StoreBuilder<?>
storeBuilder,
IntegrationTestUtils.startApplicationAndWaitUntilRunning(kafkaStreams);
}
+ /**
+ * Produces a single record (no explicit timestamp) into the input stream.
+ */
+ private <K, V> void produce(final K key, final V value) {
Review Comment:
These overloads are `<K, V>` but always serialize with `StringSerializer`,
so anything other than `String` fails at runtime. Typing the params as `String`
would turn that into a compile error.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -468,36 +528,16 @@ private <K, V> void
processKeyValueWithTimestampAndHeadersAndVerify(final K key,
final
Headers expectedHeaders)
throws Exception {
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
- inputStream,
- singletonList(KeyValue.pair(key, value)),
- TestUtils.producerConfig(CLUSTER.bootstrapServers(),
- StringSerializer.class,
- StringSerializer.class),
- headers,
- timestamp,
- false);
-
- TestUtils.waitForCondition(
- () -> {
- try {
- final ReadOnlyKeyValueStore<K, ValueTimestampHeaders<V>>
store = IntegrationTestUtils
- .getStore(STORE_NAME, kafkaStreams,
QueryableStoreTypes.timestampedKeyValueStoreWithHeaders());
-
- if (store == null)
- return false;
+ produce(key, value, timestamp, headers);
- final ValueTimestampHeaders<V> result = store.get(key);
- return result != null
- && result.value().equals(value)
- && result.timestamp() == timestamp
- && result.headers().equals(expectedHeaders);
- } catch (final Exception swallow) {
- LOG.error("Failed to retrieve expected result", swallow);
- return false;
- }
+ awaitStore(STORE_NAME, QueryableStoreTypes.<K,
V>timestampedKeyValueStoreWithHeaders(),
Review Comment:
This overload is the 6-arg one with `expectedTimestamp == timestamp` — have
it delegate instead of repeating the body.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -814,261 +822,115 @@ public void
shouldProxyTimestampedWindowStoreToTimestampedWindowStoreWithHeaders
private void processPlainWindowedKeyValueAndVerify(final String key,
final String value,
final long timestamp)
throws Exception {
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
- inputStream,
- List.of(KeyValue.pair(key, value)),
- TestUtils.producerConfig(CLUSTER.bootstrapServers(),
- StringSerializer.class,
- StringSerializer.class),
- timestamp,
- false);
-
- TestUtils.waitForCondition(() -> {
- try {
- final ReadOnlyWindowStore<String, String> store =
- IntegrationTestUtils.getStore(WINDOW_STORE_NAME,
kafkaStreams, QueryableStoreTypes.windowStore());
-
- if (store == null) {
- return false;
- }
-
- final long windowStart = timestamp - (timestamp %
WINDOW_SIZE_MS);
- final String result = store.fetch(key, windowStart);
+ produce(key, value, timestamp);
+ awaitStore(WINDOW_STORE_NAME, QueryableStoreTypes.<String,
String>windowStore(),
+ store -> {
+ final String result = store.fetch(key, windowStart(timestamp));
return result != null && result.equals(value);
- } catch (final Exception e) {
- return false;
- }
- }, 60_000L, "Could not verify plain window value in time.");
+ },
+ "Could not verify plain window value in time.");
}
private void verifyPlainWindowValueWithEmptyHeadersAndTimestamp(final
String key,
final
String value,
final long
windowTimestamp,
final long
expectedTimestamp) throws Exception {
- TestUtils.waitForCondition(() -> {
- try {
- final ReadOnlyWindowStore<String,
ValueTimestampHeaders<String>> store =
- IntegrationTestUtils.getStore(WINDOW_STORE_NAME,
kafkaStreams, QueryableStoreTypes.timestampedWindowStoreWithHeaders());
-
- if (store == null) {
- return false;
- }
-
- final long windowStart = windowTimestamp - (windowTimestamp %
WINDOW_SIZE_MS);
-
- final List<KeyValue<Windowed<String>,
ValueTimestampHeaders<String>>> results = new LinkedList<>();
- try (final KeyValueIterator<Windowed<String>,
ValueTimestampHeaders<String>> iterator = store.all()) {
- while (iterator.hasNext()) {
- final KeyValue<Windowed<String>,
ValueTimestampHeaders<String>> kv = iterator.next();
- if (kv.key.key().equals(key) &&
kv.key.window().start() == windowStart) {
- results.add(kv);
- }
- }
- }
-
- if (results.isEmpty()) {
+ awaitStore(WINDOW_STORE_NAME, QueryableStoreTypes.<String,
String>timestampedWindowStoreWithHeaders(),
+ store -> {
+ final Optional<ValueTimestampHeaders<String>> result =
findWindowedValue(store, key, windowTimestamp);
+ if (result.isEmpty()) {
return false;
}
- final ValueTimestampHeaders<String> result =
results.get(0).value;
- assertNotNull(result, "Result should not be null");
- assertEquals(value, result.value(), "Value should match");
- assertEquals(expectedTimestamp, result.timestamp(), "Timestamp
should be " + expectedTimestamp + " for plain store migration");
+ final ValueTimestampHeaders<String> value0 = result.get();
Review Comment:
`value0` reads like a leftover — `stored` or `actual` would say what it is.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -814,261 +822,115 @@ public void
shouldProxyTimestampedWindowStoreToTimestampedWindowStoreWithHeaders
private void processPlainWindowedKeyValueAndVerify(final String key,
final String value,
final long timestamp)
throws Exception {
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
- inputStream,
- List.of(KeyValue.pair(key, value)),
- TestUtils.producerConfig(CLUSTER.bootstrapServers(),
- StringSerializer.class,
- StringSerializer.class),
- timestamp,
- false);
-
- TestUtils.waitForCondition(() -> {
- try {
- final ReadOnlyWindowStore<String, String> store =
- IntegrationTestUtils.getStore(WINDOW_STORE_NAME,
kafkaStreams, QueryableStoreTypes.windowStore());
-
- if (store == null) {
- return false;
- }
-
- final long windowStart = timestamp - (timestamp %
WINDOW_SIZE_MS);
- final String result = store.fetch(key, windowStart);
+ produce(key, value, timestamp);
+ awaitStore(WINDOW_STORE_NAME, QueryableStoreTypes.<String,
String>windowStore(),
+ store -> {
+ final String result = store.fetch(key, windowStart(timestamp));
return result != null && result.equals(value);
- } catch (final Exception e) {
- return false;
- }
- }, 60_000L, "Could not verify plain window value in time.");
+ },
+ "Could not verify plain window value in time.");
}
private void verifyPlainWindowValueWithEmptyHeadersAndTimestamp(final
String key,
final
String value,
final long
windowTimestamp,
final long
expectedTimestamp) throws Exception {
- TestUtils.waitForCondition(() -> {
- try {
- final ReadOnlyWindowStore<String,
ValueTimestampHeaders<String>> store =
- IntegrationTestUtils.getStore(WINDOW_STORE_NAME,
kafkaStreams, QueryableStoreTypes.timestampedWindowStoreWithHeaders());
-
- if (store == null) {
- return false;
- }
-
- final long windowStart = windowTimestamp - (windowTimestamp %
WINDOW_SIZE_MS);
-
- final List<KeyValue<Windowed<String>,
ValueTimestampHeaders<String>>> results = new LinkedList<>();
- try (final KeyValueIterator<Windowed<String>,
ValueTimestampHeaders<String>> iterator = store.all()) {
- while (iterator.hasNext()) {
- final KeyValue<Windowed<String>,
ValueTimestampHeaders<String>> kv = iterator.next();
- if (kv.key.key().equals(key) &&
kv.key.window().start() == windowStart) {
- results.add(kv);
- }
- }
- }
-
- if (results.isEmpty()) {
+ awaitStore(WINDOW_STORE_NAME, QueryableStoreTypes.<String,
String>timestampedWindowStoreWithHeaders(),
+ store -> {
+ final Optional<ValueTimestampHeaders<String>> result =
findWindowedValue(store, key, windowTimestamp);
+ if (result.isEmpty()) {
return false;
}
- final ValueTimestampHeaders<String> result =
results.get(0).value;
- assertNotNull(result, "Result should not be null");
- assertEquals(value, result.value(), "Value should match");
- assertEquals(expectedTimestamp, result.timestamp(), "Timestamp
should be " + expectedTimestamp + " for plain store migration");
+ final ValueTimestampHeaders<String> value0 = result.get();
+ assertNotNull(value0, "Result should not be null");
+ assertEquals(value, value0.value(), "Value should match");
+ assertEquals(expectedTimestamp, value0.timestamp(), "Timestamp
should be " + expectedTimestamp + " for plain store migration");
// Verify headers exist but are empty (migrated from plain
store without headers or timestamps)
- assertNotNull(result.headers(), "Headers should not be null
for migrated data");
- assertEquals(0, result.headers().toArray().length, "Headers
should be empty for migrated data");
+ assertNotNull(value0.headers(), "Headers should not be null
for migrated data");
+ assertEquals(0, value0.headers().toArray().length, "Headers
should be empty for migrated data");
return true;
- } catch (final Exception e) {
- LOG.error("Error while verifying plain window value with empty
headers and timestamp", e);
- return false;
- }
- }, 60_000L, "Could not verify plain window value with empty headers
and timestamp in time.");
+ },
+ "Could not verify plain window value with empty headers and
timestamp in time.");
}
private void processPlainWindowedKeyValueWithHeadersAndVerify(final String
key,
final String
value,
final long
timestamp,
final
Headers headers,
final
Headers expectedHeaders) throws Exception {
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
- inputStream,
- List.of(KeyValue.pair(key, value)),
- TestUtils.producerConfig(CLUSTER.bootstrapServers(),
- StringSerializer.class,
- StringSerializer.class),
- headers,
- timestamp,
- false);
-
- TestUtils.waitForCondition(() -> {
- try {
- final ReadOnlyWindowStore<String,
ValueTimestampHeaders<String>> store =
- IntegrationTestUtils.getStore(WINDOW_STORE_NAME,
kafkaStreams, QueryableStoreTypes.timestampedWindowStoreWithHeaders());
-
- if (store == null) {
- return false;
- }
-
- final long windowStart = timestamp - (timestamp %
WINDOW_SIZE_MS);
-
- final List<KeyValue<Windowed<String>,
ValueTimestampHeaders<String>>> results = new LinkedList<>();
- try (final KeyValueIterator<Windowed<String>,
ValueTimestampHeaders<String>> iterator = store.all()) {
- while (iterator.hasNext()) {
- final KeyValue<Windowed<String>,
ValueTimestampHeaders<String>> kv = iterator.next();
- if (kv.key.key().equals(key) &&
kv.key.window().start() == windowStart) {
- results.add(kv);
- }
- }
- }
+ produce(key, value, timestamp, headers);
- if (results.isEmpty()) {
- return false;
- }
-
- final ValueTimestampHeaders<String> result =
results.get(0).value;
+ awaitStore(WINDOW_STORE_NAME, QueryableStoreTypes.<String,
String>timestampedWindowStoreWithHeaders(),
+ store -> {
+ final Optional<ValueTimestampHeaders<String>> result =
findWindowedValue(store, key, timestamp);
// For plain window stores, timestamp is always -1 since it's
not preserved
- return result != null
- && result.value().equals(value)
- && result.timestamp() == -1L
- && result.headers().equals(expectedHeaders);
- } catch (final Exception e) {
- e.printStackTrace();
- return false;
- }
- }, 60_000L, "Could not verify plain windowed value with headers in
time.");
+ return result.isPresent()
+ && result.get().value().equals(value)
+ && result.get().timestamp() == -1L
+ && result.get().headers().equals(expectedHeaders);
+ },
+ "Could not verify plain windowed value with headers in time.");
}
private void processWindowedKeyValueAndVerifyTimestamped(final String key,
final String
value,
final long
timestamp) throws Exception {
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
- inputStream,
- singletonList(KeyValue.pair(key, value)),
- TestUtils.producerConfig(CLUSTER.bootstrapServers(),
- StringSerializer.class,
- StringSerializer.class),
- timestamp,
- false);
-
- TestUtils.waitForCondition(() -> {
- try {
- final ReadOnlyWindowStore<String, ValueAndTimestamp<String>>
store =
- IntegrationTestUtils.getStore(WINDOW_STORE_NAME,
kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
-
- if (store == null) {
- return false;
- }
-
- final long windowStart = timestamp - (timestamp %
WINDOW_SIZE_MS);
- final ValueAndTimestamp<String> result = store.fetch(key,
windowStart);
+ produce(key, value, timestamp);
+ awaitStore(WINDOW_STORE_NAME, QueryableStoreTypes.<String,
String>timestampedWindowStore(),
+ store -> {
+ final ValueAndTimestamp<String> result = store.fetch(key,
windowStart(timestamp));
return result != null
&& result.value().equals(value)
&& result.timestamp() == timestamp;
- } catch (final Exception e) {
- return false;
- }
- }, 60_000L, "Could not verify timestamped value in time.");
+ },
+ "Could not verify timestamped value in time.");
}
private void processWindowedKeyValueWithHeadersAndVerify(final String key,
final String
value,
final long
timestamp,
final Headers
headers,
final Headers
expectedHeaders) throws Exception {
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
- inputStream,
- singletonList(KeyValue.pair(key, value)),
- TestUtils.producerConfig(CLUSTER.bootstrapServers(),
- StringSerializer.class,
- StringSerializer.class),
- headers,
- timestamp,
- false);
-
- TestUtils.waitForCondition(() -> {
- try {
- final ReadOnlyWindowStore<String,
ValueTimestampHeaders<String>> store =
- IntegrationTestUtils.getStore(WINDOW_STORE_NAME,
kafkaStreams, QueryableStoreTypes.timestampedWindowStoreWithHeaders());
-
- if (store == null) {
- return false;
- }
-
- final long windowStart = timestamp - (timestamp %
WINDOW_SIZE_MS);
-
- final List<KeyValue<Windowed<String>,
ValueTimestampHeaders<String>>> results = new LinkedList<>();
- try (final KeyValueIterator<Windowed<String>,
ValueTimestampHeaders<String>> iterator = store.all()) {
- while (iterator.hasNext()) {
- final KeyValue<Windowed<String>,
ValueTimestampHeaders<String>> kv = iterator.next();
- if (kv.key.key().equals(key) &&
kv.key.window().start() == windowStart) {
- results.add(kv);
- }
- }
- }
-
- if (results.isEmpty()) {
- return false;
- }
-
- final ValueTimestampHeaders<String> result =
results.get(0).value;
- return result != null
- && result.value().equals(value)
- && result.timestamp() == timestamp
- && result.headers().equals(expectedHeaders);
- } catch (final Exception e) {
- LOG.error("Error while verifying windowed value with headers",
e);
- return false;
- }
- }, 60_000L, "Could not verify windowed value with headers in time.");
+ produce(key, value, timestamp, headers);
+
+ awaitStore(WINDOW_STORE_NAME, QueryableStoreTypes.<String,
String>timestampedWindowStoreWithHeaders(),
+ store -> {
+ final Optional<ValueTimestampHeaders<String>> result =
findWindowedValue(store, key, timestamp);
+ return result.isPresent()
+ && result.get().value().equals(value)
+ && result.get().timestamp() == timestamp
+ && result.get().headers().equals(expectedHeaders);
+ },
+ "Could not verify windowed value with headers in time.");
}
private void verifyWindowValueWithEmptyHeaders(final String key,
final String value,
final long timestamp)
throws Exception {
- TestUtils.waitForCondition(() -> {
- try {
- final ReadOnlyWindowStore<String,
ValueTimestampHeaders<String>> store =
- IntegrationTestUtils.getStore(WINDOW_STORE_NAME,
kafkaStreams, QueryableStoreTypes.timestampedWindowStoreWithHeaders());
-
- if (store == null) {
+ awaitStore(WINDOW_STORE_NAME, QueryableStoreTypes.<String,
String>timestampedWindowStoreWithHeaders(),
Review Comment:
This is now the same as
`verifyPlainWindowValueWithEmptyHeadersAndTimestamp(key, value, timestamp,
timestamp)` — drop it and call that.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -1586,24 +1429,10 @@ public void
shouldSuccessfullyDowngradeFromSessionStoreWithHeadersToSessionStore
private void processSessionKeyValueAndVerify(final String key,
final String value,
final long timestamp) throws
Exception {
- IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
- inputStream,
- singletonList(KeyValue.pair(key, value)),
- TestUtils.producerConfig(CLUSTER.bootstrapServers(),
- StringSerializer.class,
- StringSerializer.class),
- timestamp,
- false);
-
- TestUtils.waitForCondition(() -> {
- try {
- final ReadOnlySessionStore<String, String> store =
- IntegrationTestUtils.getStore(SESSION_STORE_NAME,
kafkaStreams, QueryableStoreTypes.sessionStore());
-
- if (store == null) {
- return false;
- }
+ produce(key, value, timestamp);
+ awaitStore(SESSION_STORE_NAME, QueryableStoreTypes.<String,
String>sessionStore(),
+ store -> {
try (final KeyValueIterator<Windowed<String>, String> iterator
= store.fetch(key)) {
Review Comment:
This ignores `timestamp` and matches on key and value only, so it passes
even if the session window is wrong. The other session verifiers check the
window start — worth doing the same here.
--
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]
