diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryServiceMultiStateStoreTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryServiceMultiStateStoreTests.java index 4b475ac33..1daa5dcc5 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryServiceMultiStateStoreTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryServiceMultiStateStoreTests.java @@ -63,7 +63,19 @@ class InteractiveQueryServiceMultiStateStoreTests { private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); @Test - void stateStoreOnlyAvailableOnKafkaStreamsAppWhereItIsUsed() { + void stateStoreAvailableOnProperAppWhenAppServerPropertySet() { + stateStoreAvailableOnProperApp(true); + } + + @Test + void stateStoreAvailableOnProperAppWhenAppServerPropertyNotSet() { + stateStoreAvailableOnProperApp(false); + } + + private void stateStoreAvailableOnProperApp(boolean shouldSetAppServerProperty) { + String appServerArg = shouldSetAppServerProperty ? + "--spring.cloud.stream.kafka.streams.binder.configuration.application.server=" + embeddedKafka.getBrokersAsString() : + "--foo=bar"; try (ConfigurableApplicationContext context = new SpringApplicationBuilder() .sources(MultipleAppsWithUsedStateStoresTestApplication.class) .web(WebApplicationType.NONE) @@ -74,16 +86,16 @@ class InteractiveQueryServiceMultiStateStoreTests { "--spring.cloud.stream.function.bindings.app2-in-0=input2", "--spring.cloud.stream.kafka.streams.binder.functions.app1.application-id=stateStoreTestApp1", "--spring.cloud.stream.kafka.streams.binder.functions.app2.application-id=stateStoreTestApp2", - "--spring.cloud.stream.kafka.streams.binder.configuration.application.server=" - + embeddedKafka.getBrokersAsString(), - "--spring.cloud.stream.kafka.streams.binder.brokers=" - + embeddedKafka.getBrokersAsString()) + appServerArg, + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString()) ) { waitForRunningStreams(context.getBean(KafkaStreamsRegistry.class)); + + InteractiveQueryService queryService = context.getBean(InteractiveQueryService.class); + // The KafkaStreams.store() used by query service is non-deterministic so perform the operation multiple times to // surface any possible issues. Also, no need to actually write anything to the stores, the store.get() call will // cause a failure when the state store is invalid. - InteractiveQueryService queryService = context.getBean(InteractiveQueryService.class); for (int i = 0; i < 100; i++) { assertThat(queryService.getQueryableStore(STORE_1_NAME, QueryableStoreTypes.keyValueStore()) .get("someKey")).isNull(); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsInteractiveQueryIntegrationTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsInteractiveQueryIntegrationTests.java index 023a5a476..e28e2f388 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsInteractiveQueryIntegrationTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsInteractiveQueryIntegrationTests.java @@ -95,7 +95,7 @@ public class KafkaStreamsInteractiveQueryIntegrationTests { } @Test - void testStateStoreRetrievalRetry() { + void stateStoreRetrievalRetriedOnFailure() { StreamsBuilderFactoryBean mock = Mockito.mock(StreamsBuilderFactoryBean.class); KafkaStreams mockKafkaStreams = Mockito.mock(KafkaStreams.class); @@ -124,7 +124,7 @@ public class KafkaStreamsInteractiveQueryIntegrationTests { } @Test - void testStateStoreRetrievalRetryForHostInfoService() { + void hostInfoRetrievalRetriedOnFailure() { StreamsBuilderFactoryBean mock = Mockito.mock(StreamsBuilderFactoryBean.class); KafkaStreams mockKafkaStreams = Mockito.mock(KafkaStreams.class); Mockito.when(mock.getKafkaStreams()).thenReturn(mockKafkaStreams); @@ -153,7 +153,7 @@ public class KafkaStreamsInteractiveQueryIntegrationTests { } @Test - void testKstreamBinderWithPojoInputAndStringOuput() { + void queryServiceProvidesStateStoreAndHostInfo() { SpringApplication app = new SpringApplication(ProductCountApplication.class); app.setWebApplicationType(WebApplicationType.NONE); ConfigurableApplicationContext context = app.run("--server.port=0", @@ -167,7 +167,7 @@ public class KafkaStreamsInteractiveQueryIntegrationTests { + "=org.apache.kafka.common.serialization.Serdes$StringSerde", "--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" + "=org.apache.kafka.common.serialization.Serdes$StringSerde", - "--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId=ProductCountApplication-abc", + "--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId=ProductCountApplication-def", "--spring.cloud.stream.kafka.streams.binder.configuration.application.server" + "=" + embeddedKafka.getBrokersAsString(), "--spring.cloud.stream.kafka.streams.binder.brokers=" @@ -187,13 +187,13 @@ public class KafkaStreamsInteractiveQueryIntegrationTests { KafkaTemplate template = new KafkaTemplate<>(pf, true); template.setDefaultTopic("foos"); template.sendDefault("{\"id\":\"123\"}"); - ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, - "counts-id"); - assertThat(cr.value().contains("Count for product with ID 123: 1")).isTrue(); - ProductCountApplication.Foo foo = context - .getBean(ProductCountApplication.Foo.class); - assertThat(foo.getProductStock(123).equals(1L)); + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, "counts-id"); + assertThat(cr.value()).isIn("Count for product with ID 123: 1", "Count for product with ID 123: 2"); + + // Make sure the count can be retrieved from local state store + ProductCountApplication.Foo foo = context.getBean(ProductCountApplication.Foo.class); + assertThat(foo.getProductStock(123)).isIn(1L, 2L); // perform assertions on HostInfo related methods in InteractiveQueryService InteractiveQueryService interactiveQueryService = context @@ -232,6 +232,41 @@ public class KafkaStreamsInteractiveQueryIntegrationTests { .isEqualTo(embeddedKafka.getBrokersAsString()); } + @Test + void queryServiceProvidesStateStoreWithoutSettingAppServerProperty() { + SpringApplication app = new SpringApplication(ProductCountApplication.class); + app.setWebApplicationType(WebApplicationType.NONE); + try (ConfigurableApplicationContext context = app.run("--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.function.bindings.process-in-0=input", + "--spring.cloud.stream.function.bindings.process-out-0=output", + "--spring.cloud.stream.bindings.input.destination=foos", + "--spring.cloud.stream.bindings.output.destination=counts-id", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" + + "=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" + + "=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId=ProductCountApplication-def", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString()) + ) { + // Send an input product record + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("foos"); + template.sendDefault("{\"id\":\"123\"}"); + + // Wait for output from the streams app + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, "counts-id"); + assertThat(cr.value()).isIn("Count for product with ID 123: 1", "Count for product with ID 123: 2"); + + // Make sure the count can be retrieved from local state store + ProductCountApplication.Foo foo = context.getBean(ProductCountApplication.Foo.class); + assertThat(foo.getProductStock(123)).isIn(1L, 2L); + } + } + @EnableAutoConfiguration public static class ProductCountApplication { @@ -243,8 +278,7 @@ public class KafkaStreamsInteractiveQueryIntegrationTests { .groupByKey(Grouped.with(new Serdes.IntegerSerde(), new JsonSerde<>(Product.class))) .count(Materialized.as("prod-id-count-store")).toStream() - .map((key, value) -> new KeyValue<>(null, - "Count for product with ID 123: " + value)); + .map((key, value) -> new KeyValue<>(null, "Count for product with ID 123: " + value)); } @Bean