diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryServices.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryServices.java new file mode 100644 index 000000000..5224f34d5 --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryServices.java @@ -0,0 +1,124 @@ +/* + * Copyright 2018 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 + * + * http://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.util.Map; +import java.util.Optional; + +import org.apache.kafka.common.serialization.Serializer; +import org.apache.kafka.streams.KafkaStreams; +import org.apache.kafka.streams.errors.InvalidStateStoreException; +import org.apache.kafka.streams.state.HostInfo; +import org.apache.kafka.streams.state.QueryableStoreType; +import org.apache.kafka.streams.state.StreamsMetadata; + +import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties; +import org.springframework.util.StringUtils; + +/** + * Services pertinent to the interactive query capabilities of Kafka Streams. This class provides + * services such as querying for a particular store, which instance is hosting a particular store etc. + * This is part of the public API of the kafka streams binder and the users can inject this service in their + * applications to make use of it. + * + * @author Soby Chacko + * @author Renwei Han + * @since 2.1.0 + */ +public class InteractiveQueryServices { + + private final KafkaStreamsRegistry kafkaStreamsRegistry; + private final KafkaStreamsBinderConfigurationProperties binderConfigurationProperties; + + /** + * + * @param kafkaStreamsRegistry holding {@link KafkaStreamsRegistry} + * @param binderConfigurationProperties Kafka Streams binder configuration properties + */ + public InteractiveQueryServices(KafkaStreamsRegistry kafkaStreamsRegistry, + KafkaStreamsBinderConfigurationProperties binderConfigurationProperties) { + this.kafkaStreamsRegistry = kafkaStreamsRegistry; + this.binderConfigurationProperties = binderConfigurationProperties; + } + + /** + * Retrieve and return a queryable store by name created in the application. + * + * @param storeName name of the queryable store + * @param storeType type of the queryable store + * @param generic queryable store + * @return queryable store. + */ + public T getQueryableStore(String storeName, QueryableStoreType storeType) { + for (KafkaStreams kafkaStream : this.kafkaStreamsRegistry.getKafkaStreams()) { + try{ + T store = kafkaStream.store(storeName, storeType); + if (store != null) { + return store; + } + } + catch (InvalidStateStoreException ignored) { + //pass through + } + } + return null; + } + + /** + * Gets the current {@link HostInfo} that the calling kafka streams application is running on. + * + * Note that the end user applications must provide `applicaiton.server` as a configuration property + * when calling this method. If this is not available, then null is returned. + * + * @return the current {@link HostInfo} + */ + public HostInfo getCurrentHostInfo() { + Map configuration = this.binderConfigurationProperties.getConfiguration(); + if (configuration.containsKey("application.server")) { + + String applicationServer = configuration.get("application.server"); + String[] splits = StringUtils.split(applicationServer, ":"); + + return new HostInfo(splits[0], Integer.valueOf(splits[1])); + } + return null; + } + + /** + * Gets the {@link HostInfo} where the provided store and key are hosted on. This may not be the + * current host that is running the application. Kafka Streams will look through all the consumer instances + * under the same application id and retrieves the proper host. + * + * Note that the end user applications must provide `applicaiton.server` as a configuration property + * for all the application instances when calling this method. If this is not available, then null maybe returned. + * + * @param store store name + * @param key key to look for + * @param serializer {@link Serializer} for the key + * @return the {@link HostInfo} where the key for the provided store is hosted currently + */ + public HostInfo getHostInfo(String store, K key, Serializer serializer) { + StreamsMetadata streamsMetadata = this.kafkaStreamsRegistry.getKafkaStreams() + .stream() + .map(k -> Optional.ofNullable(k.metadataForKey(store, key, serializer))) + .filter(Optional::isPresent) + .map(Optional::get) + .findFirst() + .orElse(null); + return streamsMetadata != null ? streamsMetadata.hostInfo() : null; + } +} diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index 4ec89519c..0de6e9d62 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -144,14 +144,25 @@ public class KafkaStreamsBinderSupportAutoConfiguration { } @Bean - public QueryableStoreRegistry queryableStoreTypeRegistry() { - return new QueryableStoreRegistry(); + public QueryableStoreRegistry queryableStoreTypeRegistry(KafkaStreamsRegistry kafkaStreamsRegistry) { + return new QueryableStoreRegistry(kafkaStreamsRegistry); + } + + @Bean + public InteractiveQueryServices interactiveQueryServices(KafkaStreamsRegistry kafkaStreamsRegistry, + KafkaStreamsBinderConfigurationProperties binderConfigurationProperties) { + return new InteractiveQueryServices(kafkaStreamsRegistry, binderConfigurationProperties); + } + + @Bean + public KafkaStreamsRegistry kafkaStreamsRegistry() { + return new KafkaStreamsRegistry(); } @Bean public StreamsBuilderFactoryManager streamsBuilderFactoryManager(KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue, - QueryableStoreRegistry queryableStoreRegistry) { - return new StreamsBuilderFactoryManager(kafkaStreamsBindingInformationCatalogue, queryableStoreRegistry); + KafkaStreamsRegistry kafkaStreamsRegistry) { + return new StreamsBuilderFactoryManager(kafkaStreamsBindingInformationCatalogue, kafkaStreamsRegistry); } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsRegistry.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsRegistry.java new file mode 100644 index 000000000..818e891ae --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsRegistry.java @@ -0,0 +1,46 @@ +/* + * Copyright 2018 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 + * + * http://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.util.HashSet; +import java.util.Set; + +import org.apache.kafka.streams.KafkaStreams; + +/** + * An internal registry for holding {@KafkaStreams} objects maintained through + * {@link StreamsBuilderFactoryManager}. + * + * @author Soby Chacko + */ +class KafkaStreamsRegistry { + + private final Set kafkaStreams = new HashSet<>(); + + Set getKafkaStreams() { + return kafkaStreams; + } + + /** + * Register the {@link KafkaStreams} object created in the application. + * + * @param kafkaStreams {@link KafkaStreams} object created in the application + */ + void registerKafkaStreams(KafkaStreams kafkaStreams) { + this.kafkaStreams.add(kafkaStreams); + } +} diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/QueryableStoreRegistry.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/QueryableStoreRegistry.java index 15e23c03b..5a7685c0b 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/QueryableStoreRegistry.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/QueryableStoreRegistry.java @@ -16,9 +16,6 @@ package org.springframework.cloud.stream.binder.kafka.streams; -import java.util.HashSet; -import java.util.Set; - import org.apache.kafka.streams.KafkaStreams; import org.apache.kafka.streams.errors.InvalidStateStoreException; import org.apache.kafka.streams.state.QueryableStoreType; @@ -30,10 +27,15 @@ import org.apache.kafka.streams.state.QueryableStoreType; * @author Soby Chacko * @author Renwei Han * @since 2.0.0 + * @deprecated in favor of {@link InteractiveQueryServices} */ public class QueryableStoreRegistry { - private final Set kafkaStreams = new HashSet<>(); + private final KafkaStreamsRegistry kafkaStreamsRegistry; + + public QueryableStoreRegistry(KafkaStreamsRegistry kafkaStreamsRegistry) { + this.kafkaStreamsRegistry = kafkaStreamsRegistry; + } /** * Retrieve and return a queryable store by name created in the application. @@ -42,10 +44,11 @@ public class QueryableStoreRegistry { * @param storeType type of the queryable store * @param generic queryable store * @return queryable store. + * @deprecated in favor of {@link InteractiveQueryServices#getQueryableStore(String, QueryableStoreType)} */ public T getQueryableStoreType(String storeName, QueryableStoreType storeType) { - for (KafkaStreams kafkaStream : kafkaStreams) { + for (KafkaStreams kafkaStream : this.kafkaStreamsRegistry.getKafkaStreams()) { try{ T store = kafkaStream.store(storeName, storeType); if (store != null) { @@ -59,12 +62,4 @@ public class QueryableStoreRegistry { return null; } - /** - * Register the {@link KafkaStreams} object created in the application. - * - * @param kafkaStreams {@link KafkaStreams} object created in the application - */ - void registerKafkaStreams(KafkaStreams kafkaStreams) { - this.kafkaStreams.add(kafkaStreams); - } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java index 7bd748bbe..fe765cbb3 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java @@ -38,14 +38,14 @@ import org.springframework.kafka.core.StreamsBuilderFactoryBean; class StreamsBuilderFactoryManager implements SmartLifecycle { private final KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue; - private final QueryableStoreRegistry queryableStoreRegistry; + private final KafkaStreamsRegistry kafkaStreamsRegistry; private volatile boolean running; StreamsBuilderFactoryManager(KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue, - QueryableStoreRegistry queryableStoreRegistry) { + KafkaStreamsRegistry kafkaStreamsRegistry) { this.kafkaStreamsBindingInformationCatalogue = kafkaStreamsBindingInformationCatalogue; - this.queryableStoreRegistry = queryableStoreRegistry; + this.kafkaStreamsRegistry = kafkaStreamsRegistry; } @Override @@ -68,7 +68,7 @@ class StreamsBuilderFactoryManager implements SmartLifecycle { Set streamsBuilderFactoryBeans = this.kafkaStreamsBindingInformationCatalogue.getStreamsBuilderFactoryBeans(); for (StreamsBuilderFactoryBean streamsBuilderFactoryBean : streamsBuilderFactoryBeans) { streamsBuilderFactoryBean.start(); - queryableStoreRegistry.registerKafkaStreams(streamsBuilderFactoryBean.getKafkaStreams()); + kafkaStreamsRegistry.registerKafkaStreams(streamsBuilderFactoryBean.getKafkaStreams()); } this.running = true; } catch (Exception e) { diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsInteractiveQueryIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsInteractiveQueryIntegrationTests.java index 9907c201c..141b33cd9 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsInteractiveQueryIntegrationTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsInteractiveQueryIntegrationTests.java @@ -21,10 +21,12 @@ import java.util.Map; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.serialization.IntegerSerializer; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Serialized; +import org.apache.kafka.streams.state.HostInfo; import org.apache.kafka.streams.state.QueryableStoreTypes; import org.apache.kafka.streams.state.ReadOnlyKeyValueStore; import org.junit.AfterClass; @@ -32,7 +34,6 @@ import org.junit.BeforeClass; import org.junit.ClassRule; import org.junit.Test; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; @@ -89,6 +90,7 @@ public class KafkaStreamsInteractiveQueryIntegrationTests { "--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", "--spring.cloud.stream.bindings.output.producer.headerMode=raw", "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", + "--spring.cloud.stream.kafka.streams.binder.configuration.application.server=" + embeddedKafka.getBrokersAsString(), "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(), "--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); try { @@ -109,15 +111,23 @@ public class KafkaStreamsInteractiveQueryIntegrationTests { ProductCountApplication.Foo foo = context.getBean(ProductCountApplication.Foo.class); assertThat(foo.getProductStock(123).equals(1L)); + + //perform assertions on HostInfo related methods in InteractiveQueryServices + InteractiveQueryServices interactiveQueryServices = context.getBean(InteractiveQueryServices.class); + HostInfo currentHostInfo = interactiveQueryServices.getCurrentHostInfo(); + assertThat(currentHostInfo.host() + ":" + currentHostInfo.port()).isEqualTo(embeddedKafka.getBrokersAsString()); + + HostInfo hostInfo = interactiveQueryServices.getHostInfo("prod-id-count-store", 123, new IntegerSerializer()); + assertThat(hostInfo.host() + ":" + hostInfo.port()).isEqualTo(embeddedKafka.getBrokersAsString()); + + HostInfo hostInfoFoo = interactiveQueryServices.getHostInfo("prod-id-count-store-foo", 123, new IntegerSerializer()); + assertThat(hostInfoFoo).isNull(); } @EnableBinding(KafkaStreamsProcessor.class) @EnableAutoConfiguration public static class ProductCountApplication { - @Autowired - private QueryableStoreRegistry queryableStoreRegistry; - @StreamListener("input") @SendTo("output") @SuppressWarnings("deprecation") @@ -133,20 +143,21 @@ public class KafkaStreamsInteractiveQueryIntegrationTests { } @Bean - public Foo foo(QueryableStoreRegistry queryableStoreRegistry) { - return new Foo(queryableStoreRegistry); + public Foo foo(InteractiveQueryServices interactiveQueryServices) { + return new Foo(interactiveQueryServices); } static class Foo { - QueryableStoreRegistry queryableStoreRegistry; + InteractiveQueryServices interactiveQueryServices; - Foo(QueryableStoreRegistry queryableStoreRegistry) { - this.queryableStoreRegistry = queryableStoreRegistry; + Foo(InteractiveQueryServices interactiveQueryServices) { + this.interactiveQueryServices = interactiveQueryServices; } public Long getProductStock(Integer id) { ReadOnlyKeyValueStore keyValueStore = - queryableStoreRegistry.getQueryableStoreType("prod-id-count-store", QueryableStoreTypes.keyValueStore()); + interactiveQueryServices.getQueryableStore("prod-id-count-store", QueryableStoreTypes.keyValueStore()); + return (Long) keyValueStore.get(id); } }