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 322fda359..330a1fef9 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,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). * 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..763030bc5 --- /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,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. + *

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 sourceTopicsForStore = (Collection) + 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(); + } +} diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsVersionAgnosticTopologyInfoFacadeTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsVersionAgnosticTopologyInfoFacadeTests.java new file mode 100644 index 000000000..0a859d577 --- /dev/null +++ b/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.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 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 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 sourceTopicsForStore(String storeName) { + throw new RuntimeException("BOOM: " + storeName); + } + }; + + TestKafkaStreamsThrowsError() { + super(STREAM_BUILDER.build(), STREAM_PROPS); + } + + } + +}