From 50f4470fcf63e51e6caa39fb03554f1c9dc2674f Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 28 Oct 2020 15:23:30 -0400 Subject: [PATCH] Update deprecated API usage in InteractiveQueryService Use queryMetadataForKey instead of metadataForKey Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/941 --- .../streams/InteractiveQueryService.java | 27 +++++++++++++++---- ...reamsInteractiveQueryIntegrationTests.java | 7 +++++ 2 files changed, 29 insertions(+), 5 deletions(-) diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryService.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryService.java index deeee8ffd..532c84569 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryService.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryService.java @@ -28,6 +28,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.streams.KafkaStreams; +import org.apache.kafka.streams.KeyQueryMetadata; import org.apache.kafka.streams.errors.InvalidStateStoreException; import org.apache.kafka.streams.state.HostInfo; import org.apache.kafka.streams.state.QueryableStoreType; @@ -146,17 +147,33 @@ public class InteractiveQueryService { * @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 + * @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() + final KeyQueryMetadata keyQueryMetadata = this.kafkaStreamsRegistry.getKafkaStreams() .stream() - .map((k) -> Optional.ofNullable(k.metadataForKey(store, key, serializer))) + .map((k) -> Optional.ofNullable(k.queryMetadataForKey(store, key, serializer))) .filter(Optional::isPresent).map(Optional::get).findFirst().orElse(null); - return streamsMetadata != null ? streamsMetadata.hostInfo() : null; + return keyQueryMetadata != null ? keyQueryMetadata.getActiveHost() : null; } + /** + * Retrieves the {@link KeyQueryMetadata} associated with the given combination of key and state store. + * + * @param generic type for key + * @param store store name + * @param key key to look for + * @param serializer {@link Serializer} for the key + * @return the {@link KeyQueryMetadata} if available, null otherwise. + */ + public KeyQueryMetadata getKeyQueryMetadata(String store, K key, Serializer serializer) { + return this.kafkaStreamsRegistry.getKafkaStreams() + .stream() + .map((k) -> Optional.ofNullable(k.queryMetadataForKey(store, key, serializer))) + .filter(Optional::isPresent).map(Optional::get).findFirst().orElse(null); + } + + /** * Gets the list of {@link HostInfo} where the provided store is hosted on. * It also can include current host info. 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 f2608982c..13889d14f 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 @@ -25,6 +25,7 @@ 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.KafkaStreams; +import org.apache.kafka.streams.KeyQueryMetadata; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Materialized; @@ -167,6 +168,12 @@ public class KafkaStreamsInteractiveQueryIntegrationTests { assertThat(currentHostInfo.host() + ":" + currentHostInfo.port()) .isEqualTo(embeddedKafka.getBrokersAsString()); + final KeyQueryMetadata keyQueryMetadata = interactiveQueryService.getKeyQueryMetadata("prod-id-count-store", + 123, new IntegerSerializer()); + final HostInfo activeHost = keyQueryMetadata.getActiveHost(); + assertThat(activeHost.host() + ":" + activeHost.port()) + .isEqualTo(embeddedKafka.getBrokersAsString()); + HostInfo hostInfo = interactiveQueryService.getHostInfo("prod-id-count-store", 123, new IntegerSerializer()); assertThat(hostInfo.host() + ":" + hostInfo.port())