From 109295464f2dd67c37457b26b9caa78a1476bc03 Mon Sep 17 00:00:00 2001 From: slamhan Date: Sun, 22 Apr 2018 12:49:20 +0800 Subject: [PATCH] QueryableStore retrieval stops at InvalidStateStoreException If there are multiple streams, there is a code path that throws a premature InvalidStateStoreException. Fixing that issue. Fixes #366 Polishing. --- .../kafka/streams/QueryableStoreRegistry.java | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/QueryableStoreRegistry.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/QueryableStoreRegistry.java index ea1f71c6b..15e23c03b 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/QueryableStoreRegistry.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/QueryableStoreRegistry.java @@ -20,6 +20,7 @@ import java.util.HashSet; import java.util.Set; import org.apache.kafka.streams.KafkaStreams; +import org.apache.kafka.streams.errors.InvalidStateStoreException; import org.apache.kafka.streams.state.QueryableStoreType; /** @@ -27,6 +28,7 @@ import org.apache.kafka.streams.state.QueryableStoreType; * the user applications. * * @author Soby Chacko + * @author Renwei Han * @since 2.0.0 */ public class QueryableStoreRegistry { @@ -44,9 +46,14 @@ public class QueryableStoreRegistry { public T getQueryableStoreType(String storeName, QueryableStoreType storeType) { for (KafkaStreams kafkaStream : kafkaStreams) { - T store = kafkaStream.store(storeName, storeType); - if (store != null) { - return store; + try{ + T store = kafkaStream.store(storeName, storeType); + if (store != null) { + return store; + } + } + catch (InvalidStateStoreException ignored) { + //pass through } } return null;