From 8b0fbe122f88c79b1afc7e1bcffcea9d79e8c20b Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Tue, 10 Jan 2023 19:11:09 -0500 Subject: [PATCH] Allow fine tuning of StoreQueryParameters (#2618) Add StoreQueryParametersCustomizer to InteractiveQueryService in Kafka Streams binder. Fixes #2608 --- .../streams/InteractiveQueryService.java | 23 +++++-- ...StreamsBinderSupportAutoConfiguration.java | 10 ++- .../StoreQueryParametersCustomizer.java | 46 +++++++++++++ ...ctiveQueryServiceMultiStateStoreTests.java | 65 ++++++++++++++++++- .../main/asciidoc/kafka/kafka-streams.adoc | 16 +++++ 5 files changed, 153 insertions(+), 7 deletions(-) create mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StoreQueryParametersCustomizer.java diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryService.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryService.java index 330a1fef9..f1ab39778 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryService.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryService.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2022 the original author or authors. + * Copyright 2018-2023 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -67,6 +67,8 @@ public class InteractiveQueryService { private final KafkaStreamsVersionAgnosticTopologyInfoFacade topologyInfoFacade; + private StoreQueryParametersCustomizer storeQueryParametersCustomizer; + /** * Constructor for InteractiveQueryService. * @param kafkaStreamsRegistry holding {@link KafkaStreamsRegistry} @@ -86,18 +88,24 @@ public class InteractiveQueryService { * @param generic queryable store * @return queryable store. */ + @SuppressWarnings("unchecked") public T getQueryableStore(String storeName, QueryableStoreType storeType) { KafkaStreams contextSpecificKafkaStreams = getThreadContextSpecificKafkaStreams(); - final StoreQueryParameters storeQueryParams = StoreQueryParameters.fromNameAndType(storeName, storeType); + StoreQueryParameters storeQueryParams = StoreQueryParameters.fromNameAndType(storeName, storeType); + if (this.storeQueryParametersCustomizer != null) { + storeQueryParams = ((StoreQueryParametersCustomizer) this.storeQueryParametersCustomizer).customize(storeQueryParams); + } + + AtomicReference> storeQueryParametersAtomicReference = new AtomicReference<>(storeQueryParams); return getRetryTemplate().execute(context -> { T store = null; Throwable throwable = null; if (contextSpecificKafkaStreams != null) { try { - store = contextSpecificKafkaStreams.store(storeQueryParams); + store = contextSpecificKafkaStreams.store(storeQueryParametersAtomicReference.get()); } catch (InvalidStateStoreException e) { throwable = e; @@ -114,7 +122,7 @@ public class InteractiveQueryService { Map candidateStores = new HashMap<>(); for (KafkaStreams kafkaStreamApp : kafkaStreamsRegistry.getKafkaStreams()) { try { - candidateStores.put(kafkaStreamApp, kafkaStreamApp.store(storeQueryParams)); + candidateStores.put(kafkaStreamApp, kafkaStreamApp.store(storeQueryParametersAtomicReference.get())); } catch (Exception ex) { throwable = ex; @@ -304,4 +312,11 @@ public class InteractiveQueryService { .collect(Collectors.toList()); } + /** + * @param storeQueryParametersCustomizer to customize + * @since 4.0.1 + */ + public void setStoreQueryParametersCustomizer(StoreQueryParametersCustomizer storeQueryParametersCustomizer) { + this.storeQueryParametersCustomizer = storeQueryParametersCustomizer; + } } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index 6bc779c37..85d038edf 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -348,8 +348,14 @@ public class KafkaStreamsBinderSupportAutoConfiguration { @Bean public InteractiveQueryService interactiveQueryServices( KafkaStreamsRegistry kafkaStreamsRegistry, - @Qualifier("binderConfigurationProperties")KafkaStreamsBinderConfigurationProperties properties) { - return new InteractiveQueryService(kafkaStreamsRegistry, properties); + @Qualifier("binderConfigurationProperties")KafkaStreamsBinderConfigurationProperties properties, + ObjectProvider> storeQueryParametersCustomizerProvider) { + InteractiveQueryService interactiveQueryService = new InteractiveQueryService(kafkaStreamsRegistry, properties); + StoreQueryParametersCustomizer storeQueryParametersCustomizer = storeQueryParametersCustomizerProvider.getIfUnique(); + if (storeQueryParametersCustomizer != null) { + interactiveQueryService.setStoreQueryParametersCustomizer(storeQueryParametersCustomizer); + } + return interactiveQueryService; } @Bean diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StoreQueryParametersCustomizer.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StoreQueryParametersCustomizer.java new file mode 100644 index 000000000..012451eb8 --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StoreQueryParametersCustomizer.java @@ -0,0 +1,46 @@ +/* + * Copyright 2023-2023 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kafka.streams; + +import org.apache.kafka.streams.StoreQueryParameters; +import org.apache.kafka.streams.state.QueryableStoreType; + +/** + * Interface for customizing {@link StoreQueryParameters}. + * + * There are instances, in which an application wants to customize the internal + * {@link StoreQueryParameters} object created by {@link InteractiveQueryService} + * in {@link InteractiveQueryService#getQueryableStore(String, QueryableStoreType)} method. + * Applications can provide an implementation for this customizer as a bean + * and {@link InteractiveQueryService} will be provisioned with that customizer. + * + * @param state store generic type + * + * @author Soby Chacko + * @since 4.0.1 + */ +@FunctionalInterface +public interface StoreQueryParametersCustomizer { + + /** + * Customizing the internal {@link StoreQueryParameters} used by the {@link InteractiveQueryService}. + * + * @param storeQueryParameters Original {@link StoreQueryParameters} to customize + * @return the customized {@link StoreQueryParameters} + */ + StoreQueryParameters customize(StoreQueryParameters storeQueryParameters); +} diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryServiceMultiStateStoreTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryServiceMultiStateStoreTests.java index 1daa5dcc5..5c2928809 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryServiceMultiStateStoreTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryServiceMultiStateStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2022-2022 the original author or authors. + * Copyright 2022-2023 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,6 +21,7 @@ import java.util.function.Consumer; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.KafkaStreams; +import org.apache.kafka.streams.StoreQueryParameters; import org.apache.kafka.streams.errors.UnknownStateStoreException; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.ValueTransformerWithKey; @@ -30,6 +31,7 @@ import org.apache.kafka.streams.state.QueryableStoreTypes; import org.apache.kafka.streams.state.StoreBuilder; import org.apache.kafka.streams.state.Stores; import org.junit.jupiter.api.Test; +import org.mockito.Mockito; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -48,11 +50,14 @@ import org.springframework.kafka.test.context.EmbeddedKafka; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.awaitility.Awaitility.await; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; /** * Tests for the {@link InteractiveQueryService} when dealing with multiple KafkaStreams apps and state stores. * * @author Chris Bono + * @author Soby Chacko */ @EmbeddedKafka(topics = {"input1", "input2"}) class InteractiveQueryServiceMultiStateStoreTests { @@ -139,6 +144,33 @@ class InteractiveQueryServiceMultiStateStoreTests { } } + @Test + @SuppressWarnings("unchecked") + void storeQueryParameterCustomizerIsApplied() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder() + .sources(StoreQueryParameterCustomizerTestApplication.class) + .web(WebApplicationType.NONE) + .run("--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=app1", + "--spring.cloud.stream.function.bindings.app1-in-0=input1", + "--spring.cloud.stream.kafka.streams.binder.functions.app1.application-id=storeQueryParameterCustomizerIsAppliedAppId", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString()) + ) { + waitForRunningStreams(context.getBean(KafkaStreamsRegistry.class)); + + InteractiveQueryService queryService = context.getBean(InteractiveQueryService.class); + + StoreQueryParametersCustomizer storeQueryParametersCustomizer = context.getBean(StoreQueryParametersCustomizer.class); + StoreQueryParameters storeQueryParams = StoreQueryParameters.fromNameAndType(STORE_1_NAME, QueryableStoreTypes.keyValueStore()); + + when(storeQueryParametersCustomizer.customize(Mockito.any(StoreQueryParameters.class))).thenReturn(storeQueryParams); + + queryService.getQueryableStore(STORE_1_NAME, QueryableStoreTypes.keyValueStore()).get("someKey"); + verify(storeQueryParametersCustomizer).customize(Mockito.any(StoreQueryParameters.class)); + } + } + private void waitForRunningStreams(KafkaStreamsRegistry registry) { await().atMost(Duration.ofSeconds(60)) .until(() -> registry.streamsBuilderFactoryBeans().stream() @@ -231,6 +263,37 @@ class InteractiveQueryServiceMultiStateStoreTests { } + @EnableAutoConfiguration + @Configuration(proxyBeanMethods = false) + static class StoreQueryParameterCustomizerTestApplication { + + private static final Logger log = LoggerFactory.getLogger(StoreQueryParameterCustomizerTestApplication.class); + + + @Bean + public StoreBuilder> store1() { + return Stores.keyValueStoreBuilder( + Stores.persistentKeyValueStore(STORE_1_NAME), Serdes.String(), Serdes.String()); + } + + @Bean + public Consumer> app1() { + return s -> s + .transformValues(EchoTransformer::new, STORE_1_NAME) + .foreach((k, v) -> log.info("Echo {} -> {} into {}", k, v, STORE_1_NAME)); + } + + @Bean + public CleanupConfig cleanupConfig() { + return new CleanupConfig(false, true); + } + + @Bean + StoreQueryParametersCustomizer storeQueryParametersCustomizer() { + return Mockito.mock(StoreQueryParametersCustomizer.class); + } + } + static class EchoTransformer implements ValueTransformerWithKey { @Override diff --git a/docs/src/main/asciidoc/kafka/kafka-streams.adoc b/docs/src/main/asciidoc/kafka/kafka-streams.adoc index bcfb9dd61..884f0ce04 100644 --- a/docs/src/main/asciidoc/kafka/kafka-streams.adoc +++ b/docs/src/main/asciidoc/kafka/kafka-streams.adoc @@ -1078,6 +1078,22 @@ Use the following API method to retrieve the `KakfaStreams` object associated wi public KafkaStreams getKafkaStreams(String store, K key, Serializer serializer) ``` +==== Customizing Store Query Parameters + +Sometimes it is necessary that you need to fine tune the store query parameters before querying the store through `InteractiveQueryService`. +For this purpose, starting with the `4.0.1` version of the binder, you can provide a bean for `StoreQueryParametersCustomizer` which is a functional interface with a `customize` method that takes a `StoreQueryParameter` as the argument. +Here is its method signature. + +``` +StoreQueryParameters customize(StoreQueryParameters storeQueryParameters); +``` + +Using this approach, applications can further customize the `StoreQueryParameters` such as enabling stale stores. + +When this bean is present in this application, `InteractiveQueryService` will call its `customize` method before querying the state store. + +NOTE: Keep in mind that, there must be a unique bean for `StoreQueryParametersCustomizer` available in the application. + === Health Indicator The health indicator requires the dependency `spring-boot-starter-actuator`. For maven use: