muralibasani commented on code in PR #22901:
URL: https://github.com/apache/kafka/pull/22901#discussion_r3722820573
##########
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:
Good catch. restored it. findSessionValue now matches on both boundaries.
--
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]
