-FROM openjdk:11-jre-slim
+FROM eclipse-temurin:17-jre
COPY target/*.jar /opt/app.jar
EXPOSE 8081
ENTRYPOINT ["java", "-jar", "/opt/app.jar"]
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
- <version>3.0.2</version>
+ <version>3.2.5</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>de.juplo.kafka.wordcount</groupId>
<artifactId>recorder</artifactId>
- <version>1.0.2</version>
+ <version>1.2.0</version>
<name>Wordcount-Recorder</name>
<description>Recorder-service of the multi-user wordcount-example</description>
<properties>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
- <groupId>org.apache.kafka</groupId>
- <artifactId>kafka-clients</artifactId>
+ <groupId>org.springframework.kafka</groupId>
+ <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.hibernate.validator</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.springframework.kafka</groupId>
+ <artifactId>spring-kafka-test</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
<plugins>
+ <plugin>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ </plugin>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
+import org.springframework.kafka.support.serializer.JsonSerializer;
import org.springframework.util.Assert;
import java.util.Properties;
public class RecorderApplication
{
@Bean(destroyMethod = "close")
- KafkaProducer<String, String> producer(RecorderApplicationProperties properties)
+ KafkaProducer<User, Recording> producer(RecorderApplicationProperties properties)
{
Assert.hasText(properties.getBootstrapServer(), "juplo.wordcount.recorder.bootstrap-server must be set");
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServer());
- props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
- props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
+ props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
+ props.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false);
return new KafkaProducer<>(props);
}
public class RecorderController
{
private final String topic;
- private final KafkaProducer<String, String> producer;
+ private final KafkaProducer<User, Recording> producer;
- public RecorderController(RecorderApplicationProperties properties, KafkaProducer<String,String> producer)
+ public RecorderController(
+ RecorderApplicationProperties properties,
+ KafkaProducer<User, Recording> producer)
{
this.topic = properties.getTopic();
this.producer = producer;
{
DeferredResult<ResponseEntity<RecordingResult>> result = new DeferredResult<>();
- ProducerRecord<String, String> record = new ProducerRecord<>(topic, username, sentence);
+ ProducerRecord<User, Recording> record = new ProducerRecord<>(
+ topic,
+ User.of(username),
+ Recording.of(username, sentence));
+
producer.send(record, (metadata, exception) ->
{
- if (metadata != null)
+ if (exception == null)
{
result.setResult(
ResponseEntity.ok(RecordingResult.of(
--- /dev/null
+package de.juplo.kafka.wordcount.recorder;
+
+import lombok.Value;
+
+
+@Value(staticConstructor = "of")
+public class Recording
+{
+ String user;
+ String sentence;
+}
--- /dev/null
+package de.juplo.kafka.wordcount.recorder;
+
+import lombok.Value;
+
+
+@Value(staticConstructor = "of")
+public class User
+{
+ String user;
+}
+++ /dev/null
-package de.juplo.kafka.wordcount.recorder;
-
-import org.junit.jupiter.api.Test;
-import org.springframework.boot.test.context.SpringBootTest;
-
-@SpringBootTest
-class ApplicationTests
-{
- @Test
- void contextLoads()
- {
- }
-}
--- /dev/null
+package de.juplo.kafka.wordcount.recorder;
+
+import de.juplo.kafka.wordcount.splitter.TestRecording;
+import de.juplo.kafka.wordcount.splitter.TestUser;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+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.context.TestConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.http.MediaType;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.messaging.handler.annotation.Payload;
+import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.MvcResult;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+
+import java.time.Duration;
+import java.util.List;
+
+import static de.juplo.kafka.wordcount.recorder.RecorderApplicationIT.TOPIC_OUT;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.asyncDispatch;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.result.MockMvcResultHandlers.print;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+
+@SpringBootTest(
+ properties = {
+ "spring.kafka.consumer.auto-offset-reset=earliest",
+ "spring.kafka.consumer.key-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer",
+ "spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer",
+ "spring.kafka.consumer.properties.spring.json.key.default.type=de.juplo.kafka.wordcount.splitter.TestUser",
+ "spring.kafka.consumer.properties.spring.json.value.default.type=de.juplo.kafka.wordcount.splitter.TestRecording",
+ "logging.level.root=WARN",
+ "logging.level.de.juplo=DEBUG",
+ "juplo.wordcount.recorder.bootstrap-server=${spring.embedded.kafka.brokers}",
+ "juplo.wordcount.recorder.topic=" + TOPIC_OUT })
+@AutoConfigureMockMvc
+@EmbeddedKafka(topics = { TOPIC_OUT })
+@Slf4j
+class RecorderApplicationIT
+{
+ static final String TOPIC_OUT = "out";
+
+ @Autowired
+ MockMvc mockMvc;
+ @Autowired
+ Consumer consumer;
+
+
+ @Test
+ @DisplayName("Posted messages are accepted and sent to Kafka")
+ void userMessagesAreAcceptedAndSentToKafka()
+ {
+ MultiValueMap<TestUser, TestRecording> recordings = new LinkedMultiValueMap<>();
+
+ TestData
+ .getInputMessages()
+ .forEach(kv ->
+ {
+ sendRedording(kv.key, kv.value);
+ recordings.add(TestUser.of(kv.key), TestRecording.of(kv.key, kv.value));
+ });
+
+
+ await("Received expected messages")
+ .atMost(Duration.ofSeconds(5))
+ .untilAsserted(() -> recordings.forEach((user, userRecordings) ->
+ assertThat(consumer.receivedFor(user)).containsExactlyElementsOf(userRecordings)));
+ }
+
+ void sendRedording(String user, String sentence)
+ {
+ try
+ {
+ MvcResult result = mockMvc
+ .perform(post("/{user}", user)
+ .contentType(MediaType.TEXT_PLAIN)
+ .content(sentence))
+ .andReturn();
+
+ mockMvc.perform(asyncDispatch(result))
+ .andDo(print())
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.username").value(user))
+ .andExpect(jsonPath("$.sentence").value(sentence));
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
+
+ static class Consumer
+ {
+ private final MultiValueMap<TestUser, TestRecording> received = new LinkedMultiValueMap<>();
+
+ @KafkaListener(groupId = "TEST", topics = TOPIC_OUT)
+ public synchronized void receive(
+ @Header(KafkaHeaders.RECEIVED_KEY) TestUser user,
+ @Payload TestRecording recording)
+ {
+ log.debug("Received message: {}={}", user, recording);
+ received.add(user, recording);
+ }
+
+ synchronized List<TestRecording> receivedFor(TestUser user)
+ {
+ List<TestRecording> recordings = received.get(user);
+ return recordings == null
+ ? List.of()
+ : received.get(user).stream().toList();
+ }
+ }
+
+ @TestConfiguration
+ static class Configuration
+ {
+ @Bean
+ Consumer consumer()
+ {
+ return new Consumer();
+ }
+ }
+}
--- /dev/null
+package de.juplo.kafka.wordcount.recorder;
+
+import org.apache.kafka.streams.KeyValue;
+
+import java.util.stream.Stream;
+
+
+class TestData
+{
+ static final String PETER = "päter";
+ static final String KLAUS = "klühs";
+
+ static final Stream<KeyValue<String, String>> getInputMessages()
+ {
+ return Stream.of(INPUT_MESSAGES);
+ }
+
+ private static final KeyValue<String, String>[] INPUT_MESSAGES = new KeyValue[]
+ {
+ new KeyValue<>("päter", "Hall° Wält?¢*&%€!"),
+ new KeyValue<>("päter", "Hallo Welt!"),
+ new KeyValue<>("klühs", "Müsch gäb's auch!"),
+ new KeyValue<>("päter", "Boäh, echt! ß mal nä Nümmäh!"),
+ };
+}
--- /dev/null
+package de.juplo.kafka.wordcount.splitter;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor(staticName = "of")
+public class TestRecording
+{
+ String user;
+ String sentence;
+}
--- /dev/null
+package de.juplo.kafka.wordcount.splitter;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor(staticName = "of")
+public class TestUser
+{
+ String user;
+}