Fix code smells

README.md

@@ -27,7 +27,11 @@ From the root of the project execute the below commands
 ```shell script
 ./gradlew bootRun -PjvmArgs="-D--spring.config.location=config/dev.properties"
 ```
-
+- Run SonarQube Code Analysis
+```
+docker run -d --rm --name sonarqube -e SONAR_ES_BOOTSTRAP_CHECKS_DISABLE=true -p 9000:9000 sonarqube:community
+./gradlew sonarqube
+```
 ## Run native
 ```shell script
 java -jar build/libs/spring-telemetry-receiver-1.0.jar --spring.config.location=config/dev.properties

build.gradle

@@ -1,6 +1,8 @@
 plugins {
     id "java"
+    id "jacoco"
     id "org.springframework.boot" version "${springBootVersion}"
+    id "org.sonarqube" version "${sonarQubeVersion}"
 }
 
 group "${projectGroup}"
@@ -22,6 +24,7 @@ dependencies {
 
     testImplementation group: "org.springframework.boot", name: "spring-boot-starter-test", version:"${springBootVersion}"
     testImplementation group: "org.springframework.kafka", name: "spring-kafka-test", version:"${springKafkaVersion}"
+    testImplementation group: "org.awaitility", name: "awaitility", version: "${awaitilityVersion}"
 }
 
 springBoot {
@@ -34,6 +37,7 @@ compileJava {
 
 test {
     useJUnitPlatform()
+    finalizedBy jacocoTestReport
 }
 
 bootBuildImage {
@@ -45,3 +49,26 @@ bootRun {
         jvmArgs project.jvmArgs.split('\\s+')
     }
 }
+
+jacocoTestReport {
+    reports {
+        xml.enabled true
+        html.enabled true
+        csv.enabled false
+        xml.destination file("${buildDir}/jacoco/xml/jacoco.xml")
+        html.destination file("${buildDir}/jacoco/html")
+    }
+    dependsOn test
+}
+
+sonarqube {
+    properties {
+        property "sonar.sourceEncoding", "UTF-8"
+        property "sonar.exclusions", "**/Application.java, **/dto/**"
+        property "sonar.java.source", "${project.sourceCompatibility}"
+        property "sonar.java.target", "${project.targetCompatibility}"
+        property "sonar.junit.reportsPath", "${buildDir}/reports/tests"
+        property "sonar.coverage.jacoco.xmlReportPaths", "${buildDir}/jacoco/xml/jacoco.xml"
+    }
+}
+tasks.sonarqube.dependsOn test

config/dev.properties

@@ -1,6 +1,7 @@
 server.port=9080
 spring.main.banner-mode=off
 debug=false
+server.error.whitelabel.enabled=false
 
 spring.kafka.bootstrap-servers=localhost:9092
 spring.kafka.producer.topic=telemetry

config/prod.properties

@@ -1,6 +1,7 @@
 server.port=9080
 spring.main.banner-mode=off
 debug=false
+server.error.whitelabel.enabled=false
 
 spring.kafka.bootstrap-servers=localhost:9092
 spring.kafka.producer.topic=telemetry

config/qa.properties

@@ -1,6 +1,7 @@
 server.port=9080
 spring.main.banner-mode=off
 debug=false
+server.error.whitelabel.enabled=false
 
 spring.kafka.bootstrap-servers=localhost:9092
 spring.kafka.producer.topic=telemetry

gradle.properties

@@ -1,9 +1,13 @@
 springBootVersion=2.4.0
 springKafkaVersion=2.6.3
 lombokVersion=1.18.16
+awaitilityVersion=4.0.3
+sonarQubeVersion=3.0
 
 applicationClass=com.barrelsofdata.springexamples.Application
 projectGroup=com.barrelsofdata.springexamples
 projectVersion=1.0
 
+systemProp.sonar.host.url=http://localhost:9000
+
 org.gradle.daemon=false

src/main/java/com/barrelsofdata/springexamples/controller/ApplicationErrorController.java

@@ -8,7 +8,7 @@ import org.springframework.boot.web.servlet.error.ErrorController;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Controller;
-import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.context.request.RequestAttributes;
 import org.springframework.web.context.request.WebRequest;
 
@@ -22,7 +22,7 @@ public class ApplicationErrorController implements ErrorController {
     @Autowired
     private ErrorAttributes errorAttributes;
 
-    @RequestMapping("/error")
+    @GetMapping(value = "/error")
     public ResponseEntity<ExceptionDto> handleError(WebRequest request) {
         Map<String, Object> requestErrors = errorAttributes.getErrorAttributes(request, ErrorAttributeOptions.defaults());
         Object statusObject = request.getAttribute(RequestDispatcher.ERROR_STATUS_CODE, RequestAttributes.SCOPE_REQUEST);
@@ -34,6 +34,7 @@ public class ApplicationErrorController implements ErrorController {
         exceptionBuilder.error(errorMessage.toString());
         HttpStatus status = statusObject != null ? HttpStatus.resolve(Integer.parseInt(statusObject.toString())) : HttpStatus.INTERNAL_SERVER_ERROR;
         ExceptionDto exception = exceptionBuilder.build();
+        assert status != null;
         return new ResponseEntity<>(exception, status);
     }
 

src/main/java/com/barrelsofdata/springexamples/exception/JsonConversionException.java

@@ -5,13 +5,7 @@ import org.springframework.web.bind.annotation.ResponseStatus;
 
 @ResponseStatus(code = HttpStatus.BAD_REQUEST)
 public class JsonConversionException extends RuntimeException {
-    public JsonConversionException() {
-        super();
-    }
     public JsonConversionException(String message) {
         super(message);
     }
-    public JsonConversionException(String message, Exception e) {
-        super(message, e);
-    }
 }

src/main/java/com/barrelsofdata/springexamples/producer/Kafka.java

@@ -1,7 +1,5 @@
 package com.barrelsofdata.springexamples.producer;
 
-import org.springframework.kafka.KafkaException;
-
 public interface Kafka {
-    void publish(String eventRequest) throws KafkaException;
+    void publish(String eventRequest);
 }

src/main/java/com/barrelsofdata/springexamples/producer/KafkaImpl.java

@@ -4,7 +4,6 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.SendResult;
 import org.springframework.stereotype.Component;
@@ -13,14 +12,14 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
 
 @Component
 public class KafkaImpl implements Kafka {
-    private static final Logger logger = LoggerFactory.getLogger(Kafka.class);
+    private static final Logger logger = LoggerFactory.getLogger(KafkaImpl.class);
 
     @Autowired private KafkaTemplate<String, String> kafkaTemplate;
 
     @Value("${spring.kafka.producer.topic}")
     private String topic;
 
-    public void publish(String payload) throws KafkaException {
+    public void publish(String payload) {
         ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, payload); // Blocks call if kafka broker isn't available/responding
         future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
             @Override

src/main/java/com/barrelsofdata/springexamples/service/TelemetryServiceImpl.java

@@ -13,7 +13,7 @@ import org.springframework.stereotype.Service;
 
 @Service
 public class TelemetryServiceImpl implements TelemetryService {
-    private static final Logger logger = LoggerFactory.getLogger(TelemetryService.class);
+    private static final Logger logger = LoggerFactory.getLogger(TelemetryServiceImpl.class);
 
     @Autowired private Kafka producer;
 
@@ -29,7 +29,7 @@ public class TelemetryServiceImpl implements TelemetryService {
             throw new JsonConversionException("Failed json conversion");
         } catch (KafkaException e) {
             logger.error("Kafka exception for request {}", eventRequest);
-            // TODO: Handle what you want to do with the data here
+            // Handle what you want to do with the data here
         }
     }
 }

src/test/java/com/barrelsofdata/springexamples/ApplicationIntegrationTest.java

@@ -4,6 +4,7 @@ import com.barrelsofdata.springexamples.service.TelemetryService;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.assertj.core.api.Assertions;
+import org.awaitility.Durations;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -33,6 +34,7 @@ import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
+import static org.awaitility.Awaitility.await;
 import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.put;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
@@ -41,9 +43,9 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.
 @ExtendWith(SpringExtension.class)
 @EmbeddedKafka
 @AutoConfigureMockMvc
-public class ApplicationIntegrationTest {
+class ApplicationIntegrationTest {
     private int NUMBER_OF_BROKERS = 2;
-    private boolean CONTROLLER_SHUTDOWN = false;
+    private boolean CONTROLLER_SHUTDOWN = true;
     private int NUMBER_OF_PARTITIONS = 2;
     @Value("${spring.kafka.producer.topic}")
     private String TOPIC;
@@ -80,7 +82,7 @@ public class ApplicationIntegrationTest {
             "{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"LEFT_MOUSE_BUTTON_CLICK\",\"pl\":{\"x\":1000,\"y\":5000,\"w\":213,\"h\":124}};{\"timestamp\":\"2020-11-25T09:53:14.000+00:00\",\"id\":123,\"type\":\"LEFT_MOUSE_BUTTON_CLICK\",\"payload\":{\"width\":213,\"height\":124,\"x\":1000.0,\"y\":5000.0}}",
             "{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"RIGHT_MOUSE_BUTTON_CLICK\"};{\"timestamp\":\"2020-11-25T09:53:14.000+00:00\",\"id\":123,\"type\":\"RIGHT_MOUSE_BUTTON_CLICK\",\"payload\":null}"}
             , delimiter = ';')
-    public void success(String inputJson, String kafkaJson) throws Exception {
+    void success(String inputJson, String kafkaJson) throws Exception {
         HttpHeaders headers = new HttpHeaders();
 
         mockMvc.perform(
@@ -98,7 +100,7 @@ public class ApplicationIntegrationTest {
                         content().string(HttpStatus.CREATED.getReasonPhrase())
                 );
 
-        Thread.sleep(1000);
+        await().pollDelay(Durations.ONE_SECOND).until(() -> true);
         ConsumerRecord<String, String> singleRecord = records.poll(100, TimeUnit.MILLISECONDS);
         Assertions.assertThat(singleRecord).isNotNull();
         Assertions.assertThat(singleRecord.key()).isNull();
@@ -110,7 +112,7 @@ public class ApplicationIntegrationTest {
             "{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"LEFT_MOUSE_BUTTON_CLICK\",\"pl\":{\"x\":1000,\"y\":5000,\"w\":213,\"h\":124}};{\"timestamp\":\"2020-11-25T09:53:14.000+00:00\",\"id\":123,\"type\":\"LEFT_MOUSE_BUTTON_CLICK\",\"payload\":{\"width\":213,\"height\":124,\"x\":1000.0,\"y\":5000.0}}",
             "{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"RIGHT_MOUSE_BUTTON_CLICK\"};{\"timestamp\":\"2020-11-25T09:53:14.000+00:00\",\"id\":123,\"type\":\"RIGHT_MOUSE_BUTTON_CLICK\",\"payload\":null}"}
             , delimiter = ';')
-    public void unsupportedMedia(String inputJson, String kafkaJson) throws Exception {
+    void unsupportedMedia(String inputJson, String kafkaJson) throws Exception {
         HttpHeaders headers = new HttpHeaders();
 
         mockMvc.perform(
@@ -128,7 +130,7 @@ public class ApplicationIntegrationTest {
                         content().string(HttpStatus.CREATED.getReasonPhrase())
                 );
 
-        Thread.sleep(1000);
+        await().pollDelay(Durations.ONE_SECOND).until(() -> true);
         ConsumerRecord<String, String> singleRecord = records.poll(100, TimeUnit.MILLISECONDS);
         Assertions.assertThat(singleRecord).isNotNull();
         Assertions.assertThat(singleRecord.key()).isNull();

src/test/java/com/barrelsofdata/springexamples/controller/ApplicationErrorControllerTest.java

@@ -0,0 +1,101 @@
+package com.barrelsofdata.springexamples.controller;
+
+import com.barrelsofdata.springexamples.dto.EventRequestDto;
+import com.barrelsofdata.springexamples.exception.JsonConversionException;
+import com.barrelsofdata.springexamples.producer.Kafka;
+import com.barrelsofdata.springexamples.service.TelemetryService;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.Mockito;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+import org.springframework.test.web.servlet.MockMvc;
+
+import javax.servlet.RequestDispatcher;
+
+import java.util.Objects;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.put;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.*;
+
+@SpringBootTest
+@ExtendWith(SpringExtension.class)
+@AutoConfigureMockMvc
+class ApplicationErrorControllerTest {
+    @MockBean private TelemetryService telemetryService;
+    @MockBean private Kafka kafka;
+    @Autowired private MockMvc mockMvc;
+
+    @ParameterizedTest(name = "Error API request")
+    @ValueSource(strings = {"{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"LEFT_MOUSE_BUTTON_CLICK\",\"pl\":{\"x\":1000,\"y\":5000,\"w\":213,\"h\":124}}","{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"RIGHT_MOUSE_BUTTON_CLICK\"}"})
+    void testError(String json) throws Exception {
+        HttpHeaders headers = new HttpHeaders();
+        Mockito.doThrow(JsonConversionException.class).when(telemetryService).receiveTelemetry(any(EventRequestDto.class));
+
+        mockMvc.perform(
+                put("/telemetry")
+                        .contentType(MediaType.APPLICATION_JSON)
+                        .content(json)
+                        .headers(headers))
+                .andDo(result -> {
+                    if (result.getResolvedException() != null) {
+                        byte[] response = mockMvc.perform(
+                                get("/error")
+                                        .requestAttr(RequestDispatcher.ERROR_STATUS_CODE, result.getResponse().getStatus())
+                                        .requestAttr(RequestDispatcher.ERROR_REQUEST_URI, Objects.requireNonNull(result.getRequest().getRequestURI()))
+                                        .requestAttr(RequestDispatcher.ERROR_EXCEPTION, result.getResolvedException())
+                                        .requestAttr(RequestDispatcher.ERROR_MESSAGE, String.valueOf(result.getResponse().getErrorMessage())))
+                                .andReturn()
+                                .getResponse()
+                                .getContentAsByteArray();
+                        result.getResponse()
+                                .getOutputStream()
+                                .write(response);
+                    }
+                })
+                .andExpect(
+                        status().isBadRequest()
+                )
+                .andExpect(
+                        jsonPath("$.error").value("Bad Request")
+                )
+                .andExpect(
+                        jsonPath("$.timestamp").exists()
+                );
+    }
+
+    @Test
+    @DisplayName("Test default error status code")
+    void testErrorWithoutMessage() throws Exception {
+        mockMvc.perform(
+                get("/error"))
+                .andExpect(
+                        status().isInternalServerError()
+                );
+    }
+
+    @ParameterizedTest(name = "Error API request")
+    @ValueSource(strings = {"{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"LEFT_MOUSE_BUTTON_CLICK\",\"pl\":{\"x\":1000,\"y\":5000,\"w\":213,\"h\":124}}","{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"RIGHT_MOUSE_BUTTON_CLICK\""})
+    void testNotFound(String json) throws Exception {
+        HttpHeaders headers = new HttpHeaders();
+
+        mockMvc.perform(
+                put("/nonexisting")
+                        .contentType(MediaType.APPLICATION_JSON)
+                        .content(json)
+                        .headers(headers))
+                .andExpect(
+                        status().isNotFound()
+                );
+    }
+}

src/test/java/com/barrelsofdata/springexamples/controller/TelemetryControllerTest.java

@@ -27,14 +27,14 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.
 @SpringBootTest
 @ExtendWith(SpringExtension.class)
 @AutoConfigureMockMvc
-public class TelemetryControllerTest {
+class TelemetryControllerTest {
     @MockBean private TelemetryService telemetryService;
     @MockBean private Kafka kafka;
     @Autowired private MockMvc mockMvc;
 
     @ParameterizedTest(name = "Success API request")
     @ValueSource(strings = {"{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"LEFT_MOUSE_BUTTON_CLICK\",\"pl\":{\"x\":1000,\"y\":5000,\"w\":213,\"h\":124}}","{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"RIGHT_MOUSE_BUTTON_CLICK\"}"})
-    public void success(String json) throws Exception {
+    void success(String json) throws Exception {
         HttpHeaders headers = new HttpHeaders();
         Mockito.doNothing().when(telemetryService).receiveTelemetry(any(EventRequestDto.class));
 
@@ -56,7 +56,7 @@ public class TelemetryControllerTest {
 
     @ParameterizedTest(name = "Json conversion fail API response")
     @ValueSource(strings = {"{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"LEFT_MOUSE_BUTTON_CLICK\",\"pl\":{\"x\":1000,\"y\":5000,\"w\":213,\"h\":124}}","{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"RIGHT_MOUSE_BUTTON_CLICK\"}"})
-    public void failBadJson(String json) throws Exception {
+    void failBadJson(String json) throws Exception {
         String expectedErrorMessage = "Failed json conversion";
         HttpHeaders headers = new HttpHeaders();
         Mockito.doThrow(new JsonConversionException(expectedErrorMessage)).when(telemetryService).receiveTelemetry(any(EventRequestDto.class));
@@ -73,7 +73,7 @@ public class TelemetryControllerTest {
 
     @ParameterizedTest(name = "Unsupported media type")
     @ValueSource(strings = {"{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"LEFT_MOUSE_BUTTON_CLICK\",\"pl\":{\"x\":1000,\"y\":5000,\"w\":213,\"h\":124}}}","{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"RIGHT_MOUSE_BUTTON_CLICK\"}}"})
-    public void unsupportedMediaType(String json) throws Exception {
+    void unsupportedMediaType(String json) throws Exception {
         HttpHeaders headers = new HttpHeaders();
         Mockito.doNothing().when(telemetryService).receiveTelemetry(any(EventRequestDto.class));
 
@@ -89,7 +89,7 @@ public class TelemetryControllerTest {
 
     @ParameterizedTest(name = "Missing required field or wrong value for type, bad request")
     @ValueSource(strings = {"{\"ts\":\"1606297994000\",\"ty\":\"LEFT_MOUSE_BUTTON_CLICK\",\"pl\":{\"x\":1000,\"y\":5000,\"w\":213,\"h\":124}}","{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"MOUSE_BUTTON_CLICK\"}"})
-    public void missingRequiredField(String json) throws Exception {
+    void missingRequiredField(String json) throws Exception {
         HttpHeaders headers = new HttpHeaders();
         Mockito.doNothing().when(telemetryService).receiveTelemetry(any(EventRequestDto.class));
 
@@ -105,7 +105,7 @@ public class TelemetryControllerTest {
 
     @ParameterizedTest(name = "Method not allowed for non-PUT requests")
     @ValueSource(strings = {"{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"LEFT_MOUSE_BUTTON_CLICK\",\"pl\":{\"x\":1000,\"y\":5000,\"w\":213,\"h\":124}}","{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"RIGHT_MOUSE_BUTTON_CLICK\"}"})
-    public void methodNotAllowed(String json) throws Exception {
+    void methodNotAllowed(String json) throws Exception {
         HttpHeaders headers = new HttpHeaders();
         Mockito.doNothing().when(telemetryService).receiveTelemetry(any(EventRequestDto.class));
 

src/test/java/com/barrelsofdata/springexamples/producer/KafkaTest.java

@@ -23,7 +23,7 @@ import static org.mockito.ArgumentMatchers.any;
 
 @SpringBootTest
 @ExtendWith(SpringExtension.class)
-public class KafkaTest {
+class KafkaTest {
     @MockBean private KafkaTemplate<String, String> kafkaTemplate;
 
     @Autowired @InjectMocks private KafkaImpl producer;
@@ -33,7 +33,7 @@ public class KafkaTest {
 
     @ParameterizedTest(name = "Check successful send")
     @ValueSource(strings = {"{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"LEFT_MOUSE_BUTTON_CLICK\",\"pl\":{\"x\":1000,\"y\":5000,\"w\":213,\"h\":124}}","{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"RIGHT_MOUSE_BUTTON_CLICK\"}"})
-    public void successSend(String json) throws JsonProcessingException {
+    void successSend(String json) throws JsonProcessingException {
         EventRequestDto eventRequestDto = mapper.readValue(json, EventRequestDto.class);
         String kafkaPayload = mapper.writeValueAsString(eventRequestDto);
         Mockito.doReturn(new SettableListenableFuture<>()).when(kafkaTemplate).send(any(String.class), any(String.class));
@@ -42,7 +42,7 @@ public class KafkaTest {
 
     @ParameterizedTest(name = "Check failed send")
     @ValueSource(strings = {"{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"LEFT_MOUSE_BUTTON_CLICK\",\"pl\":{\"x\":1000,\"y\":5000,\"w\":213,\"h\":124}}","{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"RIGHT_MOUSE_BUTTON_CLICK\"}"})
-    public void failedSend(String json) throws JsonProcessingException, InterruptedException {
+    void failedSend(String json) throws JsonProcessingException, InterruptedException {
         EventRequestDto eventRequestDto = mapper.readValue(json, EventRequestDto.class);
         String kafkaPayload = mapper.writeValueAsString(eventRequestDto);
         Mockito.doThrow(KafkaException.class).when(kafkaTemplate).send(any(String.class), any(String.class));

src/test/java/com/barrelsofdata/springexamples/service/TelemetryServiceTest.java

@@ -5,6 +5,7 @@ import com.barrelsofdata.springexamples.exception.JsonConversionException;
 import com.barrelsofdata.springexamples.producer.Kafka;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import kafka.common.KafkaException;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -20,25 +21,33 @@ import static org.mockito.ArgumentMatchers.any;
 
 @SpringBootTest
 @ExtendWith(SpringExtension.class)
-public class TelemetryServiceTest {
-    @MockBean private Kafka kafka;
+class TelemetryServiceTest {
+    @MockBean private Kafka producer;
     @MockBean private ObjectMapper mapper;
     @Autowired @InjectMocks private TelemetryServiceImpl telemetryService;
 
     @ParameterizedTest(name = "Successful publish")
     @ValueSource(strings = {"{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"LEFT_MOUSE_BUTTON_CLICK\",\"pl\":{\"x\":1000,\"y\":5000,\"w\":213,\"h\":124}}","{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"RIGHT_MOUSE_BUTTON_CLICK\"}"})
-    public void successPublish(String json) throws JsonProcessingException {
+    void successPublish(String json) throws JsonProcessingException {
         EventRequestDto eventRequestDto = new ObjectMapper().readValue(json, EventRequestDto.class);
-        telemetryService.receiveTelemetry(eventRequestDto);
+        Assertions.assertDoesNotThrow(() ->telemetryService.receiveTelemetry(eventRequestDto));
     }
 
     @ParameterizedTest(name = "Json conversion failure")
     @ValueSource(strings = {"{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"LEFT_MOUSE_BUTTON_CLICK\",\"pl\":{\"x\":1000,\"y\":5000,\"w\":213,\"h\":124}}"})
-    public void failPublish(String json) throws JsonProcessingException {
+    void failPublish(String json) throws JsonProcessingException {
         EventRequestDto eventRequestDto = new ObjectMapper().readValue(json, EventRequestDto.class);
         Mockito.doThrow(JsonProcessingException.class).when(mapper).writeValueAsString(any(EventRequestDto.class));
         JsonConversionException exception = Assertions.assertThrows(JsonConversionException.class, () -> telemetryService.receiveTelemetry(eventRequestDto));
         Assertions.assertEquals("Failed json conversion", exception.getMessage());
     }
 
+    @ParameterizedTest(name = "Kafka timeout failure")
+    @ValueSource(strings = {"{\"ts\":\"1606297994000\",\"id\":\"123\",\"ty\":\"LEFT_MOUSE_BUTTON_CLICK\",\"pl\":{\"x\":1000,\"y\":5000,\"w\":213,\"h\":124}}"})
+    void timeoutPublish(String json) throws JsonProcessingException {
+        EventRequestDto eventRequestDto = new ObjectMapper().readValue(json, EventRequestDto.class);
+        Mockito.doThrow(KafkaException.class).when(producer).publish(any(String.class));
+        Assertions.assertDoesNotThrow(() -> telemetryService.receiveTelemetry(eventRequestDto));
+    }
+
 }