SpringKafka Consumer KafkaListenerEndpointRegistry Sınıfı - Dinamik Olarak Listener Takılabilir

Giriş

Şu satırı dahil ederiz

import  org.springframework.kafka.config.KafkaListenerEndpointRegistry;

Açıklaması şöyle

... we cannot delete the consumers we have registered, because it is not possible to do this using the KafkaListenerEndpointRegistry class. In order to make this possible, we need to create our own registry class. 

getListenerContainer

MessageListenerContainer nesnesi döner

getListenerContainerIds metodu

Örnek

Şöyle yaparız. Burada MessageListenerContainer elde edildikten sonra start(), pause(), resume(), getAssignedPartitions(), getListenerId(), getGroupId() bir sürü metodu kullanılabilir.

@Autowired
private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

@GetMapping
public List<KafkaConsumerResponse> getConsumerIds() {
  return kafkaListenerEndpointRegistry.getListenerContainerIds()
    .stream()
    .map(this::createKafkaConsumerResponse)
    .collect(Collectors.toList());
}

private KafkaConsumerResponse createKafkaConsumerResponse(String consumerId) {
  MessageListenerContainer listenerContainer = kafkaListenerEndpointRegistry
    .getListenerContainer(consumerId);
  ...
}

registerListenerContainer metodu

Açıklaması şöyle

When we create a kafka consumer using the @KafkaListener annotation, it will be read by the spring when the application is run. KafkaListenerAnnotationBeanPostProcessor is class that responsible to read those annotation. Furthermore, the kafka consumers that we have created will be registered in the KafkaListenerEndpointRegistry class by KafkaListenerEndpointRegistrar class. When the application is running, KafkaListenerEndpointRegistry will start the consumers that have been registered based on the autoStartup property.

Açıklaması şöyle

The method accepts 3 parameters, that is KafkaListenerEndpoint, KafkaListenerContainerFactory, and Boolean startImmediately. 

KafkaListenerEndpoint yazısına bakabilirsiniz

Örnek

Şöyle yaparız

@Autowired
private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

@Autowired
private KafkaListenerContainerFactory kafkaListenerContainerFactory;

MethodKafkaListenerEndpoint<String, String> kafkaListenerEndpoint = ...;
boolean startImmediately = ...;

kafkaListenerEndpointRegistry.registerListenerContainer(kafkaListenerEndpoint,
  kafkaListenerContainerFactory, 
  startImmediately);

Örnek

Şöyle yaparız

@Autowired
private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

@Autowired
private KafkaListenerContainerFactory kafkaListenerContainerFactory;

public void createAndRegisterListener(String topic) {
  KafkaListenerEndpoint listener = ...;
  kafkaListenerEndpointRegistry.registerListenerContainer(listener, 
    kafkaListenerContainerFactory, true);
}

SpringWebFlux ServerResponse Sınıfı

Giriş

Şu satırı dahil ederiz

import org.springframework.web.reactive.function.server.ServerResponse;

body metodu

İçine bir Mono nesnesi konulabilir.

Örnek

Şöyle yaparız

import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.ServerResponse;

import static org.springframework.web.reactive.function.server.RequestPredicates.GET;
import static org.springframework.web.reactive.function.server.RouterFunctions.route;

@Configuration
public class SampleRouter {
  @Bean
  public RouterFunction<ServerResponse> sampleRoute(SampleHandler handler) {
    return route(GET("/simple-route"), handler::handleRequest);
  }
}

import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.ServerResponse;

@Component
public class SampleHandler {
  public Mono<ServerResponse> handleRequest(ServerRequest request) {
    return ServerResponse.ok().body(Mono.just("Sample Route Successful!"), String.class);
  }
}

SpringData Jdbc AbstractRoutingDataSource Sınıfı - Separate Database İle Multitenant Yapı İçindir

Giriş

Şu satırı dahil ederiz

import org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource;

Kısaca

1. AbstractRoutingDataSource sınıfından kalıtan bir bean kodlarız. Bu sınıfta determineCurrentLookupKey() metodunu olmalıdır

2. AbstractRoutingDataSource sınıfından kalıtan bir bean nesnemize setTargetDataSources() ile hedef DataSource nesnelerini atarız.

3. Bu sınıf bir ThreadLocal ile birlikte kullanılır. ThreadLocal.set() ile bir enum veya string verilir. Bu enum veya string'e denk gelen DataSource nesnesi AbstractRoutingDataSource içinde setTargetDataSources() ile atanmıştır.

Açıklaması şöyle. Yani aslında AbstractRoutingDataSource sınıfından kalıtsak bile yine bir DataSource yaratıyoruz.

Abstract DataSource implementation that routes getConnection() calls to one of various target DataSources based on a lookup key. The latter is usually (but not necessarily) determined through some thread-bound transaction context.

determineCurrentLookupKey metodu

Açıklaması şöyle. Bu metod Spring tarafından çağrılır ve hangi DataSource'un kullanılacağını döner. Spring'de kendi içindeki Map'i arayarak ilgili DataSource nesnesini kullanır

A component that extends AbstractRoutingDataSource and is responsible to provide the list of datasources and also to provide the implementation of the determineCurrentLookupKey() method which will help in determining the current datasource.

Örnek

Şöyle yaparız

@Configuration
public class DataSourceConfig {

  @Bean
  public DataSource dataSource() {
    TenantRoutingDataSource customDataSource = new TenantRoutingDataSource();
    Map<Object, Object> targetDataSources = new HashMap<>();
    // Populate targetDataSources map with tenant's DataSource
    // ...
    customDataSource.setTargetDataSources(targetDataSources);
    return customDataSource;
  }
}

Açıklaması şöyle

In this example, TenantRoutingDataSource extends AbstractRoutingDataSource from Spring and overrides determineCurrentLookupKey() method to provide routing based on tenant context.

Örnek - Enum

Şöyle yaparız

public class DataSourceRouter extends AbstractRoutingDataSource {
  
  @Override
  protected Object determineCurrentLookupKey() {
    if (AsyncContextHolder.getAsyncContext() != null) {
      return AsyncContextHolder.getAsyncContext().get(READTYPE);
    }
    return null;
  }  
}

Elimizde şöyle bir enum olsun

public enum ClientDatabase {
  CLIENT_A, CLIENT_B
}

DataSourceRouter yaratmak için şöyle yaparız. Bu nesneye iki tane DataSource atanıyor

@Bean
public DataSource clientDatasource() {
  Map<Object, Object> targetDataSources = new HashMap<>();
  DataSource clientADatasource = clientADatasource();
  DataSource clientBDatasource = clientBDatasource();

  targetDataSources.put(ClientDatabase.CLIENT_A,clientADatasource);
  targetDataSources.put(ClientDatabase.CLIENT_B, clientBDatasource);

  DataSourceRouter datasourceRouter   = new ClientDataSourceRouter();
  datasourceRouter.setTargetDataSources(targetDataSources);
  datasourceRouter.setDefaultTargetDataSource(clientADatasource);
  return clientRoutingDatasource;
}

Örnek - Enum

Elimizde şöyle bir kod olsun. Burada enum içeren ThreadLocal nesne tanımlanıyor

import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

@Component
@Scope(value = ConfigurableBeanFactory.SCOPE_SINGLETON)
public class DataSourceContextHolder {
  private static ThreadLocal<DataSourceEnum> threadLocal; 

  public DataSourceContextHolder() {
    threadLocal = new ThreadLocal<>();
  }

  public void setDataSourceEnum(DataSourceEnum dataSourceEnum) {
    threadLocal.set(dataSourceEnum);
  }

  public DataSourceEnum getDataSourceEnum() {
    return threadLocal.get();
  }

  public static void clearDataSourceEnum() {
    threadLocal.remove();
  }
}

Şöyle yaparız. Burada constructor içinde setTargetDataSources ile her enum'a denk gelen DataSource atanıyor. Ayrıca varsayılan DataSource ta atanıyor.

import org.springframework.jdbc.datasource.DriverManagerDataSource;
import org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource;
import org.springframework.stereotype.Component;

@Component
public class DataSourceRouting extends AbstractRoutingDataSource {
  private DataSourceOneConfig dataSourceOneConfig;
  private DataSourceTwoConfig dataSourceTwoConfig;
  private DataSourceContextHolder dataSourceContextHolder;

  public DataSourceRouting(DataSourceContextHolder dataSourceContextHolder,
                           DataSourceOneConfig dataSourceOneConfig,
    		     	   DataSourceTwoConfig dataSourceTwoConfig) {
    this.dataSourceOneConfig = dataSourceOneConfig;
    this.dataSourceTwoConfig = dataSourceTwoConfig;
    this.dataSourceContextHolder = dataSourceContextHolder;

    Map<Object, Object> dataSourceMap = new HashMap<>();
    dataSourceMap.put(DataSourceEnum.DATASOURCE_ONE, dataSourceOneDataSource());
    dataSourceMap.put(DataSourceEnum.DATASOURCE_TWO, dataSourceTwoDataSource());
    this.setTargetDataSources(dataSourceMap);
    this.setDefaultTargetDataSource(dataSourceOneDataSource());
  }

  @Override
  protected Object determineCurrentLookupKey() {
    return dataSourceContextHolder.getBranchContext(); //Thread local object
  }
}

Controller içinde şöyle yaparız. Burada ThreadLocal nesneye değer tanıyor

@RestController
@RequiredArgsConstructor
public class DetailsController {

  private final DataSourceContextHolder dataSourceContextHolder;

  @GetMapping(value="/getEmployeeDetails/{dataSourceType}")
  public List<Employee> getAllEmployees(@PathVariable("dataSourceType")
                                        String dataSourceType){
    if(DataSourceEnum.DATASOURCE_TWO.toString().equals(dataSourceType)){
      dataSourceContextHolder.setBranchContext(DataSourceEnum.DATASOURCE_TWO);
    } else {
      dataSourceContextHolder.setBranchContext(DataSourceEnum.DATASOURCE_ONE);
    }
    return employeeService.getAllEmployeeDetails(); 
  }
}

Örnek - AOP + Anotasyon

Şöyle yaparız. Burada isme sahip iki tane DataSource tanımlanıyor. Bir tanesi varsayılan DataSource

public class AbstractRoutingDataSourceImpl extends AbstractRoutingDataSource {

  private static final ThreadLocal<String> DATABASE_NAME = new ThreadLocal<>();

  public AbstractRoutingDataSourceImpl(DataSource defaultTargetDatasource, 
                                       Map<Object,Object> targetDatasources) {
    super.setDefaultTargetDataSource(defaultTargetDatasource);
    super.setTargetDataSources(targetDatasources);
    super.afterPropertiesSet();
  }
  public static void setDatabaseName(String key) {
    DATABASE_NAME.set(key);
  }

  public static String getDatabaseName() {
    return DATABASE_NAME.get();
  }

  public static void removeDatabaseName() {
    DATABASE_NAME.remove();
  }

  @Override
  protected Object determineCurrentLookupKey() {
    return DATABASE_NAME.get();
  }
}

İki DataSource şöyle yaratılır

@Configuration
@EnableJpaRepositories(basePackages = "com.dynamicdatasource.demo",entityManagerFactoryRef = "entityManager")
public class DynamicDatabaseRouter {

    public static final String PROPERTY_PREFIX = "spring.datasource.";

    @Autowired
    private Environment environment;

    @Bean
    @Primary
    @Scope("prototype")
    public AbstractRoutingDataSourceImpl dataSource() {
        Map<Object, Object> targetDataSources = getTargetDataSources();
        return new AbstractRoutingDataSourceImpl((DataSource)targetDataSources.get("default"), targetDataSources);
    }

    @Bean(name = "entityManager")
    @Scope("prototype")
    public LocalContainerEntityManagerFactoryBean entityManagerFactoryBean(EntityManagerFactoryBuilder builder) {
        return builder.dataSource(dataSource()).packages("com.dynamicdatasource").build();
    }

    private Map<Object,Object> getTargetDataSources() {
        
        //loading the database names to a list from application.properties file
        List<String> databaseNames = environment.getProperty("spring.database-names.list",List.class);
        Map<Object,Object> targetDataSourceMap = new HashMap<>();

        for (String dbName : databaseNames) {

                DriverManagerDataSource dataSource = new DriverManagerDataSource();
                dataSource.setDriverClassName(envioronment.getProperty(PROPERTY_PREFIX + dbName + ".driver"));
                dataSource.setUrl(environment.getProperty(PROPERTY_PREFIX + dbName + ".url"));
                dataSource.setUsername(environment.getProperty(PROPERTY_PREFIX + dbName + ".username"));
                dataSource.setPassword(environment.getProperty(PROPERTY_PREFIX + dbName + ".password"));
                targetDataSourceMap.put(dbName,dataSource);

        }
        targetDataSourceMap.put("default",targetDataSourceMap.get(databaseNames.get(0)));
        return targetDataSourceMap;
    }
}

Burada bir aspect kodlanıyor

@Aspect
@Component
@Order(-10)
public class DataSourceAspect {
    private final Logger logger = LoggerFactory.getLogger(getClass());

    //defininining where the jointpoint need to be applied
    @Pointcut("@annotation(com.dynamicdatasource.demo.config.SwitchDataSource)")
    public void annotationPointCut() {
    }

    // setting the lookup key using the annotation passed value
    @Before("annotationPointCut()")
    public void before(JoinPoint joinPoint){
        MethodSignature sign =  (MethodSignature)joinPoint.getSignature();
        Method method = sign.getMethod();
        SwitchDataSource annotation = method.getAnnotation(SwitchDataSource.class);
        if(annotation != null){
            AbstractRoutingDataSourceImpl.setDatabaseName(annotation.value());
            logger.info("Switch DataSource to [{}] in Method [{}]",
                annotation.value(), joinPoint.getSignature());
        }
    }
    
    // restoring to default datasource after the execution of the method
    @After("annotationPointCut()")
    public void after(JoinPoint point){
        if(null != AbstractRoutingDataSourceImpl.getDatabaseName()) {
            AbstractRoutingDataSourceImpl.removeDatabaseName();
        }
    }
}

Aspect için anotasyon şöyle olsun

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface SwitchDataSource {

    String value() default "";

}

Kullanırken şöyle yaparız

@SwitchDataSource(value = "college")
public List<College> getAllColleges(){
  return collegeRepository.findAll();
}

@SwitchDataSource(value = "student")
public List<Student> getAllStudents(){
  return studentRepository.findAll();
}

Örnek - TransactionSynchronizationManager

Burada amaç @Transactional ve @Transactional(readOnly = true) olarak işaretli çağrıları farklı veri tabanlarına göndermek. Şeklen şöyle

AbstractRoutingDataSource şöyledir

public class TransactionRoutingDataSource extends AbstractRoutingDataSource {

  @Override
  protected Object determineCurrentLookupKey() {
    return TransactionSynchronizationManager.isCurrentTransactionReadOnly() ?
      DataSourceType.READ_ONLY :
      DataSourceType.READ_WRITE;
  }
}

Bu nesneyi yaratmak ve doldurmak için şöyle yaparız

public class TransactionRoutingConfiguration 
        extends AbstractJPAConfiguration {

   ...
  @Bean
  public DataSource readWriteDataSource() {
    ...
  }

  @Bean
  public DataSource readOnlyDataSource() {
    ...
  }

  @Bean
  public TransactionRoutingDataSource actualDataSource() {
    TransactionRoutingDataSource routingDataSource = 
      new TransactionRoutingDataSource();

    Map<Object, Object> dataSourceMap = new HashMap<>();
    dataSourceMap.put(
      DataSourceType.READ_WRITE, 
      readWriteDataSource()
    );
    dataSourceMap.put(
      DataSourceType.READ_ONLY, 
      readOnlyDataSource()
    );

    routingDataSource.setTargetDataSources(dataSourceMap);
    return routingDataSource;
  }
}

SpringData Projections - Open Projections - SpEL Kullanır

Giriş

Open Projections içinde SpEL kullanılır. Bu kullanımda Spring sorgu cümlesini optimize etmez. Açıklaması şöyle.

Spring Data cannot apply query execution optimizations in this case, because the SpEL expression could use any attribute of the aggregate root.

Örnek

Elimizde şöyle bir @Entity kodu olsun

@Data
@Entity
@Table(name = "CUSTOMER_ORDER")
public class OrderEntity {

  @Id
  @Column(name = "ID")
  private Integer id;

  @Column(name = "ORDER_NUMBER")
  private String orderNumber;

  @Column(name = "TOTAL_AMOUNT")
  private String totalAmount;

  @OneToOne
  @JoinColumn(name="CUSTOMER_ID", nullable=false)
  private CustomerEntity customer;
}

Daha sonra projection kodunu tanımlamak için şöyle yaparız

public interface CustomerDetailsDTO {

  Integer getCustomerId();

  @Value("#{target.firstName + ' ' + target.lastName}")
  String getCustomerName();

  String getCity();

  String getCountry();

  @Value("#{@mapperUtility.buildOrderDTO(target.orderNumber, target.totalAmount)}")
  OrderDTO getOrder();
}

Burada SpEL içinde bir başka kod kullanılıyor. O da şöyle

@Component
public class MapperUtility {

  public OrderDTO buildOrderDTO(Long orderNumber, Double totalAmount) {
    OrderDTO order = new OrderDTO();
    order.setOrderNumber(orderNumber);
    order.setTotalAmount("$" + totalAmount);
    return order;
  }
}

public class OrderDTO {
  private Long orderNumber;
  private String totalAmount;

  // Getters and Setter
}

Repository kodumuz şöyledir. Repository belirtilen SQL cümlesini çalıştırır, ancak dönüş tipi @Entity değil de bizim kodladığımı DTO olduğu için projection devreye girer.

@Repository
public interface CustomerRepository extends JpaRepository<CustomerEntity, Integer> {
  
  @Query(name = "customerEntity.getCustomerDetails", nativeQuery = true)
  List<CustomerDetailsDTO> getCustomerDetails();
}

Örnek

Elimizde şöyle bir @Entity kodu olsun. Post nesnesi ve buna ait Tag listesi olsun

class Post {
  @Id
  Long id;
  String title;
 @ManyToMany(fetch = FetchType.LAZY,
                cascade = {
                    CascadeType.PERSIST,
                    CascadeType.MERGE,

                })
  @JoinTable(name = "post_tags",
             joinColumns = { @JoinColumn(name = "post_id") },
             inverseJoinColumns = { @JoinColumn(name = "tag_id") })

  Set<Tag> tags;
  // constructor
  // getters
  // setters
}

class Tag{
  @Id
  Long id
  String name;
  // getters, setter, constructor
}

Daha sonra projection kodunu tanımlamak için şöyle yaparız.

interface PostProjection{
  Long getId();
  String getTitle();
  @Value("#{target.tags.size()}")
  int  getNumberOfTags();
}

SpEL yerine kodla projection yapılabilir. Şöyle yaparız.

import static java.util.stream.Collectors.toSet;

public interface PostProjection {
  String getTitle();

  Set<Tag> getTags();

  default Set<String> getTagsNames() {
    return getTags().stream().map(Tag::getName).collect(toSet());
  }

}