SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Allow fine tuning of StoreQueryParameters (#2618)

Browse Source 
Add StoreQueryParametersCustomizer to InteractiveQueryService 
in Kafka Streams binder.

Fixes #2608
This commit is contained in:
Soby Chacko
2023-01-10 19:11:09 -05:00
committed by GitHub
parent a8114cee9a
commit 8b0fbe122f
5 changed files with 153 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryService.java
View File

@@ -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 <T> generic queryable store
* @return queryable store.
*/
@SuppressWarnings("unchecked")
public <T> T getQueryableStore(String storeName, QueryableStoreType<T> storeType) {
KafkaStreams contextSpecificKafkaStreams = getThreadContextSpecificKafkaStreams();
final StoreQueryParameters<T> storeQueryParams = StoreQueryParameters.fromNameAndType(storeName, storeType);
StoreQueryParameters<T> storeQueryParams = StoreQueryParameters.fromNameAndType(storeName, storeType);
if (this.storeQueryParametersCustomizer != null) {
storeQueryParams = ((StoreQueryParametersCustomizer<T>) this.storeQueryParametersCustomizer).customize(storeQueryParams);
}
AtomicReference<StoreQueryParameters<T>> 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<KafkaStreams, T> 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;
}
}

10
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java
View File

@@ -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<StoreQueryParametersCustomizer<?>> storeQueryParametersCustomizerProvider) {
InteractiveQueryService interactiveQueryService = new InteractiveQueryService(kafkaStreamsRegistry, properties);
StoreQueryParametersCustomizer<?> storeQueryParametersCustomizer = storeQueryParametersCustomizerProvider.getIfUnique();
if (storeQueryParametersCustomizer != null) {
interactiveQueryService.setStoreQueryParametersCustomizer(storeQueryParametersCustomizer);
}
return interactiveQueryService;
}
@Bean

46
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StoreQueryParametersCustomizer.java Normal file
View File

@@ -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 <T> state store generic type
*
* @author Soby Chacko
* @since 4.0.1
*/
@FunctionalInterface
public interface StoreQueryParametersCustomizer<T> {
/**
* Customizing the internal {@link StoreQueryParameters} used by the {@link InteractiveQueryService}.
*
* @param storeQueryParameters Original {@link StoreQueryParameters} to customize
* @return the customized {@link StoreQueryParameters}
*/
StoreQueryParameters<T> customize(StoreQueryParameters<T> storeQueryParameters);
}

65
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryServiceMultiStateStoreTests.java
View File

@@ -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<KeyValueStore<String, String>> store1() {
return Stores.keyValueStoreBuilder(
Stores.persistentKeyValueStore(STORE_1_NAME), Serdes.String(), Serdes.String());
}
@Bean
public Consumer<KStream<String, String>> 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<String, String, String> {
@Override

16
docs/src/main/asciidoc/kafka/kafka-streams.adoc
View File

@@ -1078,6 +1078,22 @@ Use the following API method to retrieve the `KakfaStreams` object associated wi
public <K> KafkaStreams getKafkaStreams(String store, K key, Serializer<K> 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<T> customize(StoreQueryParameters<T> 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: