SpringData Flyway Kullanımı

Giriş

flyway'i komut satırından kullanabilmek için kurmak gerekir. Kurma işlemi aslında sadece zip dosyasını indirip açmak ve Path'e eklemekten ibaret. Şöyle yaparız

export PATH=$PATH:$HOME/flyway-9.3.0

Komut satırından 7 seçenek kullanılabilir.

migrate

clean

info

validate

undo

baseline

repair

migrate seçeneği

Şöyle yaparız

flyway migrate -configFiles=flyway.properties

undo seçeneği

Şöyle yaparız

flyway undo -configFiles=flyway.properties

Docker Olarak Kullanım

Örnek

compose.yaml dosyasında şöyle yaparız. Projeye Docker Compose Desteği ekleyince flyway otomatik çalışır

# FILE: compose.yaml

version: '3'
services:
  postgres:
    image: 'postgres:15'
    container_name: "postgres"
    environment:
      - 'POSTGRES_DB=postgres'
      - 'POSTGRES_PASSWORD=postgres'
      - 'POSTGRES_USER=postgres'
    ports:
      - '5432:5432'
  flyway:
    # Use Docker image containing Flyway CLI
    image: flyway/flyway:9.22.1
    container_name: "flyway-migration"
    # Execute migration command with input parameters on container startup
    command: -locations=filesystem:/flyway/migration -user=postgres -password=postgres -url="jdbc:postgresql://postgres:5432/postgres" -connectRetries=5 migrate
    # Copy migration script folder into container
    volumes:
      - ./db/migration:/flyway/migration
    # Wait for Postgres container to start
    depends_on:
      - postgres

Kütüphane Olarak Kullanım

1. Flyway maven veya gradle dependency eklenir

2. Ayrıca "spring.jpa.hibernate.ddl-auto : none" olmalıdır

3. application.properties dosyasında data source tanımlanır. Flyway için herhangi bir ayar belirtmemize gerek yok. Varsayılan ayarlar yeterli.

Örnek

Şöyle yaparız

server:
  port: ${port:8080}

spring:
  application:
    name: flyway-demo

  datasource:
    driver-class-name: org.h2.Driver
    username: sa
    password:
    url: "jdbc:h2:mem:db;DB_CLOSE_DELAY=-1;DB_CLOSE_ON_EXIT=FALSE;DATABASE_TO_LOWER=TRUE;CASE_INSENSITIVE_IDENTIFIERS=TRUE"
  h2:
    console:
      enabled: true
      path: /h2-console
  jpa:
    show-sql: true
    hibernate:
      ddl-auto: none

Maven

Şöyle yaparız. Böylece uygulama çalışırken schema migration (şema taşınması/değiştirilmesi) yapılabilir.

<dependency>
  <groupId>org.flywaydb</groupId>
  <artifactId>flyway-core</artifactId>
  <version>6.5.5</version>
</dependency>

Açıklaması şöyle

With this, Flyway automatically scans the classpath looking for the migration scripts, and executes them against the database during application startup. By default, it scans for files located in classpath:db/migration.

flyway_schema_history Tablosu

Açıklaması şöyle

Flyway creates a special table called the “Flyway Schema History” to store metadata, such as script name, migration version, execution status, execution date-time, etc.

Then, to keep track of the schema evolution, it logs the metadata into this table, leaving the Flyway Schema History table as the only source of truth, and allowing Flyway to check if a new migration script has been added.

In addition, it validates the integrity of older scripts using a checksum stored in the table. So, if any of the scripts in the table has been modified, the application will fail to start.

Şeklen şöyle. Bu tabloyu flyway ilk kez çalışırken yaratır. Değişiklikleri bu tabloda takip eder. Böylece bir script ikinci kez çalıştırılmaz

Tablonun satırları şöyle

| installed_rank | version | description  | type | script                 | checksum   | installed_by | installed_on   | execution_time | success |
|----------------|---------|--------------|------|------------------------|------------|--------------|----------------|----------------|---------|
| 1              | 1       | post tag     | SQL  | V1_0__post_tag.sql     | -611721954 | postgres     | 30-06-20 15:21 | 61             | TRUE    |
| 2              | 1.1     | post details | SQL  | V1_1__post_details.sql | 511495203  | postgres     | 30-06-20 15:21 | 13             | TRUE    |
| 3              | 1.2     | post comment | SQL  | V1_2__post_comment.sql | 762350400  | postgres     | 30-06-20 15:21 | 14             | TRUE    |
| 4              | 1.3     | users        | SQL  | V1_3__users.sql        | -596399497 | postgres     | 30-06-20 15:55 | 32             | TRUE    |

Checksum Hatası

Hata şöylee

org.flywaydb.core.api.FlywayException: Validate failed: Migration checksum mismatch for migration version .

Açıklaması şöyle

What is it?

The checksum validation from Flyway is basically a check between the checksum of the current migration file in your app against the checksum from the same migration it already run in the past. You can check this list on your database, under flyway_schema_history table created and used by Flyway.

What it means?

It means that the script you app has when it starts is not the same Flyway already applied in the past and since it can't figure out if that is correct or not, it fails. Ideally, you should never change a script you already applied, you should always evolve and create new ones, that's the whole idea about migrations.

How to avoid it?

As said before, you should never change scripts that were already executed before. You should always create new ones. Of course if that happens on a dev environment and you figure out changes are needed.

application.properties

application.properties yazısına taşıdım

Script Dizini

Veri tabanı scriptleri için varsayılan dizin şöyledir

src/resources/db/migration

Çift alt çizgi kullanılması gerekir. Açıklaması şöyle

Make sure you have double underscores present after the version of a file, i.e “V1__FILENAME.sql”, if you provide a single underscore after the version file is skipped when starting the application by Flyway, i.e “V1_FILENAME.sql”.

V1__create_book_table.sql

V1.1__insert_into_books.sql

Şeklen şöyle

V ile başlayanlar dışında U ve R ile de başlayan dosya isimleri olabilir. Şeklen şöyle

Açıklaması şöyle

Part 1: It is the letter “v” in uppercase. The name always starts with this letter.

Part 2: It is the migration version; it can be 1, 001, 1.2.3, 2021.09.24.12.55.32, … you got it.

Part 3: It is the two underscores (_)

Part 4: The description of the migration; you can separate words with an underscore or a space.

Part 5: It is the extension of the file .sql

Açıklaması şöyle

The Part 1, 3 and 5 are configurable using the configuration file with this properties: spring.flyway.sql-migration-prefix, spring.flyway.sql-migration-separator, spring.flyway.sql-migration-suffixes.

spring.flyway.sql-migration-prefix=T

spring.flyway.sql-migration-separator=--

spring.flyway.sql-migration-suffixes=.blog

The migration file will be: T1.0 — create_users_table.blog

Açıklaması şöyle.

sql statement should end with ‘;’ or cause execute fail.

Örnek - V1.0__initial_schema.sql

Şöyledir. Burada V1.0 ismi kullanılıyor

-- V1.0__initial_schema.sql
CREATE TABLE author
(
    ID         BIGINT       NOT NULL AUTO_INCREMENT,
    NAME       VARCHAR(100) NOT NULL UNIQUE,
    BIRTH_YEAR INT          NOT NULL,
    CONSTRAINT pk_author PRIMARY KEY (ID)
);
CREATE TABLE book
(
    ID       BIGINT       NOT NULL AUTO_INCREMENT,
    TITLE    VARCHAR(150) NULL,
    PUB_DATE datetime     NULL,
    AUTHOR   BIGINT       NULL,
    CONSTRAINT pk_book PRIMARY KEY (ID)
);
ALTER TABLE book
    ADD CONSTRAINT uc_book_title UNIQUE (TITLE);
ALTER TABLE book
    ADD CONSTRAINT FK_BOOK_ON_AUTHOR FOREIGN KEY (AUTHOR) REFERENCES author (ID);

Örnek - V1_create_person.sql

Şöyledir. Burada V1 ismi kullanılıyor

create table person
(
    id           serial primary key,
    first_name   text,
    last_name    text,
    date_created timestamp with time zone
);

Örnek

Eğer uygulamayı çalıştırmadan sadece Flyway işini yapsın istersek şöyle yaparız

import static org.springframework.boot.WebApplicationType.NONE;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.autoconfigure.flyway.FlywayAutoConfiguration;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.annotation.Import;

/**
 * Utility to run flyway migration without starting service
 */
@SpringBootConfiguration
@Import({DataSourceAutoConfiguration.class, FlywayAutoConfiguration.class})
public class FlywayMigrationRunner {

  public static void main(String[] args) {

    SpringApplication application =
        new SpringApplicationBuilder(FlywayMigrationRunner.class)
            .web(NONE).build();

    application.run(args);
  }
}

Örnek - Test

Eğer Flyway'i testlerde kullanmak istermezsek şöyle yaparız. application-test-containers.yml içinde flyway'i kapatırız

spring:
  datasource:
    url: jdbc:tc:postgresql:9.6.8:///test_database
    username: user
    password: password
  jpa:
    hibernate:
      ddl-auto: create
  flyway:
    enabled: false

application-test-containers-flyway.yml içinde yeni konfigürasyon yazarız

spring:
  datasource:
    url: jdbc:tc:postgresql:9.6.8:///test_database
    username: user
    password: password

Test içinde ActiveProfile ile kullanırız

@SpringBootTest(webEnvironment = WebEnvironment.NONE)
@Testcontainers
@ActiveProfiles("test-containers-flyway")
class PersonCreateServiceImplTestContainersFlyway {
  @Autowired
  private PersonRepository personRepository;
  @MockBean
  private PersonValidateService personValidateService;
  @Autowired
  private PersonCreateService personCreateService;

  @BeforeEach
  void init() {
    personRepository.deleteAll();
  }
  @Test
  void shouldCreateOnePerson() {
    final var people = personCreateService.createFamily(
        List.of("Simon"),"Kirekov");
    assertEquals(1, people.size());
    final var person = people.get(0);
    assertEquals("Simon", person.getFirstName());
    assertEquals("Kirekov", person.getLastName());
    assertTrue(person.getDateCreated().isBefore(ZonedDateTime.now()));
  }
  @Test
  void shouldRollbackIfAnyUserIsNotValidated() {
    doThrow(new ValidationFailedException(""))
        .when(personValidateService)
        .checkUserCreation("John", "Brown");
    assertThrows(ValidationFailedException.class, () -> personCreateService.createFamily(
        List.of("Matilda", "Vasya", "John"),
        "Brown"
    ));
    assertEquals(0, personRepository.count());
  }
}

SpringWebFlux Scheduler Sınıfı

Giriş

Şu satırı dahil ederiz

import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

Örnek

Şöyle yaparız

@Bean
public Scheduler jdbcScheduler() {
  return Schedulers.fromExecutor(Executors.newFixedThreadPool(1));
}

SpringData MongoDB ReactiveMongoOperations

Giriş

Şu satırı dahil ederiz

import org.springframework.data.mongodb.core.ReactiveMongoOperations;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;

findById metodu

Şöyle yaparız

@Service
class ReservationService {

  private final ReactiveMongoOperations reactiveMongoOperations ;

  @Autowired
  public ReservationServiceImpl(ReactiveMongoOperations reactiveMongoOperations ) {
    this.reactiveMongoOperations = reactiveMongoOperations ;
  }

  public Mono<Reservation> getReservation(String id) {
    return reactiveMongoOperations 
      .findById(id, Reservation.class);
  }

  public Mono<Reservation> createReservation(Mono<Reservation> reservation) {
    return reactiveMongoOperations 
      .save(reservation);
  }

  public Mono<Reservation> updateReservation(String id, Mono<Reservation> reservation) {
    return reservation
      .flatMap(res -> reactiveMongoOperations .
        findAndModify(Query.query(Criteria.where("id").is(id)),
                      Update.update("price", res.getPrice()), Reservation.class)
                      .flatMap(result -> {
                        result.setPrice(res.getPrice());
                        return Mono.just(result);
    }));
  }

  public Mono<Boolean> deleteReservation(String id) {
    return reactiveMongoOperations .remove(Query
      .query(Criteria.where("id").is(id)), Reservation.class)
      .flatMap(deleteResult -> Mono.just(deleteResult.wasAcknowledged()));
  }

  public Flux<Reservation> listAllReservations() {
    return reactiveMongoOperations .findAll(Reservation.class);
  }
}

Micrometer ile Metrik Yazmak

Giriş

Metric'leri loglamak için io.micrometer.core.instrument.logging.LoggingMeterRegistry sınıfı kullanılır. Açıklaması şöyle

To be able to send custom metrics we need to import MeterRegistry from the Micrometer library and inject it into our class. 

It is possible to instantiate these types of meters from MeterRegistry:

- Counter: reports merely a count over a specified property of an application

- Gauge: shows the current value of a meter

- Timers: measures latencies or frequency of events

- DistributionSummary: provides distribution of events and a simple summary

Counter vs. gauge, summary vs. histogram yazısına bakabilirsiniz

Gradle

Şöyle yaparız

implementation 'org.springframework.boot:spring-boot-starter-actuator:2.6.3'
implementation 'io.micrometer:micrometer-registry-atlas:1.8.2'
implementation 'io.micrometer:micrometer-registry-prometheus:1.8.2'

MeterRegistryCustomizer Sınıfı

Örnek

Şöyle yaparız

@Configuration
public class MicroSvcMeterRegistryConfig {
  @Value("${spring.application.name}")
  String appName;

  @Value("${env}")
  String environment;

  @Value("${instanceId}")
  String instanceId;

  @Bean
  MeterRegistryCustomizer<PrometheusMeterRegistry> configureMetricsRegistry() {
    return registry -> registry.config()
      .commonTags("appName", appName, 
                  "env", environment, 
                   "instanceId", instanceId);
  }
}

@TimedAspect Anotasyonu

Örnek

Şöyle yaparız

import org.springframework.boot.actuate.autoconfigure.metrics.MeterRegistryCustomizer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import io.micrometer.core.aop.TimedAspect;
import io.micrometer.core.instrument.MeterRegistry;

@Configuration
public class RegistryConfig {
  @Bean
  MeterRegistryCustomizer<MeterRegistry> metricsCommonTags() {
    return registry -> registry.config().commonTags("region", "someRegionName");
  }
  
  @Bean
  TimedAspect timedAspect(MeterRegistry registry) {
    return new TimedAspect(registry);
  }
}

@Timed Anotasyonu

Açıklaması şöyle

The quickest and easiest way to instrument REST controllers is to use the @Timed annotation on the controller or on individual methods of the controller. @Timed automatically adds these tags to the timer: exception, method, outcome, status, uri. It is also possible to supply additional tags to the @Timed annotation.

extraTags Alanı

Örnek

Şöyle yaparız

import io.micrometer.annotation.Timed;

@Slf4j
@RestController
public class DemoController {
  @GetMapping("/push-log")
  @ResponseStatus(HttpStatus.NO_CONTENT)
  @Timed(extraTags = {"demo-tag", "test-val"})
  public void pushLog() {
    log.debug("...");
  }
}

percentiles Alanı

Açıklaması şöyle

The '@Timed' annotation tells mirometer to record the percentiles of the method run times.

Şöyle yaparız

@Scheduled(cron = "0 10 2 ? * ?")
@SchedulerLock(name = "coinbase_avg_scheduledTask", 
        lockAtLeastFor = "PT1M", lockAtMostFor = "PT23H")
@Timed(value = "create.cb.avg", percentiles = { 0.5, 0.95, 0.99 })
public void createCbHAvg() {
  this.coinbaseService.createCbAvg();
}

MeterRegistry Sınıfı

Şu satırı dahil ederiz

import io.micrometer.core.instrument.MeterRegistry;

counter metodu - Sayaç

Counter nesnesi döndürür.

gauge metodu

gauge().set() şeklinde kullanılır

Örnek

Şöyle yaparız

@Component
public class Scheduler {

  private final AtomicInteger testGauge;
  private final Counter testCounter;

  public Scheduler(MeterRegistry meterRegistry) {
    testGauge = meterRegistry.gauge("custom_gauge", new AtomicInteger(0));
    testCounter = meterRegistry.counter("custom_counter");
  }

  @Scheduled(fixedRateString = "1000", initialDelayString = "0")
  public void schedulingTask() {
    testGauge.set(Scheduler.getRandomNumberInRange(0, 100));

    testCounter.increment();
  }

  private static int getRandomNumberInRange(int min, int max) {
    if (min >= max) {
      throw new IllegalArgumentException("max must be greater than min");
    }

    Random r = new Random();
    return r.nextInt((max - min) + 1) + min;
  }
}

Counter Sınıfı

Şu satırı dahil ederiz

import io.micrometer.core.instrument.Counter;

increment metodu

Örnek

Şöyle yaparız

import io.micrometer.core.instrument.MeterRegistry;

@RestController
@RequestMapping("/")
public class PrometheusSpringBoot {

  private final MeterRegistry meterRegistry;

  public PrometheusSpringBoot(MeterRegistry meterRegistry) {
    this.meterRegistry = meterRegistry;
  }

  @GetMapping("/2xx")
  public String simulate2xxResponse() {
    meterRegistry.counter("orders.2xx","status","OK").increment();
    return "...";
  }

  @GetMapping("/5xx")
  public String simulate5xxResponse() {
    meterRegistry.counter("orders.5xx","status","NOTOK").increment();
    return "...";
  }
	
  @PostMapping("/alert-hook")
  public void receiveAlertHook(@RequestBody Map request) throws Exception {
    System.out.println(request);
  }
}

Örnek

Şöyle yaparız

import io.micrometer.core.instrument.MeterRegistry;

@GetMapping("/2xx")
public String simulate2xxResponse() {
  meterRegistry.counter("orders.2xx","status","OK").increment();
  return "Got 2xx Response";
}

Örnek

Şöyle yaparız

import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
public class Scheduler {

  private final AtomicInteger testGauge;
  private final Counter testCounter;

  public Scheduler(MeterRegistry meterRegistry) {
    testGauge = meterRegistry.gauge("custom_gauge", new AtomicInteger(0));
    testCounter = meterRegistry.counter("custom_counter");
  }

  @Scheduled(fixedDelay=1000) // delay per 1 seconds
  public void schedulingTask() {
    testGauge.set(Scheduler.getRandomNumberInRange(0, 100));

    testCounter.increment();
  }

  private static int getRandomNumberInRange(int min, int max) {
    if (min >= max) {
      throw new IllegalArgumentException("max must be greater than min");
    }

    Random r = new Random();
    return r.nextInt((max - min) + 1) + min;
  }
}

SpringCloud Config Client

Giriş

Config Server için @EnableConfigServer yazısına bakınız.

Konfigürasyon değiştirilince tekrar okumak için @RefreshScope yazısına bakınız.

Microservice yazılımı konfigürasyon ayalarını artık Config Server'dan çekecektir

Maven

Şu satırı dahil ederiz

<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-starter-config</artifactId>
</dependency>

bootstrap.yml or bootstrap.properties Nedir

Açıklaması şöyle. İstemci tarafında kullanılır. Artık bu dosyaya gerek yok

bootstrap.yml is loaded before application.yml.

It is typically used for the following:

- when using Spring Cloud Config Server, you should specify spring.application.name and spring.cloud.config.server.git.uri inside bootstrap.yml

- some encryption/decryption information

Örnek - Eski Kullanım

Açıklaması şöyle

If you are using the latest version of spring boot use spring.config.import=... instead of spring.cloud.config.uri=...

Şöyle yaparız. Burada github'daki latest branch'ine atıfta bulunuluyor

spring.cloud.config.uri=http://localhost:8888
spring.application.name=spring-cloud-config-client
spring.cloud.config.label=latest
spring.profiles.active=dev

Örnek

Şöyle yaparız. Burada github'daki main branch'ine atıfta bulunuluyor

spring.application.name=config-client 
spring.profiles.active=development
spring.cloud.config.uri=http://localhost:8888
spring.cloud.config.label=main

Örnek - Yeni Kullanım

Şöyle yaparız.

spring.config.import=optional:configserver:http://localhost:8888 # Config sunucu adresi
spring.application.name=batch-jobs # Dosya ismi
spring.profiles.active=dev # Profile ismi

Örnek - Yeni Kullanım

Şöyle yaparız

server.port=8083

spring.application.name=neo4j-client # Dosya ismi
spring.config.import=configserver:http://localhost:8888/ # Config sunucu adresi

Örnek - Yeni Kullanım Yaml Olarak

Şöyle yaparız

spring:
  application:
    name: spring-boot-starter
  profiles:
    active: dev
  config:
    import: optional:configserver:http://localhost:8888

management:
  endpoints:
    web:
      exposure:
        include: "*"

Örnek - Retry Ayarları

Şöyle yaparız. spring.cloud.config.uri ile Config Server'ın adresi belirtilir.

spring.application.name=hello-service
server.port=8080
spring.profiles.active=development
spring.cloud.config.uri=http://localhost:8888

Eğer Config Server ulaşılamıyorsa, şu ayarlar kullanılabilir

spring.cloud.config.fail-fast
spring.cloud.config.retry.initial-interval
spring.cloud.config.retry.multiplier
spring.cloud.config.retry.max-interval
spring.cloud.config.retry.max-attempts

spring.application.name Alanı

Okunacak property dosyasını ismini belirtir. Dosya ismine aktif profile da eklenir. Açıklaması şöyle

This is the identifier for the microservice application and there will be a property file with the same name in the repository. This will identify which property file to fetch from the repository for a particular microservice. For example, for hello-service.properties, we would use hello-service-development.properties (the -development suffix tells the system to fetch code for development profile).

Örnek - Çift Spring Security Ayarı

Eğer zaten Spring Security varsa, /actuator/refresh adresini farklı bir şifre ile korumak isteyebiliriz. Şöyle yaparız. Burada @Order(1) ile actuator/refresh adresine öncelik veriliyor.

@Order(1)
@Configuration
public static class ActuatorSecurityConfig extends WebSecurityConfigurerAdapter {
  @Override
  public void configure(HttpSecurity http) throws Exception {
    http
      .csrf().disable()
      .antMatcher("/actuator/*")
      .authorizeRequests()
        .antMatchers("/actuator/*").authenticated()
      .and()
        .httpBasic();
  }
  @Override
  protected void configure(AuthenticationManagerBuilder auth) throws Exception {
    auth.inMemoryAuthentication()
      .withUser("serviceOneUser")
      .password("{noop}serviceOnePassword")
      .roles("USER");
  }
}

@Order(2)
@Configuration
public static class ApplicationSecurityConfig extends WebSecurityConfigurerAdapter {
  @Override
  public void configure(HttpSecurity http) throws Exception {
    http
      .authorizeRequests()
        .anyRequest().authenticated()
      .and()
       .oauth2Login();
  }
}

Bu adresi şifreyle tetiklemek için şöyle yaparız

curl -u serviceOneUser:serviceOnePassword -X POST http://localhost:8001/actuator/refresh

Örnek

Şöyle yaparız. Burada Config Server'dan okunacak değer @Vaue ile çekiliyor.

import org.springframework.cloud.context.config.annotation.RefreshScope;
...
@RefreshScope
@RestController
@RequestMapping("/secure")
public static class SecureController {
  @Value("${hello.message}")
  private String helloMessage;
  
  @GetMapping
  public String secure(Principal principal) {
    return helloMessage;
  }
}

SpringKafka Test @EmbeddedKafka Anotasyonu - Integration Test İçindir

Giriş

Şu satırı dahil ederiz

import org.springframework.kafka.test.context.EmbeddedKafka;

Bence TestContainers kullanmaktan daha kolay. Açıklaması şöyle

... we use the @EmbeddedKafka annotation to inject an instance of an EmbeddedKafkaBroker into our tests

Örnek

Açıklaması şöyle

... using Embedded Kafka ensures that the Kafka topics always start empty.

Maven

Şu satırı dahil ederiz

<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka-test</artifactId>
  <scope>test</scope>
</dependency>

Aslında tüm proje şöyle olmalı

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter</artifactId>
 </dependency>
 <dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-test</artifactId>
   <scope>test</scope>
  </dependency>
<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka-test</artifactId>
  <version>2.8.2</version>
  <scope>test</scope>
</dependency>
<dependency>
   <groupId>org.springframework.kafka</groupId>
   <artifactId>spring-kafka</artifactId>
</dependency>

bootstrapServersProperty Alanı

Örnek

application.yaml şöyle olsun

spring:
  config:
    activate:
      on-profile: default
  kafka:
    consumer:
      auto-offset-reset: earliest
      group-id: e2econsumer
      properties:
        max.poll.interval.ms: 8000000
        reconnect.backoff.ms: 1000
        reconnect.backoff.max.ms: 10_000
    producer:
      properties:
        reconnect.backoff.ms: 1000
        reconnect.backoff.max.ms: 10_000
test:
  topic: topic3
---
spring:
  config:
    activate:
      on-profile: serviceAProfile
  cloud:
    function:
      definition: receive
    stream:
      defaultBinder: kafka
      bindings:
        receive-in-0:
          destination: topic1
          group: service_A
          consumer:
            partitioned: true
        results-out-0:
          destination: topic2
      kafka:
        binder:
          consumer-properties:
            max.poll.interval.ms: 8000000
            reconnect.backoff.ms: 1000
            reconnect.backoff.max.ms: 10_000
          producer-properties:
            reconnect.backoff.ms: 1000
            reconnect.backoff.max.ms: 10_000
server:
  port: -1
---
spring:
  config:
    activate:
      on-profile: serviceBProfile
  cloud:
    function:
      definition: receive
    stream:
      defaultBinder: kafka
      bindings:
        receive-in-0:
          destination: topic2
          group: service_B
      kafka:
        binder:
          consumer-properties:
            max.poll.interval.ms: 8000000
            reconnect.backoff.ms: 1000
            reconnect.backoff.max.ms: 10_000
          producer-properties:
            reconnect.backoff.ms: 1000
            reconnect.backoff.max.ms: 10_000
server:
  port: -1

serviceAProfile topic1'den okuyup, topic2'ye yazıyor. serviceBProfile topic2'yi dinliyor.

Açıklaması şöyle

DirtiesContextTests — Indicates that the associated class modifies the application context as messages will be added to the topic dirtying the embedded Kafka application context.
EmbeddedKafka(partitions = 1, bootstrapServersProperty = “spring.kafka.bootstrap-servers”}) — Enables embedded kafka to start running on a random free port assigned by os.

Şöyle yaparız

@SpringBootTest
@EmbeddedKafka(partitions = 1,
        bootstrapServersProperty = "spring.kafka.bootstrap-servers",
        topics = { "topic1", "topic2" })
@DirtiesContext
@TestPropertySource(properties = "test.topic=topic2")
public class ListeningTopic2IT {
    private static ConfigurableApplicationContext serviceAContext = null;
    private static ConfigurableApplicationContext serviceBContext = null;

    @BeforeEach
    public void startServices() throws InterruptedException {
        Executors.newSingleThreadExecutor().execute(() -> {
            SpringApplication application = 
              new SpringApplication(com.demo.servicea.Application.class);
            application.setAdditionalProfiles("serviceAProfile");
            serviceAContext = application.run();
        });

        Executors.newSingleThreadExecutor().execute(() -> {
            SpringApplication application = 
              new SpringApplication(com.demo.serviceb.Application.class);
            application.setAdditionalProfiles("serviceBProfile");
            serviceBContext = application.run();
        });

        while (serviceAContext == null || !serviceAContext.isRunning()) {
            System.out.println("Waiting for service A to start, already waited for seconds:" );
            Thread.sleep(1000);
        }

        while (serviceBContext == null || !serviceBContext.isRunning()) {
            System.out.println("Waiting for service B to start, already waited for seconds:" );
            Thread.sleep(1000);
        }
    }

    @AfterEach
    public void cleanTestState() {
        serviceAContext.close();
        serviceBContext.close();
    }

    @Autowired
    private KafkaConsumer consumer;

    @Autowired
    private KafkaProducer producer;

    @Test
    public void messageSentOnTopic1CapturedOnTopic2() throws InterruptedException, IOException {
        String payload = "{\"message\" : \"message on topic 1\"}";
        this.producer.send("topic1", payload);

        consumer.getLatch().await(1, TimeUnit.MINUTES);
        assertThat(consumer.getLatch().getCount(), equalTo(0L));

        byte[] recordBytes = consumer.getRecordValue();
        assertNotNull(recordBytes);

        ObjectMapper mapper = new ObjectMapper();
        mapper.registerModule(new JavaTimeModule());

        MessageEvent messageEvent = mapper.readValue(recordBytes, MessageEvent.class);
        assertNotNull(messageEvent);
        assertEquals("message on topic 1", messageEvent.getMessage());
    }
}

brokerProperties Alanı

Örnek - Listener'ı Test Eder

Açıklaması şöyle

The @EmbeddedKafka notation starts up a clean Kafka instance as part of the Spring test context. The brokerProperties sets a non-standard port for running the Embedded Kafka. In the application-test.properties file, property spring.kafka.bootstrap-servers=localhost:19092 configures the Spring context to use this address for Kafka producers and consumers.

The test is reliable because using Embedded Kafka ensures that the Kafka topics always start empty.

test altındaki application.properties şöyledir

spring.kafka.consumer.group-id=reader-1
spring.kafka.bootstrap-servers=localhost:19092

Elimizde şöyle bir test kodu olsun. Meter sınıfı Kafka'ya mesaj gönderir. Test listener'ın azalttığı sayaç 20 saniye içinde 0 olursa başarılıdır.

@ExtendWith(SpringExtension.class)
@SpringBootTest
@ActiveProfiles("test")
@EmbeddedKafka(partitions = 2, 
  brokerProperties = { "listeners=PLAINTEXT://localhost:19092", "port=19092" })
@DirtiesContext
public class SmartMeterIntegrationTests {

  private CountDownLatch latch;

  @Autowired
  private Meter meter;

  @Autowired
  private TestListener listener;

  @Test
  public void shouldSendReadings() throws InterruptedException {
    latch = new CountDownLatch(3);
    listener.setLatch(latch);
    meter.start();
    assertTrue(latch.await(20, TimeUnit.SECONDS));
    meter.stop();
  }
}

Topicleri dinleyen şöyle bir listener kod olsun. Her kafka mesajı geldiğince sayaç bir azaltılır

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.Service;

import java.util.concurrent.CountDownLatch;

import static org.junit.jupiter.api.Assertions.assertEquals;

@Service
public class TestListener {

  private CountDownLatch latch ;

  @KafkaListener(topics = "meter.reading")
  public void listen(ConsumerRecord<String,String> cr) {
    String[] values = cr.value().split(":");
    assertEquals(3, values.length);
    if (latch != null) {
      latch.countDown();
    }
  }

  public void setLatch(CountDownLatch latch) {
    this.latch = latch;
  }
}

topics Alanı

Açıklaması şöyle

As can be seen, the topics used by the test are defined as an annotation parameter. The topics are required to be created upfront ..., even if they are configured as auto.create.topics.enable: true (which is the default) they are never automatically created. If they are not present the application will throw the following error:

MissingSourceTopicException: One or more source topics were missing during rebalance

Örnek - Listener ve Producer Test Ediliyor

Şöyle yaparız. Burada orders ve payments isimli iki topic yaratılıyor.

@SpringBootTest
@EmbeddedKafka(topics = {"orders", "payments"})
public class PaymentServiceTest {

  @Autowired
  private Producer<Order> orderProducer;
  
  @Autowired
  private Consumer<Payment> paymentConsumer;
  
  @Test
  public void shouldProcessPaymentForValidOrder() {
    var validOrder = Order
      .withCustomerId(123)
      .withItems(Item.withId("af12da"));
    
    orderProducer.send(validOrder);
    
    var payment = paymentConsumer.poll();
    assertThat(payment.getPaymentStatus()).isEqualTo("OK");
    assertThat(payment.getPaymentTransactionId()).isNotNull();
  }
}

SpringBoot Test AssertJ

Giriş

Şu satırı dahil ederiz

import static org.assertj.core.api.Assertions.*;

Açıklaması şöyle. Yani spring-boot-starter-test ile bazı kütüphaneler de otomatik olarak geliyor.

If you use the spring-boot-starter-test ‘Starter’ (in the test scope), you will find the following provided libraries:

JUnit — The de-facto standard for unit testing Java applications.

Spring Test & Spring Boot Test — Utilities and integration test support for Spring Boot applications.

AssertJ — A fluent assertion library.

Hamcrest — A library of matcher objects (also known as constraints or predicates).

Mockito — A Java mocking framework.

JSONassert — An assertion library for JSON.

JsonPath — XPath for JSON.

AssertJ bir test framework değil. JUnit içindeki testlere daha kolay assert cümleleri yazmak için. Belki aynı assert cümlelerini JUnit ile de yazmak mümkün, ancak AssertJ ile daha kolay ve kısa. Genel kullanım şuna benzer

assertThat(object1).isEqualTo(object2)

isEqualTo(), isNotNull(), isNull() en çok kullanılan metodlar

as metodu

Eğer assertion başarısız olursa, verilecek hata mesajını belirtir.

Örnek

Şöyle yaparız

@Test
public void testEqual() {
  Object object1 = new Object();
  Object object2 = object1;
  Object object3 = new Object();

  // failing assertion, with custom error message
  assertThat(object1).as("Checking whether objects are the same").isEqualTo(object3);
}

containsOnly metodu

Bir collection içinde sadece belirtilen elemanların olduğunu kontrol eder.

Örnek

Şöyle yaparız

@Test
public void testTolkienCharacterArrayList() {
  ArrayList<TolkienCharacter> characters = ...
  ...  
  TolkienCharacter aragorn = new TolkienCharacter("Aragorn", 62);
  TolkienCharacter frodo = new TolkienCharacter("Frodo", 32);  
  assertThat(characters)
    .filteredOn(character -> character.name.contains("o"))
    .containsOnly(aragorn, frodo);
}

isNotNull metodu

assertThat(..) içine verilen nesnenin null olmadığını kontrol eder

Örnek

Şöyle yaparız

@SpringBootTest
public class PaymentServiceTest {

  @Autowired
  private RestTemplate restTemplate;
  
  @Test
  public void shouldProcessPaymentForValidOrder() {
    var validOrder = Order
      .withCustomerId(123)
      .withItems(Item.withId("af12da"));
    
    var response = restTemplate.postForEntity("/payment", validOrder);
    assertThat(response.getStatusCode()).isEqualTo(HttpStatus.OK);
    assertThat(response.getBody().getPaymentTransactionId()).isNotNull();
  }
}

isNull metodu

assertThat(..) içine verilen nesnenin null olduğunu kontrol eder

Örnek

Şöyle yaparız

@SpringBootTest
public class PaymentServiceTest {

  @Autowired
  private RestTemplate restTemplate;
  
  @Test
  public void shouldRejectPaymentForOrderWithMissingCustomer() {
    var validOrder = Order
      .withItems(Item.withId("af12da"));
    
    var response = restTemplate.postForEntity("/payment", validOrder);
    assertThat(response.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST);
    assertThat(response.getBody().getPaymentTransactionId()).isNull();
  }
}