mjsax commented on code in PR #22156:
URL: https://github.com/apache/kafka/pull/22156#discussion_r3479152900
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/ListValueStoreTest.java:
##########
@@ -45,37 +49,64 @@
public class ListValueStoreTest {
public enum StoreType { InMemory, RocksDB }
+ public enum ValueMode { Plain, Headers }
+
+ static Stream<Arguments> modes() {
+ return Stream.of(
+ Arguments.of(StoreType.InMemory, ValueMode.Plain),
+ Arguments.of(StoreType.InMemory, ValueMode.Headers),
+ Arguments.of(StoreType.RocksDB, ValueMode.Plain),
+ Arguments.of(StoreType.RocksDB, ValueMode.Headers)
+ );
+ }
- private KeyValueStore<Integer, String> listStore;
+ private StoreFacade facade;
final File baseDir = TestUtils.tempDirectory("test");
- public void setup(final StoreType storeType) {
- listStore = buildStore(Serdes.Integer(), Serdes.String(), storeType);
-
+ public void setup(final StoreType storeType, final ValueMode valueMode) {
final MockRecordCollector recordCollector = new MockRecordCollector();
- final InternalMockProcessorContext<Integer, String> context = new
InternalMockProcessorContext<>(
- baseDir,
- Serdes.String(),
- Serdes.Integer(),
- recordCollector,
- new ThreadCache(
- new LogContext("testCache"),
- 0,
- new MockStreamsMetrics(new Metrics())));
- context.setTime(1L);
-
- listStore.init(context, listStore);
+ final ThreadCache cache = new ThreadCache(
+ new LogContext("testCache"),
+ 0,
+ new MockStreamsMetrics(new Metrics()));
+
+ if (valueMode == ValueMode.Plain) {
+ final KeyValueStore<Integer, String> store =
buildPlainStore(Serdes.Integer(), Serdes.String(), storeType);
+ final InternalMockProcessorContext<Integer, String> context = new
InternalMockProcessorContext<>(
+ baseDir,
+ Serdes.String(),
+ Serdes.Integer(),
+ recordCollector,
+ cache);
+ context.setTime(1L);
+ store.init(context, store);
+ facade = new PlainFacade(store);
+ } else {
+ final KeyValueStore<Integer, ValueTimestampHeaders<String>> store =
+ buildHeadersStore(Serdes.Integer(), Serdes.String(),
storeType);
+ final InternalMockProcessorContext<Integer,
ValueTimestampHeaders<String>> context = new InternalMockProcessorContext<>(
+ baseDir,
+ Serdes.Integer(),
+ null,
Review Comment:
Might be worth to add a comment?
--
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]
