Merge pull request #728 from sobychacko/gh-706

Introducing retry for finding state stores.

docs/src/main/asciidoc/kafka-streams.adoc

@@ -952,6 +952,16 @@ ReadOnlyKeyValueStore<Object, Object> keyValueStore =
 						interactiveQueryService.getQueryableStoreType("my-store", QueryableStoreTypes.keyValueStore());
 ----
 
+During the startup, the above method call to retrieve the startup might fail.
+For e.g it might still be in the middle of initializing the state store.
+In such cases, it will be useful to retry this operation.
+Kafka Streams binder provides a simple retry mechanism to accommodate this.
+
+Following are the two properties that you can use to control this retrying.
+
+* spring.cloud.stream.binder.kafka.streams.stateStoreRetry.maxAttempts - Default is `1` .
+* spring.cloud.stream.binder.kafka.streams.stateStoreRetry.backOffInterval - Default is `1000` milliseconds.
+
 If there are multiple instances of the kafka streams application running, then before you can query them interactively, you need to identify which application instance hosts the key.
 `InteractiveQueryService` API provides methods for identifying the host information.
 

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

@@ -19,6 +19,8 @@ package org.springframework.cloud.stream.binder.kafka.streams;
 import java.util.Map;
 import java.util.Optional;
 
+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.errors.InvalidStateStoreException;
@@ -27,6 +29,10 @@ import org.apache.kafka.streams.state.QueryableStoreType;
 import org.apache.kafka.streams.state.StreamsMetadata;
 
 import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties;
+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.StringUtils;
 
 /**
@@ -41,6 +47,8 @@ import org.springframework.util.StringUtils;
  */
 public class InteractiveQueryService {
 
+	private static final Log LOG = LogFactory.getLog(InteractiveQueryService.class);
+
 	private final KafkaStreamsRegistry kafkaStreamsRegistry;
 
 	private final KafkaStreamsBinderConfigurationProperties binderConfigurationProperties;
@@ -64,18 +72,32 @@ public class InteractiveQueryService {
 	 * @return queryable store.
 	 */
 	public <T> T getQueryableStore(String storeName, QueryableStoreType<T> storeType) {
-		for (KafkaStreams kafkaStream : this.kafkaStreamsRegistry.getKafkaStreams()) {
-			try {
-				T store = kafkaStream.store(storeName, storeType);
-				if (store != null) {
-					return store;
+
+		RetryTemplate retryTemplate = new RetryTemplate();
+
+		KafkaStreamsBinderConfigurationProperties.StateStoreRetry stateStoreRetry = this.binderConfigurationProperties.getStateStoreRetry();
+		RetryPolicy retryPolicy = new SimpleRetryPolicy(stateStoreRetry.getMaxAttempts());
+		FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
+		backOffPolicy.setBackOffPeriod(stateStoreRetry.getBackoffPeriod());
+
+		retryTemplate.setBackOffPolicy(backOffPolicy);
+		retryTemplate.setRetryPolicy(retryPolicy);
+
+		return retryTemplate.execute(context -> {
+			T store;
+			for (KafkaStreams kafkaStream : InteractiveQueryService.this.kafkaStreamsRegistry.getKafkaStreams()) {
+				try {
+					store = kafkaStream.store(storeName, storeType);
+					if (store != null) {
+						return store;
+					}
+				}
+				catch (InvalidStateStoreException e) {
+					LOG.warn("Error when retrieving state store: " + storeName, e);
 				}
 			}
-			catch (InvalidStateStoreException ignored) {
-				// pass through
-			}
-		}
-		return null;
+			throw new IllegalStateException("Error when retrieving state store: " + storeName);
+		});
 	}
 
 	/**

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java

@@ -54,6 +54,16 @@ public class KafkaStreamsBinderConfigurationProperties
 
 	private String applicationId;
 
+	private StateStoreRetry stateStoreRetry = new StateStoreRetry();
+
+	public StateStoreRetry getStateStoreRetry() {
+		return stateStoreRetry;
+	}
+
+	public void setStateStoreRetry(StateStoreRetry stateStoreRetry) {
+		this.stateStoreRetry = stateStoreRetry;
+	}
+
 	public String getApplicationId() {
 		return this.applicationId;
 	}
@@ -79,4 +89,27 @@ public class KafkaStreamsBinderConfigurationProperties
 		this.serdeError = serdeError;
 	}
 
+	public static class StateStoreRetry {
+
+		private int maxAttempts = 1;
+
+		private long backoffPeriod = 1000;
+
+		public int getMaxAttempts() {
+			return maxAttempts;
+		}
+
+		public void setMaxAttempts(int maxAttempts) {
+			this.maxAttempts = maxAttempts;
+		}
+
+		public long getBackoffPeriod() {
+			return backoffPeriod;
+		}
+
+		public void setBackoffPeriod(long backoffPeriod) {
+			this.backoffPeriod = backoffPeriod;
+		}
+	}
+
 }

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

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.stream.binder.kafka.streams.integration;
+package org.springframework.cloud.stream.binder.kafka.streams;
 
 import java.util.Map;
 
@@ -23,25 +23,29 @@ import org.apache.kafka.clients.consumer.ConsumerConfig;
 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.KeyValue;
 import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.kstream.Materialized;
 import org.apache.kafka.streams.kstream.Serialized;
 import org.apache.kafka.streams.state.HostInfo;
+import org.apache.kafka.streams.state.QueryableStoreType;
 import org.apache.kafka.streams.state.QueryableStoreTypes;
 import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.Test;
+import org.mockito.Mockito;
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.WebApplicationType;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
 import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.annotation.StreamListener;
-import org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService;
 import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsProcessor;
+import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
@@ -54,6 +58,7 @@ import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.messaging.handler.annotation.SendTo;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.internal.verification.VerificationModeFactory.times;
 
 /**
  * @author Soby Chacko
@@ -86,6 +91,29 @@ public class KafkaStreamsInteractiveQueryIntegrationTests {
 		consumer.close();
 	}
 
+	@Test
+	public void testStateStoreRetrievalRetry() {
+
+		KafkaStreams mock = Mockito.mock(KafkaStreams.class);
+		KafkaStreamsRegistry kafkaStreamsRegistry = new KafkaStreamsRegistry();
+		kafkaStreamsRegistry.registerKafkaStreams(mock);
+		KafkaStreamsBinderConfigurationProperties binderConfigurationProperties =
+				new KafkaStreamsBinderConfigurationProperties(new KafkaProperties());
+		binderConfigurationProperties.getStateStoreRetry().setMaxAttempts(3);
+		InteractiveQueryService interactiveQueryService = new InteractiveQueryService(kafkaStreamsRegistry,
+				binderConfigurationProperties);
+
+		QueryableStoreType<ReadOnlyKeyValueStore<Object, Object>> storeType = QueryableStoreTypes.keyValueStore();
+		try {
+			interactiveQueryService.getQueryableStore("foo", storeType);
+		}
+		catch (Exception ignored) {
+
+		}
+
+		Mockito.verify(mock, times(3)).store("foo", storeType);
+	}
+
 	@Test
 	public void testKstreamBinderWithPojoInputAndStringOuput() throws Exception {
 		SpringApplication app = new SpringApplication(ProductCountApplication.class);