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

Make InteractiveQueryService only return used state stores (backport #2445)

Browse Source 
This restores to KafkaStreams 3.0 behavior in this area.
This commit is contained in:
Chris Bono
2022-06-27 21:59:47 -05:00
committed by Soby Chacko
parent b105d8a2e9
commit 17145dcc27
3 changed files with 263 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/pom.xml
View File

@@ -80,6 +80,11 @@
<version>${avro.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>

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

@@ -16,12 +16,10 @@
package org.springframework.cloud.stream.binder.kafka.streams;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
@@ -33,6 +31,7 @@ import org.apache.kafka.streams.KeyQueryMetadata;
import org.apache.kafka.streams.StoreQueryParameters;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.errors.InvalidStateStoreException;
import org.apache.kafka.streams.errors.UnknownStateStoreException;
import org.apache.kafka.streams.state.HostInfo;
import org.apache.kafka.streams.state.QueryableStoreType;
import org.apache.kafka.streams.state.StreamsMetadata;
@@ -84,18 +83,14 @@ public class InteractiveQueryService {
*/
public <T> T getQueryableStore(String storeName, QueryableStoreType<T> storeType) {
final RetryTemplate retryTemplate = getRetryTemplate();
KafkaStreams contextSpecificKafkaStreams = getThreadContextSpecificKafkaStreams();
return retryTemplate.execute(context -> {
return getRetryTemplate().execute(context -> {
T store = null;
Throwable throwable = null;
if (contextSpecificKafkaStreams != null) {
try {
store = contextSpecificKafkaStreams.store(
StoreQueryParameters.fromNameAndType(
storeName, storeType));
store = contextSpecificKafkaStreams.store(StoreQueryParameters.fromNameAndType(storeName, storeType));
}
catch (InvalidStateStoreException e) {
// pass through..
@@ -105,32 +100,34 @@ public class InteractiveQueryService {
if (store != null) {
return store;
}
else if (contextSpecificKafkaStreams != null) {
LOG.warn("Store " + storeName
+ " could not be found in Streams context, falling back to all known Streams instances");
if (contextSpecificKafkaStreams != null) {
LOG.warn("Store (" + storeName + ") could not be found in Streams context, falling back to all known Streams instances");
}
final Set<KafkaStreams> kafkaStreams = kafkaStreamsRegistry.getKafkaStreams();
final Iterator<KafkaStreams> iterator = kafkaStreams.iterator();
while (iterator.hasNext()) {
for (KafkaStreams kafkaStreamApp : kafkaStreamsRegistry.getKafkaStreams()) {
try {
store = iterator.next()
.store(StoreQueryParameters.fromNameAndType(
storeName, storeType));
return getStateStoreFromKafkaStreams(kafkaStreamApp, storeName, storeType);
}
catch (InvalidStateStoreException e) {
// pass through..
throwable = e;
catch (Exception ex) {
throwable = ex;
}
}
if (store != null) {
return store;
}
throw new IllegalStateException(
"Error when retrieving state store: " + storeName,
throwable);
throw new IllegalStateException("Error retrieving state store: " + storeName, throwable);
});
}
private <T> T getStateStoreFromKafkaStreams(KafkaStreams kafkaStreams, String storeName, QueryableStoreType<T> storeType) {
// Check KafkaStreams app knows about the state store
T store = kafkaStreams.store(StoreQueryParameters.fromNameAndType(storeName, storeType));
// Check KafkaStreams app actually has the state store
if (kafkaStreams.streamsMetadataForStore(storeName).stream()
.noneMatch((sm) -> sm.stateStoreNames().contains(storeName))) {
throw new UnknownStateStoreException("Store (" + storeName + ") not available to Streams instance");
}
return store;
}
/**
* Retrieves the current {@link KafkaStreams} context if executing Thread is created by a Streams App (contains a matching application id in Thread's name).
*

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

@@ -0,0 +1,235 @@
/*
* Copyright 2022-2022 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 java.time.Duration;
import java.util.function.Consumer;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.errors.UnknownStateStoreException;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.ValueTransformerWithKey;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.state.KeyValueStore;
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.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.CleanupConfig;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.condition.EmbeddedKafkaCondition;
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;
/**
* Tests for the {@link InteractiveQueryService} when dealing with multiple KafkaStreams apps and state stores.
*
* @author Chris Bono
*/
@EmbeddedKafka(topics = {"input1", "input2"})
class InteractiveQueryServiceMultiStateStoreTests {
private static final String STORE_1_NAME = "store1";
private static final String STORE_2_NAME = "store2";
private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker();
@Test
void stateStoreOnlyAvailableOnKafkaStreamsAppWhereItIsUsed() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder()
.sources(MultipleAppsWithUsedStateStoresTestApplication.class)
.web(WebApplicationType.NONE)
.run("--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.function.definition=app1;app2",
"--spring.cloud.stream.function.bindings.app1-in-0=input1",
"--spring.cloud.stream.function.bindings.app2-in-0=input2",
"--spring.cloud.stream.kafka.streams.binder.functions.app1.application-id=stateStoreTestApp1",
"--spring.cloud.stream.kafka.streams.binder.functions.app2.application-id=stateStoreTestApp2",
"--spring.cloud.stream.kafka.streams.binder.configuration.application.server="
+ embeddedKafka.getBrokersAsString(),
"--spring.cloud.stream.kafka.streams.binder.brokers="
+ embeddedKafka.getBrokersAsString())
) {
waitForRunningStreams(context.getBean(KafkaStreamsRegistry.class));
// The KafkaStreams.store() used by query service is non-deterministic so perform the operation multiple times to
// surface any possible issues. Also, no need to actually write anything to the stores, the store.get() call will
// cause a failure when the state store is invalid.
InteractiveQueryService queryService = context.getBean(InteractiveQueryService.class);
for (int i = 0; i < 100; i++) {
assertThat(queryService.getQueryableStore(STORE_1_NAME, QueryableStoreTypes.keyValueStore())
.get("someKey")).isNull();
assertThat(queryService.getQueryableStore(STORE_2_NAME, QueryableStoreTypes.keyValueStore())
.get("someKey")).isNull();
}
}
}
@Test
void stateStoreNotAvailableThrowsException() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder()
.sources(MultipleAppsWithUnusedStateStoresTestApplication.class)
.web(WebApplicationType.NONE)
.run("--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.function.definition=app1;app2",
"--spring.cloud.stream.function.bindings.app1-in-0=input1",
"--spring.cloud.stream.function.bindings.app2-in-0=input2",
"--spring.cloud.stream.kafka.streams.binder.functions.app1.application-id=stateStoreTestApp3",
"--spring.cloud.stream.kafka.streams.binder.functions.app2.application-id=stateStoreTestApp4",
"--spring.cloud.stream.kafka.streams.binder.configuration.application.server="
+ embeddedKafka.getBrokersAsString(),
"--spring.cloud.stream.kafka.streams.binder.brokers="
+ embeddedKafka.getBrokersAsString())
) {
waitForRunningStreams(context.getBean(KafkaStreamsRegistry.class));
InteractiveQueryService queryService = context.getBean(InteractiveQueryService.class);
assertThatThrownBy(() -> queryService.getQueryableStore(STORE_1_NAME, QueryableStoreTypes.keyValueStore()))
.isInstanceOf(IllegalStateException.class)
.hasMessage("Error retrieving state store: " + STORE_1_NAME)
.hasRootCauseInstanceOf(UnknownStateStoreException.class);
assertThatThrownBy(() -> queryService.getQueryableStore(STORE_2_NAME, QueryableStoreTypes.keyValueStore()))
.isInstanceOf(IllegalStateException.class)
.hasMessage("Error retrieving state store: " + STORE_2_NAME)
.hasRootCauseInstanceOf(UnknownStateStoreException.class);
}
}
private void waitForRunningStreams(KafkaStreamsRegistry registry) {
await().atMost(Duration.ofSeconds(60))
.until(() -> registry.streamsBuilderFactoryBeans().stream()
.allMatch(x -> x.getKafkaStreams().state().equals(KafkaStreams.State.RUNNING)));
}
@EnableAutoConfiguration
@Configuration(proxyBeanMethods = false)
static class MultipleAppsWithUsedStateStoresTestApplication {
private static final Logger log = LoggerFactory.getLogger(MultipleAppsWithUsedStateStoresTestApplication.class);
public static void main(String[] args) {
SpringApplication.run(MultipleAppsWithUsedStateStoresTestApplication.class, args);
}
@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 StoreBuilder<KeyValueStore<String, String>> store2() {
return Stores.keyValueStoreBuilder(
Stores.persistentKeyValueStore(STORE_2_NAME), Serdes.String(), Serdes.String());
}
@Bean
public Consumer<KStream<String, String>> app2() {
return s -> s
.transformValues(EchoTransformer::new, STORE_2_NAME)
.foreach((k, v) -> log.info("Echo {} -> {} into {}", k, v, STORE_2_NAME));
}
@Bean
public CleanupConfig cleanupConfig() {
return new CleanupConfig(false, true);
}
}
@EnableAutoConfiguration
@Configuration(proxyBeanMethods = false)
static class MultipleAppsWithUnusedStateStoresTestApplication {
private static final Logger log = LoggerFactory.getLogger(MultipleAppsWithUnusedStateStoresTestApplication.class);
public static void main(String[] args) {
SpringApplication.run(MultipleAppsWithUnusedStateStoresTestApplication.class, args);
}
@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() {
// NOTE: No reference to the state store via transformer
return s -> s
.foreach((k, v) -> log.info("Echo {} -> {} into {}", k, v, STORE_1_NAME));
}
@Bean
public StoreBuilder<KeyValueStore<String, String>> store2() {
return Stores.keyValueStoreBuilder(
Stores.persistentKeyValueStore(STORE_2_NAME), Serdes.String(), Serdes.String());
}
@Bean
public Consumer<KStream<String, String>> app2() {
// NOTE: No reference to the state store via transformer
return s -> s
.foreach((k, v) -> log.info("Echo {} -> {} into {}", k, v, STORE_2_NAME));
}
@Bean
public CleanupConfig cleanupConfig() {
return new CleanupConfig(false, true);
}
}
static class EchoTransformer implements ValueTransformerWithKey<String, String, String> {
@Override
public void init(ProcessorContext context) {
}
@Override
public String transform(String key, String value) {
return value;
}
@Override
public void close() {
}
}
}