SpringJMX application.properties

Örnek

Şöyle yaparız

spring:
  jmx:
    enabled: true
    default-domain: <package_name>

SpringWebFlux Sink Sınıfı

Giriş
Şu satırı dahil ederiz 

import reactor.core.publisher.Sink;

Sanırım bir Sink asenkron programlama ile Reactor arasındaki köprüyü sağlıyor. Açıklaması şöyle

Sink allows to programmatically push reactive streams signals.

Örnek

Elimizde şöyle bir kod olsun

public Mono<Void> index(T doc) {
  IndexRequest req = indexRequest(doc);
  return Mono.create(sink -> 
    client.indexAsync(req,RequestOptions.DEFAULT,new IndexActionListener<>(sink)));
}

Burada Elasticsearch HighLevelRestClient kullanılıyor. Bu kütüphane hep listener'lar ile çalışır. Şöyle yaparız

@RequiredArgsConstructor
public class IndexActionListener<T> implements ActionListener<IndexResponse> {

  private final MonoSink<T> sink;

  @Override
  public void onResponse(IndexResponse res) {
    if (res.status().getStatus() < 400) {
      sink.success();
      return;
    }

    sink.error(new RuntimeException(res.toString()));
  }

  @Override
  public void onFailure(Exception e) {
    sink.error(e);
  }
}

SpringBatch StepBuilder Dealing With Errors

faultTolerant metodu

Örnek

Şöyle yaparız.

Step personPorcessStep(String someParam) throws Exception {
  return stepBuilderFactory.get("personProcessStep")
    .<PersonInput, PersonOutput>chunk(1)
    .reader(new PersonReader(someParam))
    .faultTolerant()
    .skipPolicy(new DataDuplicateSkipper())
    .processor(new PersonProcessor())
    .writer(new PersonWriter())
    .build();
}

listener metodu - Logging bad records

Örnek - onReadError

Elimizde şöyle bir kod olsun

public class TitleItemListenner extends ItemReadListener {

  private final static Logger logger = LoggerFactory.getLogger(TitleItemListenner.class);

  @Override
  public void onReadError(Exception ex) {
    if (ex instanceof FlatFileParseException parseException){
      String builder = "an error has occurred when reading "
                    + parseException.getLineNumber()
                    + " the line. Here are its details about the bad input\n "
                    + parseException.getInput()
                    + "\n";

      logger.error(builder,parseException);
    } else {
      logger.error("An error occur ", ex);
    }
  }
}

Şöyle yaparız

// step that load the content of the csv with titlecsvReader
// wrote them via titleItemWriter
@Bean 
public Step loadCsv(JobRepository jobRepository, 
                    PlatformTransactionManager transactionManager, 
                    FlatFileItemReader<Title> titleCsvReader){

  return new StepBuilder("load csv",jobRepository)
    .<Title,Title>chunk(10,transactionManager)
    .reader(titleCsvReader)
    .writer(titleItemWriter())
    .faultTolerant()
    .skipLimit(100)
    .skip(Exception.class)
    .listener(new TitleItemListenner())// adding the listenner for logging bad records
    .build();
}

Örnek - afterStep

Elimizde şöyle bir kod olsun

@Component
public class StepCompletionNotificationListener implements StepListener {

  @BeforeStep
  public void beforeStep(StepExecution stepExecution){
    log.info("Step {} is started time {}",
      stepExecution.getStepName(),stepExecution.getStartTime());
  }

  @AfterStep
  public ExitStatus afterStep(StepExecution stepExecution){
    log.info("Step {} is ended time {}",
      stepExecution.getStepName(),stepExecution.getEndTime());
    return stepExecution.getExitStatus() == ExitStatus.COMPLETED ? 
      ExitStatus.COMPLETED : ExitStatus.FAILED;
  }
}

Şöyle yaparız

@Bean
public Step stepRealNewsReport(
  @Qualifier("readerNewsReportByRealStatus") ItemReader<NewsReport> itemReader, 
  @Qualifier("writerRealNewsReportTable") ItemWriter<RealNews> itemWriter, 
  StepCompletionNotificationListener listener) {

  return stepBuilderFactory
    .get("stepRealNewsReport")
    .listener(listener)
    .<NewsReport, RealNews>chunk(50)
    .reader(itemReader)
    .processor(realNewsProcessor())
    .writer(itemWriter)
    .build();
}

Örnek - afterStep

Elimizde şöyle bir kod olsun

public class EmptyInputFailer implements StepExecutionListener {
  @Override
  public ExitStatus afterStep(StepExecution stepExecution) {
    if (stepExecution.getReadCount() > 0){
      return stepExecution.getExitStatus();
   }
   else return ExitStatus.FAILED;
  }
}

Örnek

Şöyle yaparız

// step that load the content of the csv with titlecsvReader
// wrote them via titleItemWriter
 @Bean 
public Step loadCsv(JobRepository jobRepository, 
                    PlatformTransactionManager transactionManager, 
                    FlatFileItemReader<Title> titleCsvReader){

  return new StepBuilder("load csv",jobRepository)
    .<Title,Title>chunk(10,transactionManager)
    .reader(titleCsvReader)
    .writer(titleItemWriter())
    .listener(new EmptyInputFailer())
}

retryLimit metodu

Açıklaması şöyle

Spring Batch provides comprehensive support for error handling and retrying. Developers can configure retry policies for each step in the job, enabling the application to recover from transient errors such as network failures or database timeouts.

Örnek

Şöyle yaparız

@Bean
public Step stockStatsStep() {
  return stepBuilderFactory.get("stockStatsStep")
    .<StockPrice, StockStats>chunk(100)
    .reader(stockPriceItemReader())
    .processor(stockPriceItemProcessor())
    .writer(stockStatsItemWriter())
    .faultTolerant()
    .retryLimit(3)
    .retry(MyException.class)
    .build();
}

public class MyException extends Exception {

  public MyException(String message) {
   super(message);
  }
}

Açıklaması şöyle

In this implementation, we have configured the `stockStatsStep` step to be fault-tolerant with a retry limit of 3. We have also specified that we want to retry the step in case of an exception of type `MyException`.

If an exception occurs during the processing of a chunk, the step will be retried up to the configured limit. If the retry limit is exceeded, the step will fail, and the job will stop.

We have defined a custom exception called `MyException` that extends the `Exception` class. This exception can be thrown by the `StockPriceItemProcessor` or any other component in the step that needs to be retried.

You can customize these implementations based on your specific error handling and retry requirements. Spring Batch provides many other features and options for error handling, such as skip and recovery, which can also be configured for each step in the job.

skip metodu - Skipping record

Açıklaması şöyle

skip : Determine the exception (exception class) to skip

skipLimit : The number of times you want to make a skip

noSkip : Identify the exception you don’t want to skip

Örnek

Şöyle yaparız

// step that load the content of the csv with titlecsvReader
// wrote them via titleItemWriter
@Bean 
public Step loadCsv(JobRepository jobRepository, 
                    PlatformTransactionManager transactionManager, 
                    FlatFileItemReader<Title> titleCsvReader){

  return new StepBuilder("load csv",jobRepository)
    .<Title,Title>chunk(10,transactionManager)
    .reader(titleCsvReader)
    .writer(titleItemWriter())
    .faultTolerant() // tell to spring batch that this step can face errors
    .skip(Exception.class) // skipp all Exception 
    .noSkip(FlatFileParseException.class) // but do not skip this one 
    .skipLimit(20) // the the number of times you want to skip Exeception.class
    .build();
}

SpringBatch KafkaItemWriter Sınıfı

Örnek

Şöyle yaparız

@Configuration
public class TitleBatchConfig {

  private final KafkaTemplate<String,Title> template;
  
  @Bean
  public KafkaItemWriter<String,Title> titleItemWriter(){
    return new KafkaItemWriterBuilder<String,Title>()
      .kafkaTemplate(this.template)
      .itemKeyMapper(Title::id)
      .build();
  }  
  ...
}

SpringWebFlux Mono.timeout metodu

Örnek

Şöyle yaparız

webClient.get()
    .uri("/endpoint")
    .retrieve()
    .bodyToMono(String.class)
    .timeout(Duration.ofSeconds(10));

SpringWebFlux Mono.retryWhen metodu

Örnek

Şöyle yaparız

this.retryBackoffSpec = Retry.backoff(maxAttempts, Duration.ofSeconds(backoffSeconds))
  .doBeforeRetry(retrySignal -> 
    log.debug("Waiting {} seconds. Retry #{} of {} after exception: {}",
        backoffSeconds, (retrySignal.totalRetriesInARow()+1), maxAttempts,
        retrySignal.failure().getLocalizedMessage()
    ))
  .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> retrySignal.failure());

this.webClient.post().uri(uri)
  .bodyValue(emailText)
  .retrieve()
  …
  .bodyToMono(CtfdUserResponse.class)
  .retryWhen(retryBackoffSpec)
  .block();

Örnek

Şöyle yaparız

WebClient webClient = WebClient.builder()
  .baseUrl("http://example.com")
  .build();

Mono<String> response = webClient.get()
    .uri("/retry-endpoint")
    .retrieve()
    .bodyToMono(String.class)
    // number of retries and backoff configuration
    .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
      // maximum backoff time
      .maxBackoff(Duration.ofSeconds(10))) 
    // fallback if retries all fail
    .onErrorResume(e -> Mono.just("Fallback response")); 

response.subscribe(result -> System.out.println(result));

Örnek

Şöyle yaparız

Predicate<Throwable> exceptionFilter() {
  return throwable -> throwable instanceof RuntimeException
      || (throwable instanceof WebClientResponseException
          && (throwable.getStatusCode() == HttpStatus.GATEWAY_TIMEOUT
              || throwable.getStatusCode() == HttpStatus.SERVICE_UNAVAILABLE
              || throwable.getStatusCode() == HttpStatus.BAD_GATEWAY));
}

Retry retry = Retry.backoff(3, Duration.ofSeconds(2))
    .jitter(0.7)
    .filter(exceptionFilter())
    .onRetryExhaustedThrow((retrySpec, retrySignal) -> {
      log.error("Service at {} failed to respond, after max attempts of: {}", 
        uri, retrySignal.totalRetries());
      return retrySignal.failure();
    });

SpringWebFlux Transitioning from RestTemplate to WebClient in Spring Boot

GET

Örnek

Şöyle yaparız

RestTemplate restTemplate = new RestTemplate();
ResponseEntity<String> response = restTemplate.getForEntity("http://example.com",
  String.class);


WebClient webClient = WebClient.create();
Mono<String> response = webClient.get()
    .uri("http://example.com")
    .retrieve()
    .bodyToMono(String.class);
response.subscribe(result -> System.out.println(result));

Handling Errors

Açıklaması şöyle

RestTemplate’s error handling occurs through the ErrorHandler interface, which requires a separate block of code. WebClient streamlines this with more fluent handling.

Örnek

Şöyle yaparız

WebClient webClient = WebClient.create();
webClient.get()
    .uri("http://example.com/some-error-endpoint")
    .retrieve()
    .onStatus(HttpStatus::isError, response -> {
        // Handle error status codes
        return Mono.error(new CustomException("Custom error occurred."));
    })
    .bodyToMono(String.class);

Açıklaması şöyle

The onStatus() method allows for handling specific HTTP statuses directly within the chain of operations, providing a more readable and maintainable approach.

POST

Örnek

Şöyle yaparız

RestTemplate restTemplate = new RestTemplate();
HttpHeaders headers = new HttpHeaders();
headers.setContentType(MediaType.APPLICATION_JSON);
HttpEntity<String> request = new HttpEntity<>("{\"key\":\"value\"}", headers);
ResponseEntity<String> response = restTemplate
  .postForEntity("http://example.com",  request, String.class);

WebClient webClient = WebClient.create();
Mono<String> response = webClient.post()
    .uri("http://example.com")
    .contentType(MediaType.APPLICATION_JSON)
    .bodyValue("{\"key\":\"value\"}")
    .retrieve()
    .bodyToMono(String.class);

Asynchronous Processing