aliehsaeedii commented on code in PR #22901:
URL: https://github.com/apache/kafka/pull/22901#discussion_r3712790482
##########
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.
Keys and values are
+ * always {@link String}, matching the {@link StringSerializer} used below.
+ */
+ private void produce(final String key, final String 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 void produce(final String key, final String 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 void produce(final String key, final String 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 {
+ return condition.test(IntegrationTestUtils.getStore(storeName,
kafkaStreams, storeType));
+ } 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 entry stored for {@code key} in the window that {@code
timestamp} falls into, by
+ * scanning {@link ReadOnlyWindowStore#all()} and matching on key and
window start. Returns the
+ * matched {@link KeyValue} so an empty result means "no such entry" and
is not conflated with a
+ * matched entry that happens to have a {@code null} value — callers can
assert on the value.
+ */
+ private static Optional<KeyValue<Windowed<String>,
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.of(kv);
+ }
+ }
+ }
+ return Optional.empty();
+ }
+
+ /**
+ * Finds the entry stored for {@code key} in the session whose window
starts at {@code timestamp},
+ * by scanning {@link ReadOnlySessionStore#fetch(Object)}. Sessions in
this test are always
+ * created as {@code SessionWindow(ts, ts)}, so matching on window start
is equivalent to matching
+ * both start and end. Returns the matched {@link KeyValue} so an empty
result means "no such
+ * entry" and is not conflated with a matched entry that happens to have a
{@code null} value.
+ */
+ private static Optional<KeyValue<Windowed<String>,
AggregationWithHeaders<String>>> findSessionValue(
+ final ReadOnlySessionStore<String, AggregationWithHeaders<String>>
store,
+ final String key,
+ final long timestamp) {
+ try (final KeyValueIterator<Windowed<String>,
AggregationWithHeaders<String>> iterator = store.fetch(key)) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>,
AggregationWithHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() ==
timestamp) {
Review Comment:
The old session verifiers also required `kv.key.window().end() ==
timestamp`. Dropping it means a migration bug that mangles a session's end
boundary no longer fails the test — please add the end check back.
##########
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.
Keys and values are
+ * always {@link String}, matching the {@link StringSerializer} used below.
+ */
+ private void produce(final String key, final String 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 void produce(final String key, final String 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 void produce(final String key, final String 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.
Review Comment:
`waitForCondition` does catch `AssertionError` and retries until the timeout
(`TestUtils.retryOnExceptionWithTimeout`), so these helpers don't fail fast —
they retry for 60s and then rethrow the assert. The doc needs fixing.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -1586,146 +1359,66 @@ 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 hand-rolls the scan that `findSessionValue` already does. Making
`findSessionValue` generic in the value type would let this reuse it.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -814,261 +797,89 @@ 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<KeyValue<Windowed<String>,
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> actual =
result.get().value;
+ assertNotNull(actual, "Stored value should not be null");
+ assertEquals(value, actual.value(), "Value should match");
+ assertEquals(expectedTimestamp, actual.timestamp(), "Timestamp
should be " + expectedTimestamp + " for plain store migration");
Review Comment:
This says "for plain store migration", but the timestamped-store migration
and proxy tests now call this method too. Same for the `Plain` in the method
name — worth dropping both.
##########
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.
Keys and values are
+ * always {@link String}, matching the {@link StringSerializer} used below.
+ */
+ private void produce(final String key, final String 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 void produce(final String key, final String 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 void produce(final String key, final String 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 {
+ return condition.test(IntegrationTestUtils.getStore(storeName,
kafkaStreams, storeType));
+ } catch (final Exception swallow) {
+ LOG.error("Error while checking store result", swallow);
Review Comment:
Add `storeName` to this log line — now that every helper shares it, the log
no longer says which check failed.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -814,261 +797,89 @@ 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<KeyValue<Windowed<String>,
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> actual =
result.get().value;
+ assertNotNull(actual, "Stored value should not be null");
+ assertEquals(value, actual.value(), "Value should match");
+ assertEquals(expectedTimestamp, actual.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(actual.headers(), "Headers should not be null
for migrated data");
+ assertEquals(0, actual.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.");
- }
-
- 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);
- }
- }
- }
-
- if (results.isEmpty()) {
- return false;
- }
-
- final ValueTimestampHeaders<String> result =
results.get(0).value;
- // 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.");
+ },
+ "Could not verify plain window value with empty headers and
timestamp 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.");
+ processWindowedKeyValueWithHeadersAndVerify(key, value, timestamp,
timestamp, headers, expectedHeaders);
}
- 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) {
- 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;
- assertNotNull(result, "Result should not be null");
- assertEquals(value, result.value(), "Value should match");
- assertEquals(timestamp, result.timestamp(), "Timestamp should
match");
-
- // Verify headers exist but are empty (migrated from
timestamped store without headers)
- assertNotNull(result.headers(), "Headers should not be null
for migrated data");
- assertEquals(0, result.headers().toArray().length, "Headers
should be empty for migrated data");
-
- return true;
- } catch (final Exception e) {
- LOG.error("Error while verifying legacy value with empty
headers", e);
- return false;
- }
- }, 60_000L, "Could not verify legacy value with empty headers in
time.");
+ /**
+ * Produces a windowed record with headers and verifies the stored
value/headers, expecting
+ * {@code expectedTimestamp} in the store. For a plain window store (no
timestamp preserved)
+ * pass {@code expectedTimestamp == -1L}; otherwise pass the produced
{@code timestamp}.
+ */
+ private void processWindowedKeyValueWithHeadersAndVerify(final String key,
+ final String
value,
+ final long
timestamp,
+ final long
expectedTimestamp,
+ final Headers
headers,
+ final Headers
expectedHeaders) throws Exception {
+ produce(key, value, timestamp, headers);
+
+ awaitStore(WINDOW_STORE_NAME, QueryableStoreTypes.<String,
String>timestampedWindowStoreWithHeaders(),
Review Comment:
`verifyPlainWindowValueWithEmptyHeadersAndTimestamp` checks the same fields
with `expectedHeaders` fixed to empty. Giving that verifier an
`expectedHeaders` parameter and making this `produce(...)` plus a call to it
would keep the find-and-check logic in one place. Same duplication between
`verifyLegacyValuesWithEmptyHeaders` and the 6-arg
`processKeyValueWithTimestampAndHeadersAndVerify`.
--
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]
