GH-2942: Kafka Streams queryable stores #3022

sobychacko · 2024-02-08T23:42:19Z

This is an initial iteration for providing a basic API around interactive query service in Kafka Streams. In this iteration, we introduce a single API for retrieving the queryable state store from the Kafka Streams topology, namely, retrieveQueryableStore.

Fixes: spring-projects#2942 This is an initial iteration for providing a basic API around interactive query service in Kafka Streams. In this iteration, we introduce a single API for retrieving the queryable state store from the Kafka Streams topology, namely, retrieveQueryableStore.

artembilan

Checkstyle violation:

 Error: eckstyle] [ERROR] /home/runner/work/spring-kafka/spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryService.java:96:25: Reference to instance variable 'retryTemplate' needs "this.". [RequireThis]
Error: eckstyle] [ERROR] /home/runner/work/spring-kafka/spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryService.java:98:25: Reference to instance variable 'retryTemplate' needs "this.". [RequireThis]

artembilan · 2024-02-09T14:38:59Z

...fka/src/main/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryService.java

+ * {@link KafkaStreams} under consideration.
+ *
+ * @author Soby Chacko
+ * @since 3.2.0


Just 3.2 for consistency with the rest of styles.

...fka/src/main/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryService.java

artembilan · 2024-02-09T14:41:05Z

...fka/src/main/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryService.java

+	private KafkaStreams kafkaStreams;
+
+	/**
+	 * Constructs an instance for querying state stores from the KafkaStreams in the {@link StreamsBuilderFactoryBean}.


Method Javadocs must be imperative: https://github.com/spring-projects/spring-framework/wiki/Code-Style#javadoc-formatting

Plus no blank lines in the method Javadocs.

...fka/src/main/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryService.java

artembilan · 2024-02-09T14:42:25Z

...fka/src/main/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryService.java

+	/**
+	 * Underlying {@link KafkaStreams} from {@link StreamsBuilderFactoryBean}.
+	 */
+	private KafkaStreams kafkaStreams;


I believe this has to be volatile since you initialize this property in lazy manner on demand.

artembilan · 2024-02-09T14:48:29Z

...fka/src/main/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryService.java

+		}
+		Assert.notNull(this.kafkaStreams, "KafkaStreams cannot be null");
+		StoreQueryParameters<T> storeQueryParams = StoreQueryParameters.fromNameAndType(storeName, storeType);
+		AtomicReference<StoreQueryParameters<T>> storeQueryParametersReference = new AtomicReference<>(storeQueryParams);


I'm not sure in this extra reference: that storeQueryParams is effectively already final.

...rc/test/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryServiceTests.java

artembilan · 2024-02-09T14:51:50Z

...rc/test/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryServiceTests.java

+						QueryableStoreTypes.keyValueStore());
+
+		assertThat(objectObjectReadOnlyKeyValueStore.get(123)).isNotNull();
+		assertThat((Long) objectObjectReadOnlyKeyValueStore.get(123) >= 1L).isTrue();


I don't think we need to call get() for the same value twice.
There is also isGreaterThanOrEqualTo() instead of that isTrue() for more precise reporting (if that).

artembilan · 2024-02-09T14:52:56Z

...rc/test/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryServiceTests.java

+					.retrieveQueryableStore(NON_EXISTENT_STORE, QueryableStoreTypes.keyValueStore());
+		}
+		catch (Exception e) {
+			exc = e;


I prefer to use assertThatException() instead of try..catch: must better reporting.

...rc/test/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryServiceTests.java

artembilan

I think we are good so far.
We can turn this into a normal PR and go ahead with docs.

Thanks

...fka/src/main/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryService.java

artembilan

I guess the whats-new.adoc has to mention this new feature as well.

Thanks

artembilan · 2024-02-12T21:04:56Z

...fka/src/main/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryService.java

+		if (this.kafkaStreams == null) {
+			this.kafkaStreams = this.streamsBuilderFactoryBean.getKafkaStreams();
+		}
+		Assert.notNull(this.kafkaStreams, "KafkaStreams cannot be null");


Let's improve this assert with something like ". Was provided StreamsBuilderFactoryBean been started?"!

artembilan · 2024-02-12T21:05:21Z

...fka/src/main/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryService.java

+
+	/**
+	 * Retrieve and return a queryable store by name created in the application.
+	 *


No blank line.

...fka/src/main/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryService.java

...rc/test/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryServiceTests.java

artembilan · 2024-02-12T21:38:17Z

spring-kafka-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc

+A new API `KafkaStreamsInteractiveQuerySupport` for accessing queryable stores used in Kafka Streams interactive queries.
+See xref:streams.adoc#kafka-streams-iq-support[Kafka Streams Interactive Support] for more details.
+
+A new `TransactionIdSuffixStrategy` interface was introduced to manage `transactional.id` suffix.


I believe this paragraph is a copy/paste artifact.
Will fix it shortly from GH GUI.

My bad. Sorry.

spring-kafka-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc

sobychacko requested a review from artembilan February 8, 2024 23:42

sobychacko force-pushed the gh-2942 branch from f6bfd80 to b10bcf9 Compare February 8, 2024 23:44

artembilan requested changes Feb 9, 2024

View reviewed changes

Addressing PR review comments

e38e7b8

artembilan requested changes Feb 9, 2024

View reviewed changes

...fka/src/main/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryService.java Show resolved Hide resolved

Adding docs

fee839e

sobychacko changed the title ~~[REVIEW-ONLY-DO-NOT-MERGE] GH-2942: Kafka Streams queryable stores~~ GH-2942: Kafka Streams queryable stores Feb 12, 2024

artembilan requested changes Feb 12, 2024

View reviewed changes

Addressing PR review

861cd72

artembilan requested changes Feb 12, 2024

View reviewed changes

Remove copied/pasted paragraph in docs

99a8796

artembilan enabled auto-merge (squash) February 12, 2024 21:56

artembilan merged commit f2e77e6 into spring-projects:main Feb 12, 2024

GH-2942: Kafka Streams queryable stores #3022

GH-2942: Kafka Streams queryable stores #3022

Uh oh!

Conversation

sobychacko commented Feb 8, 2024

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants