Updating Spring Boot to v2.3.3 and Kafka Streams to v2.6.0

Author: Ben Chen

README.md

@@ -1,4 +1,4 @@
-# Spring Boot (2.3.2) RESTful API with Kafka Streams (2.5.0)
+# Spring Boot (2.3.3) RESTful API with Kafka Streams (2.6.0)
 
 While looking through the Kafka Tutorials to see how I could setup a Spring Boot API project with Kafka Streams, I found it strange that there wasn't a complete or more informative example on how this could be achieved. Most use cases demonstrated how to compute aggregations and how to build simple topologies, but it was difficult to find a concrete example on how to build an API service that could query into these materialized name stores. Anyways, I thought I’d create my own using a more recent version of Spring Boot with Java 14.
 
@@ -71,7 +71,7 @@ After the application runs, navigate to [http://localhost:7001/swagger-ui/index.
 ```json
 {
   "movieId": 362,
-  "rating": 9.0
+  "rating": 9
 }
 ```
 

pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>org.springframework.boot</groupId>
 		<artifactId>spring-boot-starter-parent</artifactId>
-		<version>2.3.2.RELEASE</version>
+		<version>2.3.3.RELEASE</version>
 		<relativePath/>
 	</parent>
 	<groupId>com.example.kafkastreams.restapi</groupId>
@@ -23,7 +23,7 @@
 	<properties>
 		<java.version>14</java.version>
 		<avro.version>1.10.0</avro.version>
-		<kafka.version>2.5.0</kafka.version>
+		<kafka.version>2.6.0</kafka.version>
 		<openapi.version>1.4.3</openapi.version>
 	</properties>
 	<dependencies>

src/main/java/com/example/kafkastreams/restapi/springbootapp/controller/v1/MovieController.java

@@ -23,9 +23,11 @@
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.*;
 
-@OpenAPIDefinition(servers = { @Server(url = "http://localhost:7001") }, info = @Info(title = "Sample Spring Boot Kafka Stream API", version = "v1", description = "A demo project using Spring Boot with Kafka Stream", license = @License(name = "MIT License", url = "https://github.com/bchen04/springboot-kafka-streams-rest-api/blob/master/LICENSE"), contact = @Contact(url = "https://www.linkedin.com/in/bchen04/", name = "Ben Chen")))
+@OpenAPIDefinition(servers = { @Server(url = "http://localhost:7001") }, info = @Info(title = "Sample Spring Boot Kafka Stream API", version = "v1", description = "A demo project using Spring Boot with Kafka Streams.", license = @License(name = "MIT License", url = "https://github.com/bchen04/springboot-kafka-streams-rest-api/blob/master/LICENSE"), contact = @Contact(url = "https://www.linkedin.com/in/bchen04/", name = "Ben Chen")))
 @RestController
 @RequestMapping("v1/movie")
 public class MovieController {
@@ -42,11 +44,13 @@ public MovieController(KafkaStreams streams) {
 
     @Operation(summary = "Returns the average rating for a particular movie")
     @ApiResponses(value = {
-            @ApiResponse(responseCode = "200", description = "successful operation", content = @Content(schema = @Schema(type = "object"))) })
+            @ApiResponse(responseCode = "200", description = "successful operation", content = @Content(schema = @Schema(type = "object"))),
+            @ApiResponse(responseCode = "500", description = "internal server error")})
     @GetMapping(value = "{movieId}/rating", produces = { "application/json" })
-    public MovieAverageRatingResponse getMovieAverageRating(@Parameter(required = true, example = "362") @PathVariable Long movieId) {
+    public ResponseEntity<MovieAverageRatingResponse> getMovieAverageRating(@Parameter(description = "Movie identifier", required = true, example = "362") @PathVariable Long movieId) {
         try {
             //find active, standby host list and partition for key
+            //get the metadata related to the key.
             final KeyQueryMetadata keyQueryMetadata = streams.queryMetadataForKey(stateStoreName, movieId, Serdes.Long().serializer());
 
             //use the above information to redirect the query to the host containing the partition for the key
@@ -64,11 +68,16 @@ public MovieAverageRatingResponse getMovieAverageRating(@Parameter(required = tr
             //get the value by key
             Double result = store.get(movieId);
 
-            return new MovieAverageRatingResponse(movieId, result);
+            return ResponseEntity
+                    .ok()
+                    .body(new MovieAverageRatingResponse(movieId, result));
         }
         catch(Exception ex) {
             logger.error("Failed due to exception: {}", ex.getMessage());
-            return null;
+
+            return ResponseEntity
+                    .status(HttpStatus.INTERNAL_SERVER_ERROR)
+                    .build();
         }
     }
 }