You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/01/05 21:48:12 UTC
[GitHub] [kafka] vvcephei opened a new pull request #11650: MINOR: Add PAPI Window and Session store tests for IQv2
vvcephei opened a new pull request #11650:
URL: https://github.com/apache/kafka/pull/11650
During some recent reviews, @mjsax pointed out that StateStore layers
are constructed differently the stores are added via the PAPI vs. the DSL.
This PR adds PAPI construction for Window and Session stores to the
IQv2StoreIntegrationTest so that we can ensure IQv2 works on every
possible state store.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on a change in pull request #11650: KAFKA-13553: Add PAPI Window and Session store tests for IQv2
Posted by GitBox <gi...@apache.org>.
vvcephei commented on a change in pull request #11650:
URL: https://github.com/apache/kafka/pull/11650#discussion_r779188902
##########
File path: streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java
##########
@@ -626,6 +627,118 @@ public void process(final Record<Integer, Integer> record) {
}
+ private void setUpWindowPAPITopology(final WindowBytesStoreSupplier supplier,
+ final StreamsBuilder builder) {
+ final StoreBuilder<?> windowStoreStoreBuilder;
+ final ProcessorSupplier<Integer, Integer, Void, Void> processorSupplier;
+ if (storeToTest.timestamped()) {
+ windowStoreStoreBuilder = Stores.timestampedWindowStoreBuilder(
+ supplier,
+ Serdes.Integer(),
+ Serdes.Integer()
+ );
+ processorSupplier = () -> new ContextualProcessor<Integer, Integer, Void, Void>() {
+ @Override
+ public void process(final Record<Integer, Integer> record) {
+ final TimestampedWindowStore<Integer, Integer> stateStore =
+ context().getStateStore(windowStoreStoreBuilder.name());
+ stateStore.put(
+ record.key(),
+ ValueAndTimestamp.make(
+ record.value(), record.timestamp()
+ ),
+ WINDOW_START
+ );
+ }
+ };
+ } else {
+ windowStoreStoreBuilder = Stores.windowStoreBuilder(
+ supplier,
+ Serdes.Integer(),
+ Serdes.Integer()
+ );
+ processorSupplier =
+ () -> new ContextualProcessor<Integer, Integer, Void, Void>() {
+ @Override
+ public void process(final Record<Integer, Integer> record) {
+ final WindowStore<Integer, Integer> stateStore =
+ context().getStateStore(windowStoreStoreBuilder.name());
+ stateStore.put(record.key(), record.value(), WINDOW_START);
+ }
+ };
+ }
+ if (cache) {
+ windowStoreStoreBuilder.withCachingEnabled();
+ } else {
+ windowStoreStoreBuilder.withCachingDisabled();
+ }
+ if (log) {
+ windowStoreStoreBuilder.withLoggingEnabled(Collections.emptyMap());
+ } else {
+ windowStoreStoreBuilder.withCachingDisabled();
+ }
+ if (storeToTest.global()) {
+ builder.addGlobalStore(
+ windowStoreStoreBuilder,
+ INPUT_TOPIC_NAME,
+ Consumed.with(Serdes.Integer(), Serdes.Integer()),
+ processorSupplier
+ );
+ } else {
+ builder.addStateStore(windowStoreStoreBuilder);
+ builder
+ .stream(INPUT_TOPIC_NAME, Consumed.with(Serdes.Integer(), Serdes.Integer()))
+ .process(processorSupplier, windowStoreStoreBuilder.name());
+ }
+
+ }
+
+ private void setUpSessionPAPITopology(final SessionBytesStoreSupplier supplier,
+ final StreamsBuilder builder) {
+ final StoreBuilder<?> sessionStoreStoreBuilder;
+ final ProcessorSupplier<Integer, Integer, Void, Void> processorSupplier;
+ sessionStoreStoreBuilder = Stores.sessionStoreBuilder(
+ supplier,
+ Serdes.Integer(),
+ Serdes.Integer()
+ );
+ processorSupplier = () -> new ContextualProcessor<Integer, Integer, Void, Void>() {
+ @Override
+ public void process(final Record<Integer, Integer> record) {
+ final SessionStore<Integer, Integer> stateStore =
+ context().getStateStore(sessionStoreStoreBuilder.name());
+ stateStore.put(
+ new Windowed<>(record.key(), new SessionWindow(WINDOW_START, WINDOW_START)),
+ record.value()
+ );
+ }
+ };
+ if (cache) {
+ sessionStoreStoreBuilder.withCachingEnabled();
+ } else {
+ sessionStoreStoreBuilder.withCachingDisabled();
+ }
+ if (log) {
+ sessionStoreStoreBuilder.withLoggingEnabled(Collections.emptyMap());
+ } else {
+ sessionStoreStoreBuilder.withCachingDisabled();
Review comment:
Huh, it looks like I missed updating the PAPI method in the other PR, and then amplified it here when I copy/pasted it for the other stores. Sorry about that!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on a change in pull request #11650: MINOR: Add PAPI Window and Session store tests for IQv2
Posted by GitBox <gi...@apache.org>.
vvcephei commented on a change in pull request #11650:
URL: https://github.com/apache/kafka/pull/11650#discussion_r779160412
##########
File path: streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java
##########
@@ -800,13 +914,13 @@ public void verifyStore() {
throw new AssertionError(queryResult.toString());
}
assertThat(partitionResult.getFailureReason(), is(FailureReason.UNKNOWN_QUERY_TYPE));
- assertThat(partitionResult.getFailureMessage(), is(
+ assertThat(partitionResult.getFailureMessage(), matchesPattern(
"This store"
- + " (class org.apache.kafka.streams.state.internals.MeteredTimestampedWindowStore)"
+ + " \\(class org.apache.kafka.streams.state.internals.Metered.*WindowStore\\)"
Review comment:
Regardless of whether you supply a bytes store to the DSL, you wind up wrapped in a Timestamped store. Now that we're registering the non-timestamped store directly with the PAPI, we can also get a non-timestamped metered store, hence swapping this for a regex that matches both `MeteredWindowStore` and `MeteredTimestampedWindowStore`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei merged pull request #11650: KAFKA-13553: Add PAPI Window and Session store tests for IQv2
Posted by GitBox <gi...@apache.org>.
vvcephei merged pull request #11650:
URL: https://github.com/apache/kafka/pull/11650
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a change in pull request #11650: KAFKA-13553: Add PAPI Window and Session store tests for IQv2
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #11650:
URL: https://github.com/apache/kafka/pull/11650#discussion_r779180990
##########
File path: streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java
##########
@@ -626,6 +627,118 @@ public void process(final Record<Integer, Integer> record) {
}
+ private void setUpWindowPAPITopology(final WindowBytesStoreSupplier supplier,
+ final StreamsBuilder builder) {
+ final StoreBuilder<?> windowStoreStoreBuilder;
+ final ProcessorSupplier<Integer, Integer, Void, Void> processorSupplier;
+ if (storeToTest.timestamped()) {
+ windowStoreStoreBuilder = Stores.timestampedWindowStoreBuilder(
+ supplier,
+ Serdes.Integer(),
+ Serdes.Integer()
+ );
+ processorSupplier = () -> new ContextualProcessor<Integer, Integer, Void, Void>() {
+ @Override
+ public void process(final Record<Integer, Integer> record) {
+ final TimestampedWindowStore<Integer, Integer> stateStore =
+ context().getStateStore(windowStoreStoreBuilder.name());
+ stateStore.put(
+ record.key(),
+ ValueAndTimestamp.make(
+ record.value(), record.timestamp()
+ ),
+ WINDOW_START
+ );
+ }
+ };
+ } else {
+ windowStoreStoreBuilder = Stores.windowStoreBuilder(
+ supplier,
+ Serdes.Integer(),
+ Serdes.Integer()
+ );
+ processorSupplier =
+ () -> new ContextualProcessor<Integer, Integer, Void, Void>() {
+ @Override
+ public void process(final Record<Integer, Integer> record) {
+ final WindowStore<Integer, Integer> stateStore =
+ context().getStateStore(windowStoreStoreBuilder.name());
+ stateStore.put(record.key(), record.value(), WINDOW_START);
+ }
+ };
+ }
+ if (cache) {
+ windowStoreStoreBuilder.withCachingEnabled();
+ } else {
+ windowStoreStoreBuilder.withCachingDisabled();
+ }
+ if (log) {
+ windowStoreStoreBuilder.withLoggingEnabled(Collections.emptyMap());
+ } else {
+ windowStoreStoreBuilder.withCachingDisabled();
+ }
+ if (storeToTest.global()) {
+ builder.addGlobalStore(
+ windowStoreStoreBuilder,
+ INPUT_TOPIC_NAME,
+ Consumed.with(Serdes.Integer(), Serdes.Integer()),
+ processorSupplier
+ );
+ } else {
+ builder.addStateStore(windowStoreStoreBuilder);
+ builder
+ .stream(INPUT_TOPIC_NAME, Consumed.with(Serdes.Integer(), Serdes.Integer()))
+ .process(processorSupplier, windowStoreStoreBuilder.name());
+ }
+
+ }
+
+ private void setUpSessionPAPITopology(final SessionBytesStoreSupplier supplier,
+ final StreamsBuilder builder) {
+ final StoreBuilder<?> sessionStoreStoreBuilder;
+ final ProcessorSupplier<Integer, Integer, Void, Void> processorSupplier;
+ sessionStoreStoreBuilder = Stores.sessionStoreBuilder(
+ supplier,
+ Serdes.Integer(),
+ Serdes.Integer()
+ );
+ processorSupplier = () -> new ContextualProcessor<Integer, Integer, Void, Void>() {
+ @Override
+ public void process(final Record<Integer, Integer> record) {
+ final SessionStore<Integer, Integer> stateStore =
+ context().getStateStore(sessionStoreStoreBuilder.name());
+ stateStore.put(
+ new Windowed<>(record.key(), new SessionWindow(WINDOW_START, WINDOW_START)),
+ record.value()
+ );
+ }
+ };
+ if (cache) {
+ sessionStoreStoreBuilder.withCachingEnabled();
+ } else {
+ sessionStoreStoreBuilder.withCachingDisabled();
+ }
+ if (log) {
+ sessionStoreStoreBuilder.withLoggingEnabled(Collections.emptyMap());
+ } else {
+ sessionStoreStoreBuilder.withCachingDisabled();
Review comment:
Seems you got the same :) and ditto elsewhere.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org