Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat : upgrade to spring boot 3.4 and polish tc config #606

Merged
merged 5 commits into from
Dec 7, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@
*/
package com.example.springbootkafkaavro.model;

import org.apache.avro.generic.GenericArray;
import org.apache.avro.specific.SpecificData;
import org.apache.avro.util.Utf8;
import org.apache.avro.message.BinaryMessageEncoder;
Expand Down
16 changes: 3 additions & 13 deletions kafka-avro/spring-boot-kafka-avro-consumer/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -66,11 +66,6 @@
<artifactId>h2</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>

<dependency>
<groupId>org.apache.avro</groupId>
Expand All @@ -89,8 +84,8 @@
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-testcontainers</artifactId>
<scope>test</scope>
</dependency>
<dependency>
Expand All @@ -103,11 +98,6 @@
<artifactId>kafka</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
<scope>test</scope>
</dependency>
</dependencies>

<build>
Expand Down Expand Up @@ -158,7 +148,7 @@
<configuration>
<java>
<googleJavaFormat>
<version>1.19.2</version>
<version>1.25.0</version>
<style>AOSP</style>
</googleJavaFormat>
</java>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,17 +6,9 @@
import jakarta.persistence.GenerationType;
import jakarta.persistence.Id;
import jakarta.persistence.Table;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;

@Entity
@Table(name = "person_entity")
@Setter
@Getter
@AllArgsConstructor
@NoArgsConstructor
public class PersonEntity {

@Id
Expand All @@ -27,4 +19,33 @@ public class PersonEntity {
private String name;

private Integer age;

public PersonEntity() {}

public Integer getAge() {
return age;
}

public PersonEntity setAge(Integer age) {
this.age = age;
return this;
}

public Long getId() {
return id;
}

public PersonEntity setId(Long id) {
this.id = id;
return this;
}

public String getName() {
return name;
}

public PersonEntity setName(String name) {
this.name = name;
return this;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -4,25 +4,28 @@
import com.example.springbootkafkaavro.model.Person;
import com.example.springbootkafkaavro.repository.PersonRepository;
import com.example.springbootkafkaavro.util.ApplicationConstants;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@RequiredArgsConstructor
public class AvroKafkaListener {

private static final Logger log = LoggerFactory.getLogger(AvroKafkaListener.class);
private final PersonRepository personRepository;

public AvroKafkaListener(PersonRepository personRepository) {
this.personRepository = personRepository;
}

@KafkaListener(topics = ApplicationConstants.PERSONS_TOPIC, groupId = "group_id")
public void handler(ConsumerRecord<String, Person> personConsumerRecord) {
Person person = personConsumerRecord.value();
log.info("Person received : {} : {} ", person.getName(), person.getAge());
PersonEntity personEntity =
new PersonEntity(null, person.getName().toString(), person.getAge());
new PersonEntity().setName(person.getName().toString()).setAge(person.getAge());
this.personRepository.save(personEntity);
}
}
Original file line number Diff line number Diff line change
@@ -1,9 +1,5 @@
package com.example.springbootkafkaavro.util;

import lombok.experimental.UtilityClass;

@UtilityClass
public class ApplicationConstants {

public static final String PERSONS_TOPIC = "persons";
public interface ApplicationConstants {
String PERSONS_TOPIC = "persons";
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
package com.example.springbootkafkaavro;

import static java.util.concurrent.TimeUnit.SECONDS;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;

import com.example.springbootkafkaavro.common.KafkaContainersConfig;
import com.example.springbootkafkaavro.model.Person;
import com.example.springbootkafkaavro.repository.PersonRepository;
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.context.annotation.Import;
import org.springframework.test.web.servlet.MockMvc;

@SpringBootTest(
properties = {
"spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer",
"spring.kafka.producer.value-serializer=io.confluent.kafka.serializers.KafkaAvroSerializer"
},
classes = {KafkaContainersConfig.class})
@AutoConfigureMockMvc
@Import(KafkaProducer.class)
class ApplicationIntTests {

@Autowired MockMvc mockMvc;
@Autowired PersonRepository personRepository;
@Autowired KafkaProducer kafkaProducer;

@Test
void contextLoads() {
Person person = new Person();
person.setAge(33);
person.setName("junit");
this.kafkaProducer.sendMessage(person);
await().atMost(10, SECONDS)
.untilAsserted(() -> assertThat(personRepository.count()).isEqualTo(1));
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -2,16 +2,18 @@

import com.example.springbootkafkaavro.model.Person;
import com.example.springbootkafkaavro.util.ApplicationConstants;
import lombok.RequiredArgsConstructor;
import org.springframework.boot.test.context.TestComponent;
import org.springframework.kafka.core.KafkaTemplate;

@TestComponent
@RequiredArgsConstructor
public class KafkaProducer {

private final KafkaTemplate<String, Person> kafkaTemplate;

public KafkaProducer(KafkaTemplate<String, Person> kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
}

public void sendMessage(Person person) {
this.kafkaTemplate.send(
ApplicationConstants.PERSONS_TOPIC, person.getName().toString(), person);
Expand Down

This file was deleted.

Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
package com.example.springbootkafkaavro;

import com.example.springbootkafkaavro.common.KafkaContainersConfig;
import org.springframework.boot.SpringApplication;

class TestSpringBootKafkaAvroConsumerApplication {

public static void main(String[] args) {
SpringApplication.from(SpringBootKafkaAvroConsumerApplication::main)
.with(KafkaContainersConfig.class)
.run(args);
}
}
Loading
Loading