GH-2523: Port KafkaStreamsVersionAgnosticTopologyInfoFacade from 3.2.x

See #2523

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

@@ -16,7 +16,6 @@
 
 package org.springframework.cloud.stream.binder.kafka.streams;
 
-import java.lang.reflect.Field;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -34,7 +33,6 @@ 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.processor.internals.TopologyMetadata;
 import org.apache.kafka.streams.state.HostInfo;
 import org.apache.kafka.streams.state.QueryableStoreType;
 import org.apache.kafka.streams.state.StreamsMetadata;
@@ -44,7 +42,6 @@ import org.springframework.retry.RetryPolicy;
 import org.springframework.retry.backoff.FixedBackOffPolicy;
 import org.springframework.retry.policy.SimpleRetryPolicy;
 import org.springframework.retry.support.RetryTemplate;
-import org.springframework.util.ReflectionUtils;
 import org.springframework.util.StringUtils;
 
 /**
@@ -68,7 +65,7 @@ public class InteractiveQueryService {
 
 	private final KafkaStreamsBinderConfigurationProperties binderConfigurationProperties;
 
-	private final Field topologyMetadataField;
+	private final KafkaStreamsVersionAgnosticTopologyInfoFacade topologyInfoFacade;
 
 	/**
 	 * Constructor for InteractiveQueryService.
@@ -79,8 +76,7 @@ public class InteractiveQueryService {
 			KafkaStreamsBinderConfigurationProperties binderConfigurationProperties) {
 		this.kafkaStreamsRegistry = kafkaStreamsRegistry;
 		this.binderConfigurationProperties = binderConfigurationProperties;
-		this.topologyMetadataField = ReflectionUtils.findField(KafkaStreams.class, "topologyMetadata");
-		this.topologyMetadataField.setAccessible(true);
+		this.topologyInfoFacade = new KafkaStreamsVersionAgnosticTopologyInfoFacade();
 	}
 
 	/**
@@ -134,7 +130,7 @@ public class InteractiveQueryService {
 			if (candidateStores.size() > 1) {
 
 				candidateStores = candidateStores.entrySet().stream()
-						.filter((e) -> storeActuallyAvailable(e.getKey(), storeName))
+						.filter((e) -> this.topologyInfoFacade.streamsAppActuallyHasStore(e.getKey(), storeName))
 						.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
 
 				if (candidateStores.size() == 1) {
@@ -150,17 +146,6 @@ public class InteractiveQueryService {
 		});
 	}
 
-	private boolean storeActuallyAvailable(KafkaStreams kafkaStreams, String storeName) {
-		try {
-			TopologyMetadata topologyMetadata = (TopologyMetadata) ReflectionUtils.getField(topologyMetadataField, kafkaStreams);
-			return !topologyMetadata.sourceTopicsForStore(storeName, null).isEmpty();
-		}
-		catch (Exception ex) {
-			LOG.error("Unable to determine if store (" + storeName + ") is available to the app due to: " + ex.getMessage(), ex);
-		}
-		return false;
-	}
-
 	/**
 	 * Retrieves the current {@link KafkaStreams} context if executing Thread is created by a Streams App (contains a matching application id in Thread's name).
 	 *

binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsVersionAgnosticTopologyInfoFacade.java

@@ -0,0 +1,130 @@
+/*
+ * 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.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.Collection;
+import java.util.StringJoiner;
+
+import org.apache.kafka.streams.KafkaStreams;
+
+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;
+
+	private boolean sourceTopicsForStoreMethodHasTwoArgs;
+
+	KafkaStreamsVersionAgnosticTopologyInfoFacade() {
+		this(KafkaStreams.class);
+	}
+
+	KafkaStreamsVersionAgnosticTopologyInfoFacade(Class<?> rootClass) {
+
+		// First look for KafkaStreams.internalTopologyBuilder (exists in kafka-streams <= 3.0)
+		Field internalTopologyBuilderField = ReflectionUtils.findField(rootClass, "internalTopologyBuilder");
+		if (internalTopologyBuilderField != null) {
+			internalTopologyBuilderField.setAccessible(true);
+			this.topologyInfoField = internalTopologyBuilderField;
+			this.sourceTopicsForStoreMethod = ReflectionUtils.findMethod(internalTopologyBuilderField.getType(), "sourceTopicsForStore", String.class);
+		}
+
+		// Otherwise look for KafkaStreams.topologyMetadata (exists in kafka-streams >= 3.1)
+		if (this.sourceTopicsForStoreMethod == null) {
+			Field topologyMetadataField = ReflectionUtils.findField(rootClass, "topologyMetadata");
+			if (topologyMetadataField != null) {
+				topologyMetadataField.setAccessible(true);
+				this.topologyInfoField = topologyMetadataField;
+				this.sourceTopicsForStoreMethod = ReflectionUtils.findMethod(topologyMetadataField.getType(), "sourceTopicsForStore", String.class);
+				if (this.sourceTopicsForStoreMethod == null) {
+					// The sourceTopicsForStore method has extra arg in kafka-streams >= 3.3
+					this.sourceTopicsForStoreMethod = ReflectionUtils.findMethod(topologyMetadataField.getType(), "sourceTopicsForStore", String.class, String.class);
+					this.sourceTopicsForStoreMethodHasTwoArgs = true;
+				}
+			}
+		}
+
+		if (this.sourceTopicsForStoreMethod != null) {
+			this.sourceTopicsForStoreMethod.setAccessible(true);
+			logger.info(() -> "Using " + methodDescription(this.sourceTopicsForStoreMethod));
+		}
+		else {
+			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;
+			}
+			Object[] args = this.sourceTopicsForStoreMethodHasTwoArgs ?
+					new Object[] { storeName, null } : new Object[] { storeName };
+			Collection<String> sourceTopicsForStore = (Collection<String>)
+					ReflectionUtils.invokeMethod(this.sourceTopicsForStoreMethod, topologyInfo, args);
+			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;
+	}
+
+	private String methodDescription(Method method) {
+		StringJoiner sj = new StringJoiner(",", method.getName() + "(", ")");
+		for (Class<?> parameterType : method.getParameterTypes()) {
+			sj.add(parameterType.getTypeName());
+		}
+		return "method " + method.getDeclaringClass().getTypeName() + '.' + sj.toString();
+	}
+}

binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsVersionAgnosticTopologyInfoFacadeTests.java

@@ -0,0 +1,242 @@
+/*
+ * 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.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Properties;
+
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link KafkaStreamsVersionAgnosticTopologyInfoFacade}.
+ *
+ * @author Chris Bono
+ */
+class KafkaStreamsVersionAgnosticTopologyInfoFacadeTests {
+
+	private static final Properties STREAM_PROPS = new Properties();
+
+	private static final StreamsBuilder STREAM_BUILDER = new StreamsBuilder();
+
+	static {
+		STREAM_PROPS.put(StreamsConfig.APPLICATION_ID_CONFIG, "topology-facade-tests-app");
+		STREAM_PROPS.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+		STREAM_BUILDER.<String, String>stream("foo").to("bar");
+	}
+
+	@Nested
+	class KafkaStreams30 {
+
+		@Test
+		void sourceTopicsForStoreWithTopics() {
+			KafkaStreamsVersionAgnosticTopologyInfoFacade facade =
+					new KafkaStreamsVersionAgnosticTopologyInfoFacade(TestKafkaStreams30.class);
+			assertThat(facade.streamsAppActuallyHasStore(new TestKafkaStreams30("topic1"), "store1")).isTrue();
+		}
+
+		@Test
+		void sourceTopicsForStoreWithNoTopics() {
+			KafkaStreamsVersionAgnosticTopologyInfoFacade facade =
+					new KafkaStreamsVersionAgnosticTopologyInfoFacade(TestKafkaStreams30.class);
+			assertThat(facade.streamsAppActuallyHasStore(new TestKafkaStreams30(), "store1")).isFalse();
+		}
+
+	}
+
+	@Nested
+	class KafkaStreams31_32 {
+
+		@Test
+		void sourceTopicsForStoreWithTopics() {
+			KafkaStreamsVersionAgnosticTopologyInfoFacade facade =
+					new KafkaStreamsVersionAgnosticTopologyInfoFacade(TestKafkaStreams31_32.class);
+			assertThat(facade.streamsAppActuallyHasStore(new TestKafkaStreams31_32("topic1"), "store1")).isTrue();
+		}
+
+		@Test
+		void sourceTopicsForStoreWithNoTopics() {
+			KafkaStreamsVersionAgnosticTopologyInfoFacade facade =
+					new KafkaStreamsVersionAgnosticTopologyInfoFacade(TestKafkaStreams31_32.class);
+			assertThat(facade.streamsAppActuallyHasStore(new TestKafkaStreams31_32(), "store1")).isFalse();
+		}
+
+	}
+
+	@Nested
+	class KafkaStreams33 {
+
+		@Test
+		void sourceTopicsForStoreWithTopics() {
+			KafkaStreamsVersionAgnosticTopologyInfoFacade facade =
+					new KafkaStreamsVersionAgnosticTopologyInfoFacade(TestKafkaStreams33.class);
+			assertThat(facade.streamsAppActuallyHasStore(new TestKafkaStreams33("topic1"), "store1")).isTrue();
+		}
+
+		@Test
+		void sourceTopicsForStoreWithNoTopics() {
+			KafkaStreamsVersionAgnosticTopologyInfoFacade facade =
+					new KafkaStreamsVersionAgnosticTopologyInfoFacade(TestKafkaStreams33.class);
+			assertThat(facade.streamsAppActuallyHasStore(new TestKafkaStreams33(), "store1")).isFalse();
+		}
+	}
+
+	@Nested
+	class NegativeCases {
+		@Test
+		void nullTopologyInfo() {
+			KafkaStreamsVersionAgnosticTopologyInfoFacade facade =
+					new KafkaStreamsVersionAgnosticTopologyInfoFacade(TestKafkaStreams30.class);
+			TestToplogyInfoOneArg internalTopologyBuilder = null;
+			assertThat(facade.streamsAppActuallyHasStore(new TestKafkaStreams30(internalTopologyBuilder), "store1")).isFalse();
+		}
+
+		@Test
+		void sourceTopicsForStoreMethodNotFound() {
+			KafkaStreamsVersionAgnosticTopologyInfoFacade facade =
+					new KafkaStreamsVersionAgnosticTopologyInfoFacade(TestKafkaStreamsNoSourceTopicsMethod.class);
+			assertThat(facade.streamsAppActuallyHasStore(new TestKafkaStreamsNoSourceTopicsMethod(), "store1")).isFalse();
+		}
+
+
+		@Test
+		void sourceTopicsForStoreThrowsException() {
+			KafkaStreamsVersionAgnosticTopologyInfoFacade facade =
+					new KafkaStreamsVersionAgnosticTopologyInfoFacade(TestKafkaStreamsThrowsError.class);
+			assertThat(facade.streamsAppActuallyHasStore(new TestKafkaStreamsThrowsError(), "store1")).isFalse();
+		}
+	}
+
+	/**
+	 * A test version of KafkaStreams as it exists in kafka-streams 3.0 w/ a topology
+	 * info field named 'internalTopologyBuilder' that in turn has a single-arg version
+	 * of sourceTopicsForStore(String)' that holds the info we need.
+	 */
+	static class TestKafkaStreams30 extends KafkaStreams {
+
+		private TestToplogyInfoOneArg internalTopologyBuilder;
+
+		TestKafkaStreams30(String... topics) {
+			this(new TestToplogyInfoOneArg(topics));
+		}
+
+		TestKafkaStreams30(TestToplogyInfoOneArg internalTopologyBuilder) {
+			super(STREAM_BUILDER.build(), STREAM_PROPS);
+			this.internalTopologyBuilder = internalTopologyBuilder;
+		}
+	}
+
+	/**
+	 * A test version of KafkaStreams as it exists in kafka-streams [3.1,3.2] w/ a topology
+	 * info field named 'topologyMetadata' that in turn has a single-arg version
+	 * of sourceTopicsForStore(String)' that holds the info we need.
+	 */
+	static class TestKafkaStreams31_32 extends KafkaStreams {
+
+		private TestToplogyInfoOneArg topologyMetadata;
+
+		TestKafkaStreams31_32(String... topics) {
+			super(STREAM_BUILDER.build(), STREAM_PROPS);
+			this.topologyMetadata = new TestToplogyInfoOneArg(topics);
+		}
+	}
+
+	/**
+	 * A test version of KafkaStreams as it exists in kafka-streams 3.3+ w/ a topology
+	 * info field named 'topologyMetadata' that in turn has a two-arg version
+	 * of sourceTopicsForStore(String,String)' that holds the info we need.
+	 */
+	static class TestKafkaStreams33 extends KafkaStreams {
+
+		private TestToplogyInfoTwoArgs topologyMetadata;
+
+		TestKafkaStreams33(String... topics) {
+			super(STREAM_BUILDER.build(), STREAM_PROPS);
+			this.topologyMetadata = new TestToplogyInfoTwoArgs(topics);
+		}
+	}
+
+	static class TestToplogyInfoOneArg {
+		private String[] sourceTopics;
+
+		TestToplogyInfoOneArg(String... sourceTopics) {
+			this.sourceTopics = sourceTopics;
+		}
+
+		public Collection<String> sourceTopicsForStore(String storeName) {
+			return this.sourceTopics == null ? Collections.emptyList() : Arrays.asList(this.sourceTopics);
+		}
+	}
+
+	static class TestToplogyInfoTwoArgs {
+		private String[] sourceTopics;
+
+		TestToplogyInfoTwoArgs(String... sourceTopics) {
+			this.sourceTopics = sourceTopics;
+		}
+
+		public Collection<String> sourceTopicsForStore(String storeName, String topologyName) {
+			return this.sourceTopics == null ? Collections.emptyList() : Arrays.asList(this.sourceTopics);
+		}
+	}
+
+	/**
+	 * A test version of KafkaStreams w/ a topology info field named 'internalTopologyBuilder'
+	 * that in turn has NO 'sourceTopicsForStore' method to use.
+	 */
+	static class TestKafkaStreamsNoSourceTopicsMethod extends KafkaStreams {
+
+		private TestTopologyInfoNoSourceTopicsMethod internalTopologyBuilder = new TestTopologyInfoNoSourceTopicsMethod();
+
+		TestKafkaStreamsNoSourceTopicsMethod() {
+			super(STREAM_BUILDER.build(), STREAM_PROPS);
+		}
+
+		static class TestTopologyInfoNoSourceTopicsMethod {
+
+		}
+
+	}
+
+	/**
+	 * A test version of KafkaStreams w/ a topology info field named 'internalTopologyBuilder'
+	 * that in turn has a single-arg version of sourceTopicsForStore(String)' that throws
+	 * an exception when invoked.
+	 */
+	static class TestKafkaStreamsThrowsError extends KafkaStreams {
+
+		private TestToplogyInfoOneArg internalTopologyBuilder = new TestToplogyInfoOneArg() {
+			@Override
+			public Collection<String> sourceTopicsForStore(String storeName) {
+				throw new RuntimeException("BOOM: " + storeName);
+			}
+		};
+
+		TestKafkaStreamsThrowsError() {
+			super(STREAM_BUILDER.build(), STREAM_PROPS);
+		}
+
+	}
+
+}