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

QueryableStore retrieval stops at InvalidStateStoreException

Browse Source 
If there are multiple streams, there is a code path that throws
a premature InvalidStateStoreException. Fixing that issue.

Fixes #366

Polishing.
This commit is contained in:
slamhan
2018-04-22 12:49:20 +08:00
committed by Soby Chacko
parent cf41a8a4eb
commit 109295464f
1 changed files with 10 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/QueryableStoreRegistry.java
View File

@@ -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> T getQueryableStoreType(String storeName, QueryableStoreType<T> 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;