aliehsaeedii commented on code in PR #22901:
URL: https://github.com/apache/kafka/pull/22901#discussion_r3813387025
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -814,261 +807,91 @@ 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()) {
- 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");
-
- // 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");
-
- 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.");
+ /**
+ * Verifies the value stored for {@code key} in the window that {@code
windowTimestamp} falls into,
+ * expecting {@code expectedTimestamp} and empty headers (i.e. a store
migrated without headers).
+ */
+ private void verifyWindowValue(final String key,
+ final String value,
+ final long windowTimestamp,
+ final long expectedTimestamp) throws
Exception {
+ verifyWindowValue(key, value, windowTimestamp, expectedTimestamp, new
RecordHeaders());
}
- 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()) {
+ /**
+ * Verifies the value stored for {@code key} in the window that {@code
windowTimestamp} falls into,
+ * expecting {@code expectedTimestamp} and {@code expectedHeaders} in the
store. Pass an empty
+ * {@link RecordHeaders} for stores migrated without headers.
+ */
+ private void verifyWindowValue(final String key,
+ final String value,
+ final long windowTimestamp,
+ final long expectedTimestamp,
+ final Headers expectedHeaders) throws
Exception {
+ 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;
- // 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.");
+ final ValueTimestampHeaders<String> actual =
result.get().value;
+ assertNotNull(actual, "Stored value should not be null");
Review Comment:
This `assertNotNull(actual, ...)` can't fire: after the `isEmpty()` guard
`actual` is the matched entry's value, and the store never yields a null-valued
entry (a null inner value would trip the `assertEquals` below, not this). It's
the dead assert we agreed to drop earlier — back here and in
`verifySessionValueWithEmptyHeaders` (line 1396).
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -1586,146 +1371,57 @@ 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);
- try (final KeyValueIterator<Windowed<String>, String> iterator
= store.fetch(key)) {
- while (iterator.hasNext()) {
- final KeyValue<Windowed<String>, String> kv =
iterator.next();
- if (kv.key.key().equals(key) &&
kv.value.equals(value)) {
- return true;
- }
- }
- }
- return false;
- } catch (final Exception e) {
- return false;
- }
- }, 60_000L, "Could not verify session value in time.");
+ awaitStore(SESSION_STORE_NAME, QueryableStoreTypes.<String,
String>sessionStore(),
+ store -> {
+ final Optional<KeyValue<Windowed<String>, String>> result =
findSessionValue(store, key, timestamp);
+ return result.isPresent() && value.equals(result.get().value);
+ },
+ "Could not verify session value in time.");
}
private void verifySessionValueWithEmptyHeaders(final String key,
final String value,
final long timestamp)
throws Exception {
- TestUtils.waitForCondition(() -> {
- try {
- final ReadOnlySessionStore<String,
AggregationWithHeaders<String>> store =
- IntegrationTestUtils.getStore(SESSION_STORE_NAME,
kafkaStreams, QueryableStoreTypes.sessionStoreWithHeaders());
-
- if (store == null) {
+ awaitStore(SESSION_STORE_NAME, QueryableStoreTypes.<String,
String>sessionStoreWithHeaders(),
Review Comment:
The window side now shares one `verifyWindowValue(..., expectedHeaders)`,
but the session side still hand-rolls the find-and-check twice — here and in
`processSessionKeyValueWithHeadersAndVerify`. A `verifySessionValue(key, value,
timestamp, expectedHeaders)` would fold both together the same way.
--
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]
