Jess668 commented on code in PR #22666:
URL: https://github.com/apache/kafka/pull/22666#discussion_r3492486366
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilderWithHeadersTest.java:
##########
@@ -228,356 +230,142 @@ public void
shouldWrapPlainKeyValueStoreAsHeadersStore() {
assertInstanceOf(PlainToHeadersStoreAdapter.class,
((WrappedStateStore) store).wrapped());
}
- @Test
- public void shouldHandleKeyQueryOnInMemoryStore() {
- when(supplier.name()).thenReturn("test-store");
- when(supplier.metricsScope()).thenReturn("in-memory");
- when(supplier.get()).thenReturn(new
InMemoryKeyValueStore("test-store"));
-
- builder = new TimestampedKeyValueStoreBuilderWithHeaders<>(
- supplier,
- Serdes.String(),
- Serdes.String(),
- new MockTime()
- );
-
- final TimestampedKeyValueStoreWithHeaders<String, String> store =
builder
- .withLoggingDisabled()
- .withCachingDisabled()
- .build();
-
- final File dir = TestUtils.tempDirectory();
- final Properties props = StreamsTestUtils.getStreamsConfig();
- final InternalMockProcessorContext<String, String> context = new
InternalMockProcessorContext<>(
- dir,
- Serdes.String(),
- Serdes.String(),
- new StreamsConfig(props)
- );
- store.init(context, store);
-
- try {
- // Put data into the store
- final Headers headers = new RecordHeaders();
- headers.add("key1", "value1".getBytes());
- store.put("test-key", ValueTimestampHeaders.make("test-value",
12345L, headers));
-
- // Verify wrapper type for InMemoryKeyValueStore
- final StateStore wrapped = ((WrappedStateStore) store).wrapped();
- assertInstanceOf(HeadersBytesStore.class, wrapped,
- "Expected wrapper to implement HeadersBytesStore for
InMemoryKeyValueStore");
-
- // Query at typed level - KeyQuery should return just the value
- final KeyQuery<String, String> query =
KeyQuery.withKey("test-key");
- final QueryResult<String> result = store.query(query,
PositionBound.unbounded(), new QueryConfig(false));
-
- // Verify IQv2 query result
- assertTrue(result.isSuccess(), "Expected query to succeed on
InMemoryKeyValueStore");
- assertNotNull(result.getPosition(), "Expected position to be set");
- assertInstanceOf(String.class, result.getResult());
- assertEquals("test-value", result.getResult(), "KeyQuery should
return just the value");
- } finally {
- store.close();
+ //
----------------------------------------------------------------------------------------------
+ // IQv2 query handling for the built header store.
+ //
+ // The header store can be built three ways; each is exercised with
caching on and off through a
+ // single helper that builds the real store chain (Metered -> [Caching] ->
inner) with a real
+ // record cache, then writes and reads real data at the typed (metered)
level:
+ // NATIVE -> RocksDBTimestampedStoreWithHeaders
(persists headers)
+ // ADAPTER -> RocksDBTimestampedStore via
TimestampedToHeadersStoreAdapter (drops headers)
+ // IN_MEMORY -> InMemoryKeyValueStore via the in-memory headers marker
+ //
----------------------------------------------------------------------------------------------
+
+ private enum StoreType { NATIVE, ADAPTER, IN_MEMORY }
+
+ private KeyValueStore<Bytes, byte[]> innerStore(final StoreType storeType)
{
+ switch (storeType) {
+ case NATIVE: return new
RocksDBTimestampedStoreWithHeaders("test-store", "metrics-scope");
+ case ADAPTER: return new RocksDBTimestampedStore("test-store",
"metrics-scope");
+ case IN_MEMORY: return new InMemoryKeyValueStore("test-store");
+ default: throw new IllegalArgumentException("unknown store
type: " + storeType);
}
}
- @Test
- public void shouldHandleTimestampedKeyQueryOnInMemoryStore() {
- when(supplier.name()).thenReturn("test-store");
- when(supplier.metricsScope()).thenReturn("in-memory");
- when(supplier.get()).thenReturn(new
InMemoryKeyValueStore("test-store"));
+ private TimestampedKeyValueStoreWithHeaders<String, String>
buildAndInitStore(
+ final StoreType storeType,
+ final boolean cachingEnabled) {
+ lenient().when(supplier.name()).thenReturn("test-store");
+ lenient().when(supplier.metricsScope()).thenReturn("metricScope");
+ lenient().when(supplier.get()).thenReturn(innerStore(storeType));
builder = new TimestampedKeyValueStoreBuilderWithHeaders<>(
- supplier,
- Serdes.String(),
- Serdes.String(),
- new MockTime()
- );
-
- final TimestampedKeyValueStoreWithHeaders<String, String> store =
builder
- .withLoggingDisabled()
- .withCachingDisabled()
- .build();
+ supplier, Serdes.String(), Serdes.String(), new MockTime());
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
+ (cachingEnabled
+ ? builder.withLoggingDisabled().withCachingEnabled()
+ : builder.withLoggingDisabled().withCachingDisabled())
+ .build();
- final File dir = TestUtils.tempDirectory();
- final Properties props = StreamsTestUtils.getStreamsConfig();
+ final ThreadCache cache = new ThreadCache(
+ new LogContext("test "),
+ cachingEnabled ? 10 * 1024 * 1024 : 0,
+ new MockStreamsMetrics(new Metrics()));
final InternalMockProcessorContext<String, String> context = new
InternalMockProcessorContext<>(
- dir,
- Serdes.String(),
- Serdes.String(),
- new StreamsConfig(props)
- );
+ TestUtils.tempDirectory(), Serdes.String(), Serdes.String(), null,
cache);
+ context.setRecordContext(new ProcessorRecordContext(0L, 0L, 0,
"topic", new RecordHeaders()));
store.init(context, store);
-
- try {
- // Put data into the store
- final Headers headers = new RecordHeaders();
- headers.add("key1", "value1".getBytes());
- store.put("test-key", ValueTimestampHeaders.make("test-value",
12345L, headers));
-
- // Verify wrapper type for InMemoryKeyValueStore
- final StateStore wrapped = ((WrappedStateStore) store).wrapped();
- assertInstanceOf(HeadersBytesStore.class, wrapped,
- "Expected wrapper to implement HeadersBytesStore for
InMemoryKeyValueStore");
-
- // Query at typed level - TimestampedKeyQuery should return value
+ timestamp
- final TimestampedKeyQuery<String, String> query =
TimestampedKeyQuery.withKey("test-key");
- final QueryResult<ValueAndTimestamp<String>> result =
store.query(query, PositionBound.unbounded(), new QueryConfig(false));
-
- // Verify IQv2 query result
- assertTrue(result.isSuccess(), "Expected query to succeed on
InMemoryKeyValueStore");
- assertNotNull(result.getPosition(), "Expected position to be set");
- assertNotNull(result.getResult(), "Expected non-null result");
- assertInstanceOf(ValueAndTimestamp.class, result.getResult());
- assertEquals("test-value", result.getResult().value(),
"TimestampedKeyQuery should return the value");
- assertEquals(12345L, result.getResult().timestamp(),
"TimestampedKeyQuery should return the timestamp");
- } finally {
- store.close();
- }
+ return store;
}
- @Test
- public void shouldReturnPositionFromHeadersStore() {
- when(supplier.name()).thenReturn("test-store");
- when(supplier.metricsScope()).thenReturn("metricScope");
- when(supplier.get()).thenReturn(new
RocksDBTimestampedStoreWithHeaders("test-store", "metrics-scope"));
-
- builder = new TimestampedKeyValueStoreBuilderWithHeaders<>(
- supplier,
- Serdes.String(),
- Serdes.String(),
- new MockTime()
- );
-
- final TimestampedKeyValueStoreWithHeaders<String, String> store =
builder
- .withLoggingDisabled()
- .withCachingDisabled()
- .build();
-
- final File dir = TestUtils.tempDirectory();
- final Properties props = StreamsTestUtils.getStreamsConfig();
- final InternalMockProcessorContext<String, String> context = new
InternalMockProcessorContext<>(
- dir,
- Serdes.String(),
- Serdes.String(),
- new StreamsConfig(props)
- );
- store.init(context, store);
-
- try {
- final StateStore wrapped = ((WrappedStateStore) store).wrapped();
- final Position position = wrapped.getPosition();
-
- // Verify: Position is returned (should be non-null)
- assertNotNull(position, "Expected non-null position");
- assertTrue(position.isEmpty(), "Expected position to be empty
initially");
- } finally {
- store.close();
- }
+ private static Headers headersWith(final String key, final String value) {
+ return new RecordHeaders().add(key,
value.getBytes(StandardCharsets.UTF_8));
}
- @Test
- public void shouldReturnPositionFromAdaptedTimestampedStore() {
- when(supplier.name()).thenReturn("test-store");
- when(supplier.metricsScope()).thenReturn("metricScope");
- when(supplier.get()).thenReturn(new
RocksDBTimestampedStore("test-store", "metrics-scope"));
-
- builder = new TimestampedKeyValueStoreBuilderWithHeaders<>(
- supplier,
- Serdes.String(),
- Serdes.String(),
- new MockTime()
- );
-
- final TimestampedKeyValueStoreWithHeaders<String, String> store =
builder
- .withLoggingDisabled()
- .withCachingDisabled()
- .build();
-
- final File dir = TestUtils.tempDirectory();
- final Properties props = StreamsTestUtils.getStreamsConfig();
- final InternalMockProcessorContext<String, String> context = new
InternalMockProcessorContext<>(
- dir,
- Serdes.String(),
- Serdes.String(),
- new StreamsConfig(props)
- );
- store.init(context, store);
-
+ @ParameterizedTest
+ @CsvSource({"NATIVE, true", "NATIVE, false", "ADAPTER, true", "ADAPTER,
false", "IN_MEMORY, true", "IN_MEMORY, false"})
+ public void shouldHandleKeyQuery(final StoreType storeType, final boolean
cachingEnabled) {
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(storeType, cachingEnabled);
try {
- // Verify adapter is used
- final StateStore wrapped = ((WrappedStateStore) store).wrapped();
- assertInstanceOf(TimestampedToHeadersStoreAdapter.class, wrapped);
+ store.put("k", ValueTimestampHeaders.make("v", 123L,
headersWith("h", "x")));
- // Get position from adapter (should delegate to underlying store)
- final Position position = wrapped.getPosition();
+ final QueryResult<String> result =
+ store.query(KeyQuery.withKey("k"), PositionBound.unbounded(),
new QueryConfig(false));
- assertNotNull(position, "Expected non-null position from adapter");
- assertTrue(position.isEmpty(), "Expected position to be empty
initially");
+ assertTrue(result.isSuccess(), "Expected KeyQuery to succeed");
+ assertEquals("v", result.getResult(), "KeyQuery returns the value
only");
+ assertNotNull(result.getPosition(), "Expected position to be set");
} finally {
store.close();
}
}
- @Test
- public void shouldReturnPositionFromInMemoryStore() {
- when(supplier.name()).thenReturn("test-store");
- when(supplier.metricsScope()).thenReturn("in-memory");
- when(supplier.get()).thenReturn(new
InMemoryKeyValueStore("test-store"));
-
- builder = new TimestampedKeyValueStoreBuilderWithHeaders<>(
- supplier,
- Serdes.String(),
- Serdes.String(),
- new MockTime()
- );
-
- final TimestampedKeyValueStoreWithHeaders<String, String> store =
builder
- .withLoggingDisabled()
- .withCachingDisabled()
- .build();
-
- final File dir = TestUtils.tempDirectory();
- final Properties props = StreamsTestUtils.getStreamsConfig();
- final InternalMockProcessorContext<String, String> context = new
InternalMockProcessorContext<>(
- dir,
- Serdes.String(),
- Serdes.String(),
- new StreamsConfig(props)
- );
- store.init(context, store);
-
+ @ParameterizedTest
+ @CsvSource({"NATIVE, true", "NATIVE, false", "ADAPTER, true", "ADAPTER,
false", "IN_MEMORY, true", "IN_MEMORY, false"})
+ public void shouldHandleTimestampedKeyQuery(final StoreType storeType,
final boolean cachingEnabled) {
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(storeType, cachingEnabled);
try {
- // Verify marker wrapper is used
- final StateStore wrapped = ((WrappedStateStore) store).wrapped();
- assertInstanceOf(HeadersBytesStore.class, wrapped);
+ store.put("k", ValueTimestampHeaders.make("v", 123L,
headersWith("h", "x")));
- // Get position from marker (should delegate to
InMemoryKeyValueStore)
- final Position position = wrapped.getPosition();
+ final QueryResult<ValueAndTimestamp<String>> result =
+ store.query(TimestampedKeyQuery.withKey("k"),
PositionBound.unbounded(), new QueryConfig(false));
- assertNotNull(position, "Expected non-null position from in-memory
store");
- assertTrue(position.isEmpty(), "Expected position to be empty
initially");
+ assertTrue(result.isSuccess(), "Expected TimestampedKeyQuery to
succeed");
+ assertEquals("v", result.getResult().value());
+ assertEquals(123L, result.getResult().timestamp());
+ assertNotNull(result.getPosition(), "Expected position to be set");
} finally {
store.close();
}
}
- @Test
- public void shouldMaintainPositionAcrossOperationsOnHeadersStore() {
- when(supplier.name()).thenReturn("test-store");
- when(supplier.metricsScope()).thenReturn("metricScope");
- when(supplier.get()).thenReturn(new
RocksDBTimestampedStoreWithHeaders("test-store", "metrics-scope"));
-
- builder = new TimestampedKeyValueStoreBuilderWithHeaders<>(
- supplier,
- Serdes.String(),
- Serdes.String(),
- new MockTime()
- );
-
- final TimestampedKeyValueStoreWithHeaders<String, String> store =
builder
- .withLoggingDisabled()
- .withCachingDisabled()
- .build();
-
- final File dir = TestUtils.tempDirectory();
- final Properties props = StreamsTestUtils.getStreamsConfig();
- final InternalMockProcessorContext<String, String> context = new
InternalMockProcessorContext<>(
- dir,
- Serdes.String(),
- Serdes.String(),
- new StreamsConfig(props)
- );
- store.init(context, store);
-
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void shouldHandleTimestampedKeyWithHeadersQueryOnNativeStore(final
boolean cachingEnabled) {
Review Comment:
Thanks, I agree on closing the gap. After taking a second look, I found that
the in-memory implementation doesn’t drop headers. Non-persistent stores are
wrapped by `InMemoryTimestampedKeyValueStoreWithHeadersMarker`, which is just a
pass-through. Since the metered layer serializes the headers into the value
bytes, the in-memory store round-trips them the same way as the native store.
So I’ll add an `IN_MEMORY` case asserting that the headers come back, just
like `NATIVE`.
--
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]
