vvcephei commented on a change in pull request #11624:
URL: https://github.com/apache/kafka/pull/11624#discussion_r778456441
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java
##########
@@ -365,109 +395,200 @@ public void beforeTest() {
final Properties streamsConfig = streamsConfiguration(
cache,
log,
- storeToTest.name()
+ storeToTest.name(),
+ kind
);
final StreamsBuilder builder = new StreamsBuilder();
- if (supplier instanceof KeyValueBytesStoreSupplier) {
- final Materialized<Integer, Integer, KeyValueStore<Bytes, byte[]>>
materialized =
- Materialized.as((KeyValueBytesStoreSupplier) supplier);
+ if (Objects.equals(kind, "DSL") && supplier instanceof
KeyValueBytesStoreSupplier) {
+ setUpKeyValueDSLTopology((KeyValueBytesStoreSupplier) supplier,
builder);
+ } else if (Objects.equals(kind, "PAPI") && supplier instanceof
KeyValueBytesStoreSupplier) {
+ setUpKeyValuePAPITopology((KeyValueBytesStoreSupplier) supplier,
builder);
+ } else if (Objects.equals(kind, "DSL") && supplier instanceof
WindowBytesStoreSupplier) {
+ setUpWindowDSLTopology((WindowBytesStoreSupplier) supplier,
builder);
+ } else if (Objects.equals(kind, "PAPI") && supplier instanceof
WindowBytesStoreSupplier) {
+ throw new AssumptionViolatedException("Case not implemented yet");
+ } else if (Objects.equals(kind, "DSL") && supplier instanceof
SessionBytesStoreSupplier) {
+ setUpSessionDSLTopology((SessionBytesStoreSupplier) supplier,
builder);
+ } else if (Objects.equals(kind, "PAPI") && supplier instanceof
SessionBytesStoreSupplier) {
+ throw new AssumptionViolatedException("Case not implemented yet");
+ } else {
+ throw new AssertionError("Store supplier is an unrecognized
type.");
+ }
- if (cache) {
- materialized.withCachingEnabled();
- } else {
- materialized.withCachingDisabled();
- }
+ // Don't need to wait for running, since tests can use iqv2 to wait
until they
+ // get a valid response.
- if (log) {
- materialized.withLoggingEnabled(Collections.emptyMap());
- } else {
- materialized.withCachingDisabled();
- }
+ kafkaStreams =
+ IntegrationTestUtils.getStartedStreams(
+ streamsConfig,
+ builder,
+ true
+ );
+ }
- if (storeToTest.global()) {
- builder.globalTable(
- INPUT_TOPIC_NAME,
- Consumed.with(Serdes.Integer(), Serdes.Integer()),
- materialized
- );
- } else {
- builder.table(
- INPUT_TOPIC_NAME,
- Consumed.with(Serdes.Integer(), Serdes.Integer()),
- materialized
- );
- }
- } else if (supplier instanceof WindowBytesStoreSupplier) {
- final Materialized<Integer, Integer, WindowStore<Bytes, byte[]>>
materialized =
- Materialized.as((WindowBytesStoreSupplier) supplier);
+ private void setUpSessionDSLTopology(final SessionBytesStoreSupplier
supplier,
+ final StreamsBuilder builder) {
+ final Materialized<Integer, Integer, SessionStore<Bytes, byte[]>>
materialized =
+ Materialized.as(supplier);
- if (cache) {
- materialized.withCachingEnabled();
- } else {
- materialized.withCachingDisabled();
- }
+ if (cache) {
+ materialized.withCachingEnabled();
+ } else {
+ materialized.withCachingDisabled();
+ }
- if (log) {
- materialized.withLoggingEnabled(Collections.emptyMap());
- } else {
- materialized.withCachingDisabled();
- }
+ if (log) {
+ materialized.withLoggingEnabled(Collections.emptyMap());
+ } else {
+ materialized.withCachingDisabled();
Review comment:
Oops! Thanks for the catch!
--
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]
