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 29c43ac42..49eba478e 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,6 +16,7 @@ package org.springframework.cloud.stream.binder.kafka.streams; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Objects; @@ -53,6 +54,7 @@ import org.springframework.util.StringUtils; * @author Renwei Han * @author Serhii Siryi * @author Nico Pommerening + * @author Chris Bono * @since 2.1.0 */ public class InteractiveQueryService { @@ -63,6 +65,8 @@ public class InteractiveQueryService { private final KafkaStreamsBinderConfigurationProperties binderConfigurationProperties; + private final KafkaStreamsVersionAgnosticTopologyInfoFacade topologyInfoFacade; + /** * Constructor for InteractiveQueryService. * @param kafkaStreamsRegistry holding {@link KafkaStreamsRegistry} @@ -72,6 +76,7 @@ public class InteractiveQueryService { KafkaStreamsBinderConfigurationProperties binderConfigurationProperties) { this.kafkaStreamsRegistry = kafkaStreamsRegistry; this.binderConfigurationProperties = binderConfigurationProperties; + this.topologyInfoFacade = new KafkaStreamsVersionAgnosticTopologyInfoFacade(); } /** @@ -85,15 +90,16 @@ public class InteractiveQueryService { KafkaStreams contextSpecificKafkaStreams = getThreadContextSpecificKafkaStreams(); + final StoreQueryParameters storeQueryParams = StoreQueryParameters.fromNameAndType(storeName, storeType); + return getRetryTemplate().execute(context -> { T store = null; Throwable throwable = null; if (contextSpecificKafkaStreams != null) { try { - store = contextSpecificKafkaStreams.store(StoreQueryParameters.fromNameAndType(storeName, storeType)); + store = contextSpecificKafkaStreams.store(storeQueryParams); } catch (InvalidStateStoreException e) { - // pass through.. throwable = e; } } @@ -104,30 +110,42 @@ public class InteractiveQueryService { LOG.warn("Store (" + storeName + ") could not be found in Streams context, falling back to all known Streams instances"); } + // Find all apps that know about the store + Map candidateStores = new HashMap<>(); for (KafkaStreams kafkaStreamApp : kafkaStreamsRegistry.getKafkaStreams()) { try { - return getStateStoreFromKafkaStreams(kafkaStreamApp, storeName, storeType); + candidateStores.put(kafkaStreamApp, kafkaStreamApp.store(storeQueryParams)); } catch (Exception ex) { throwable = ex; } } + + // Store exists in a single app - no further resolution required + if (candidateStores.size() == 1) { + return candidateStores.values().stream().findFirst().get(); + } + + // If the store is in multiple streams apps - discard any apps that do not actually have the store + if (candidateStores.size() > 1) { + + candidateStores = candidateStores.entrySet().stream() + .filter((e) -> this.topologyInfoFacade.streamsAppActuallyHasStore(e.getKey(), storeName)) + .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); + + if (candidateStores.size() == 1) { + return candidateStores.values().stream().findFirst().get(); + } + + throwable = (candidateStores.size() == 0) ? + new UnknownStateStoreException("Store (" + storeName + ") not available to Streams instance") : + new InvalidStateStoreException("Store (" + storeName + ") available to more than one Streams instance"); + + } 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/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsVersionAgnosticTopologyInfoFacade.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsVersionAgnosticTopologyInfoFacade.java new file mode 100644 index 000000000..3c8bbc7b0 --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsVersionAgnosticTopologyInfoFacade.java @@ -0,0 +1,107 @@ +/* + * Copyright 2018-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.lang.reflect.Field; +import java.lang.reflect.Method; +import java.util.Collection; + +import org.apache.kafka.streams.KafkaStreams; +import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder; +import org.apache.kafka.streams.processor.internals.TopologyMetadata; + +import org.springframework.core.log.LogAccessor; +import org.springframework.lang.Nullable; +import org.springframework.util.CollectionUtils; +import org.springframework.util.ReflectionUtils; + +/** + * A facade to access topology info for a Kafka Streams application in a version agnostic + * manner. + *

Before kafka-streams 3.1 the topology exists at 'KafkaStreams.internalTopologyBuilder'. + * Starting in kafka-streams 3.1 the topology exists at 'KafkaStreams.topologyMetadata'. + * + * @author Chris Bono + * @since 3.2.6 + */ +class KafkaStreamsVersionAgnosticTopologyInfoFacade { + + private final LogAccessor logger = new LogAccessor(KafkaStreamsVersionAgnosticTopologyInfoFacade.class); + + @Nullable + private Field topologyInfoField; + + @Nullable + private Method sourceTopicsForStoreMethod; + + KafkaStreamsVersionAgnosticTopologyInfoFacade() { + + // First look for KafkaStreams.internalTopologyBuilder (exists in kafka-streams <= 3.0) + Field internalTopologyBuilderField = ReflectionUtils.findField(KafkaStreams.class, "internalTopologyBuilder"); + if (internalTopologyBuilderField != null) { + internalTopologyBuilderField.setAccessible(true); + this.topologyInfoField = internalTopologyBuilderField; + this.sourceTopicsForStoreMethod = ReflectionUtils.findMethod(InternalTopologyBuilder.class, "sourceTopicsForStore", String.class); + logger.info("Using KafkaStreams.internalTopologyBuilder.sourceTopicsForStore for kafka-streams <= 3.0"); + } + + // Otherwise look for KafkaStreams.topologyMetadata (exists in kafka-streams >= 3.1) + Field topologyMetadataField = ReflectionUtils.findField(KafkaStreams.class, "topologyMetadata"); + if (topologyMetadataField != null) { + topologyMetadataField.setAccessible(true); + this.topologyInfoField = topologyMetadataField; + this.sourceTopicsForStoreMethod = ReflectionUtils.findMethod(TopologyMetadata.class, "sourceTopicsForStore", String.class); + logger.info("Using KafkaStreams.topologyMetadata.sourceTopicsForStore for kafka-streams >= 3.1"); + } + + if (this.sourceTopicsForStoreMethod == null) { + logger.warn("Could not find 'topologyMetadata.sourceTopicsForStore' or 'internalTopologyBuilder.sourceTopicsForStore' " + + "from KafkaStreams class - will be unable to reason about state stores."); + } + + } + + /** + * Determines if a state store is actually available to a KafkaStreams instance by + * querying the topology info source topics for the requested store. + * + * @param kafkaStreams the streams app + * @param storeName the name of the state store + * @return {@code true} if state store is available or {@code false} if the state store is + * not available or there was a problem reflecting on the topology info + */ + boolean streamsAppActuallyHasStore(KafkaStreams kafkaStreams, String storeName) { + if (this.sourceTopicsForStoreMethod == null) { + logger.warn("Unable to reason about state store because sourceTopicsForStore method was not found - returning false"); + return false; + } + try { + Object topologyInfo = ReflectionUtils.getField(this.topologyInfoField, kafkaStreams); + if (topologyInfo != null) { + logger.warn("Unable to reason about state store because topologyInfo field was null - returning false"); + return false; + } + Collection sourceTopicsForStore = (Collection) + ReflectionUtils.invokeMethod(this.sourceTopicsForStoreMethod, topologyInfo, storeName); + return !CollectionUtils.isEmpty(sourceTopicsForStore); + } + catch (Exception ex) { + logger.error(ex, () -> "Unable to reason about state store due to error: " + ex.getMessage() + " - returning false"); + } + return false; + } +}