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

GH-2523: Smarter state store retrieval in InteractiveQueryService

Browse Source 
See #2523
This commit is contained in:
Chris Bono
2022-10-14 10:20:40 -05:00
parent faae7705c4
commit 80e300fa61
2 changed files with 140 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<T> 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<KafkaStreams, T> 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> 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).
*

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

@@ -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.
* <p>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<String> sourceTopicsForStore = (Collection<String>)
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;
}
}