diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/pom.xml b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/pom.xml index 45c76a4a5..c0ec938f1 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/pom.xml +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/pom.xml @@ -80,6 +80,11 @@ ${avro.version} provided + + org.awaitility + awaitility + test + 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 33e7890c3..29c43ac42 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 @@ -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 getQueryableStore(String storeName, QueryableStoreType 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 = kafkaStreamsRegistry.getKafkaStreams(); - final Iterator 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 getStateStoreFromKafkaStreams(KafkaStreams kafkaStreams, String storeName, QueryableStoreType 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). * 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 new file mode 100644 index 000000000..f25d549c3 --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryServiceMultiStateStoreTests.java @@ -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> 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 StoreBuilder> store2() { + return Stores.keyValueStoreBuilder( + Stores.persistentKeyValueStore(STORE_2_NAME), Serdes.String(), Serdes.String()); + } + + @Bean + public Consumer> 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> store1() { + return Stores.keyValueStoreBuilder( + Stores.persistentKeyValueStore(STORE_1_NAME), Serdes.String(), Serdes.String()); + } + + @Bean + public Consumer> 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> store2() { + return Stores.keyValueStoreBuilder( + Stores.persistentKeyValueStore(STORE_2_NAME), Serdes.String(), Serdes.String()); + } + + @Bean + public Consumer> 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 { + + @Override + public void init(ProcessorContext context) { + } + + @Override + public String transform(String key, String value) { + return value; + } + + @Override + public void close() { + } + } +}