SpringKafka Consumer SeekToCurrentErrorHandler

Giriş

Şu satırı dahil ederiz

import org.springframework.kafka.listener.SeekToCurrentErrorHandler;

Stateful Retry İçindir. Açıklaması şöyle

Stateful retry is not offered by the Java Apache client by itself, but is available as a configuration option out of the box with Spring Kafka, using the SeekToCurrentErrorHandler.

Açıklaması şöyle

An additional feature of the SeekToCurrentErrorHandler is that those events within the batch that have successfully been processed prior to the event that results in a RetryableException being thrown are still able to be successfully marked as consumed, so are not themselves re-delivered too in the next poll.

constructor - Custom ConsumerRecord + BackOff 

Kullanmayın

Örnek - FixedBackOff

Şöyle yaparız

//Stateful retry listener.
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> 
  kafkaStatefulRetryListenerContainerFactory
    ConsumerFactory<String, String> consumerFactory) {

  SeekToCurrentErrorHandler errorHandler = new SeekToCurrentErrorHandler(
    (record, exception) -> {}, 
                           // 4 seconds pause, 4 retries.
                           new FixedBackOff(4000L, 4L));

  ConcurrentKafkaListenerContainerFactory<String, String> factory = 
    new ConcurrentKafkaListenerContainerFactory();
  factory.setConsumerFactory(consumerFactory);
  factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.RECORD);
  factory.setErrorHandler(errorHandler);
  return factory;
}

Örnek - ExponentialBackOff 
Elimizde şöyle bir kod olsun. SeekToCurrentErrorHandler kullanımı ile ilgili bir örnek burada ve burada.

@Configuration
@EnableKafka
public class SpringConfiguration {

  @Bean
  public SeekToCurrentErrorHandler eh() {
    long initialMillis = 500;
    long factor = 2;
    long maxElapsedTimeSecs = 60;
    ExponentialBackOff backoff = new ExponentialBackOff(initialMillis, factor);
    backoff.setMaxElapsedTime(maxElapsedTimeSecs*1000);

    BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer = (rec, exc) ->  {
      // TODO In the final app do something more useful here
      logger.error("* Maximum retry policy has been reached {} - acknowledging and
        proceeding *", rec);
    };

    SeekToCurrentErrorHandler eh = new SeekToCurrentErrorHandler(recoverer, backoff);
    eh.setCommitRecovered(true);
    return eh;
  }
  ...
}

Örnek - FixedBackOff 

Şöyle yaparız

//Stateful retry listener.
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> 
  kafkaStatefulRetryListenerContainerFactory
    ConsumerFactory<String, String> consumerFactory) {

  SeekToCurrentErrorHandler errorHandler = new SeekToCurrentErrorHandler(
    (record, exception) -> {}, 
                           // 4 seconds pause, 4 retries.
                           new FixedBackOff(4000L, 4L));

  ConcurrentKafkaListenerContainerFactory<String, String> factory = 
    new ConcurrentKafkaListenerContainerFactory();
  factory.setConsumerFactory(consumerFactory);
  factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.RECORD);
  factory.setErrorHandler(errorHandler);
  return factory;
}

Örnek - ExponentialBackOff 

Elimizde şöyle bir kod olsun. SeekToCurrentErrorHandler kullanımı ile ilgili bir örnek burada ve burada.

@Configuration
@EnableKafka
public class SpringConfiguration {

  @Bean
  public SeekToCurrentErrorHandler eh() {
    long initialMillis = 500;
    long factor = 2;
    long maxElapsedTimeSecs = 60;
    ExponentialBackOff backoff = new ExponentialBackOff(initialMillis, factor);
    backoff.setMaxElapsedTime(maxElapsedTimeSecs*1000);

    BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer = (rec, exc) ->  {
      // TODO In the final app do something more useful here
      logger.error("* Maximum retry policy has been reached {} - acknowledging and
        proceeding *", rec);
    };

    SeekToCurrentErrorHandler eh = new SeekToCurrentErrorHandler(recoverer, backoff);
    eh.setCommitRecovered(true);
    return eh;
  }
  ...
}

Şöyle yaparız.

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String>
  kafkaListenerContainerFactory(RetryTemplate retryTemplate) {
  ConcurrentKafkaListenerContainerFactory<String, String> factory =
    new ConcurrentKafkaListenerContainerFactory<>();
  factory.setConsumerFactory(consumerFactory());
  factory.setMissingTopicsFatal(missingTopicsFatal);//True in prod, false otherwise

  factory.getContainerProperties().setAckMode
    (ContainerProperties.AckMode.MANUAL_IMMEDIATE);
  factory.getContainerProperties().setSyncCommitTimeout(Duration.ofSeconds(60));
  factory.setStatefulRetry(true);
  factory.setErrorHandler(eh());
  return factory;
}

constructor - Sadece BackOff 

Örnek - FixedBackOff

Şöyle yaparız. Eğer hata varsa tekrar işlemeye çalışır. Her hata arasında 60 saniye bekler.

@Bean
KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<byte[], byte[]>>
 kafkaListenerContainerFactory() {
  ConcurrentKafkaListenerContainerFactory<byte[], byte[]> factory =
    new ConcurrentKafkaListenerContainerFactory<>();
  factory.setErrorHandler(new SeekToCurrentErrorHandler(
    new FixedBackOff(60000L))); // per listener
  factory.setConsumerFactory(consumerFactory());
  return factory;
}

Şöyle yaparız.

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String>
  kafkaListenerContainerFactory(RetryTemplate retryTemplate) {
  ConcurrentKafkaListenerContainerFactory<String, String> factory =
    new ConcurrentKafkaListenerContainerFactory<>();
  factory.setConsumerFactory(consumerFactory());
  factory.setMissingTopicsFatal(missingTopicsFatal);//True in prod, false otherwise

  factory.getContainerProperties().setAckMode
    (ContainerProperties.AckMode.MANUAL_IMMEDIATE);
  factory.getContainerProperties().setSyncCommitTimeout(Duration.ofSeconds(60));
  factory.setStatefulRetry(true);
  factory.setErrorHandler(eh());
  return factory;
}

Örnek - FixedBackOff

Şöyle yaparız. Eğer hata varsa tekrar işlemeye çalışır. Her hata arasında 60 saniye bekler.

@Bean
KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<byte[], byte[]>>
 kafkaListenerContainerFactory() {
  ConcurrentKafkaListenerContainerFactory<byte[], byte[]> factory =
    new ConcurrentKafkaListenerContainerFactory<>();
  factory.setErrorHandler(new SeekToCurrentErrorHandler(
    new FixedBackOff(60000L))); // per listener
  factory.setConsumerFactory(consumerFactory());
  return factory;
}

SpringContext ClassPathScanningCandidateComponentProvider Sınıfı

Giriş

Şu satırı dahil ederiz

import org.springframework.context.annotation.ClassPathScanningCandidateComponentProvider;

Örnek

Şöyle yaparız

import org.springframework.context.annotation.ClassPathScanningCandidateComponentProvider;
import org.springframework.core.type.filter.AnnotationTypeFilter;

ClassPathScanningCandidateComponentProvider scanner = 
  new ClassPathScanningCandidateComponentProvider(false);
scanner.addIncludeFilter(new AnnotationTypeFilter(FooAnnotation.class));

Set<BeanDefinition> beanDefinitions = scanner.findCandidateComponents("com.foo");

if (!CollectionUtils.isEmpty(beanDefinitions)) {
  for (final BeanDefinition beanDefinition : beanDefinitions) {
    Class<?> fooClass = Class.forName(beanDefinition.getBeanClassName());
    ...
  }
}

SpringCloud @EnableDiscoveryClient Anotasyonu

Giriş

Açıklaması şöyle

@EnableDiscoveryClient which comes from spring-cloud-commons is the same as @EnableEurekaClient but it is a more generic implementation of “Discovery Service”.

@EnableEurekaClient only works with Eureka whereas @EnableDiscoveryClient works with eureka, consul, zookeeper. But if Eureka is on the application classpath, they are effectively the same.

@EnableZookeeper Anotasyonu

Maven

Şu satırı dahil ederiz

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.5.7</version>
</dependency>

Örnek

Şöyle yaparız

@Configuration
@EnableZookeeper
public class ZooKeeperConfiguration extends ZookeeperProperties {
    public ZooKeeperConfiguration() {
        setConnectString("localhost:2181");
        setBaseSleepTimeMs(1000);
        setMaxRetries(3);
        setSessionTimeoutMs(5000);
    }
}

@Service
public class MyService {
  @Autowired
  private ZooKeeperClient zooKeeperClient;

  public List<String> getNodes(String path) throws KeeperException, InterruptedException {
    return zooKeeperClient.getChildren(path);
  }
}

Açıklaması şöyle

Next, you’ll need to configure the connection to your ZooKeeper server. You can do this by creating a ZooKeeperConfiguration class that extends org.springframework.cloud.zookeeper.ZookeeperProperties and sets the connection properties, like so:

This sets the connection string to localhost:2181 and configures some other properties for the connection.

Finally, you can use the ZooKeeper client in your Spring Boot application by autowiring a ZooKeeperClient instance. 

SpringWebFlux Flux.merge

Örnek

Şöyle yaparız

public Flux<Incidence> findAllFlux() {
  Flux<Incidence> lstPre1 = WebClient.create("...").get().retrieve()
			.bodyToFlux(Incidence.class);
  Flux<Incidence> lstPre2 = WebClient.create("...").get().retrieve()
			.bodyToFlux(Incidence.class);
  Flux<Incidence> lstPro1 = WebClient.create("...").get().retrieve()
			.bodyToFlux(Incidence.class);
  Flux<Incidence> lstPro2 = WebClient.create("...").get().retrieve()
			.bodyToFlux(Incidence.class);
  return Flux.merge(lstPre1, lstPre2, lstPro1, lstPro2);
}

Flux.delaySequence

Giriş

delayElements metodundan farklı. Açıklaması şöyle. Yani initial delay olarak düşünülebilir.

With this operator, a source emitting at 10Hz with a delaySequence Duration of 1s will still emit at 10Hz, with an initial "hiccup" of 1s. On the other hand, delayElements(Duration) would end up emitting at 1Hz.

Örnek

Şöyle yaparız

@GetMapping("/incidenceFlux/{env}/{server}")
public Flux<Incidence> findAllFlux(@PathVariable String env,
                                   @PathVariable String server) {
  return Flux.just(new Incidence(...),
	  	   new Incidence(...),
		   new Incidence(...)
    )
    .delaySequence(Duration.ofSeconds(3));
}