enhancement: #15 [Spring Batch Job] local Kafka Producer(CCTV)

[yeollow]

Spring Batch Job

자세한 Spring Batch내용은 Spring Batch Docs
Spring Batch metadata Table은 Meta-Data Schema Appendix 참조

• Scope Job (CCTV)
  • ItemReader : Flat file, XML, DB등의 다양한 입력으로 부터 Data를 읽는 수단.
    • FlatFileItemReader를 통해 .csv파일을 읽음
  • ItemProcessor : ItemReader와 ItemWriter사이에서 비즈니스 로직을 수행한다
    • 해당 프로젝트에서는 구현하지 않는다.
  • ItemWriter : ItemReader와는 반대로 Data를 쓰는 수단.
    • KafkaItemWriter를 통해 ItemReader로 읽은 .csv파일의 데이터를 kafka broker로 보냄
    • KafkaItemWriter 시 kafkaTemplate에서 항상 defaultTopic을 지정해줘야 한다. kafkaTemplate에서 지정해주지 않을 시 application-local.yml에 기술되어있는 default-topic의 topic이름으로 자동으로 전송되며 이 마저도 작성해주지 않으면 topic은 생성되었을지언정, 보낼 topic을 지정하지 않아 데이터를 보낼 수 없다.

Job이나 Step은 각각 JobBuilderFactory, StepBuilderFactory를 생성자 DI를 하여 build하며 Factory Pattern은 Factory Pattern이란?을 참조하자

Kafka

kafka producer로 kafka broker의 topic에 데이터를 전송하기 위해서는 몇가지 설정이 필요하다. 자세한 내용은 Spring Kafka Docs 참조

• Topic 생성 : kafkaTemplate에서 default-topic지정이 필요하다.
  • Spring Boot에서 Application Context에 KafkaAdmin에 대한 Bean을 configuration하면 자동으로 broker에 지정한 Topic을 생성해준다.
  • Topicbuilder로 topic의 name(), partitions(), replicas() 등을 builder pattern을 통해 build하여 NewTopic Bean을 생성한다.
    • Spring Kafka 2.7 version부터는, KafkaAdmin.NewTopics를 정의하여 여러개의 NewTopic을 한번에 정의할 수 있도록 지원한다.

• Producer 설정 : application-local.yml에 spring.kafka.producer에 관련한 설정 내용을 포함시킨다.
  • bootstrap-servers : localhost:9092로 지정
  • key-serializer : LongSerializer
  • value-serializer : JsonSerializer
  • acks : 1
  • ...

bootstrap-servers 및 application-{phase}.yml, Topic의 replicas 및 partitions는 개발 환경에 따라 다름

[yeollow]

Bean Circuler Reference

KafkaTemplate에 대한 설정관련 내용을 Spring Kafka Docs를 참조하여 아래와 같이 구성

   @Bean
    public ProducerFactory<Long,CCTV> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    private Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

        return props;
    }

    @Bean
    public KafkaTemplate<Long, CCTV> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

이후 KafkaItemWriter에서 KafkaItemWriterBuilder를 통해 kafkaTemplate()관련 내용을 작성 시 매개변수로 위의 kafkaTemplate()을 전달하니 아래와 같은 Exception이 발생.

• org.springframework.beans.factory.BeanCurrentlyInCreationException
  • unresolvable circular reference가 발생 하였다는 문구 확인
• org.springframework.beans.factory.UnsatisfiedDependencyException
  • 해당 Exception은 BeanCreationException으로 Spring Bean생성 중 특정 예외가 발생하여 특정 field에 의존 주입이 되어있지 않아 Bean생성이 되지 못하는 경우 던져지는 예외.
  • 위와 같은 kafkaTemplate관련 Bean중 특정 Bean이 생성되지 않으니, 특정 Bean을 참조하는 경우 circular reference가 발생할 수 있다고 판단.
    • 실제로 Spring에서 DI시 생성자 주입을 사용하는 경우에는 circular reference가 발생할 수 있음.

kafkaTemplate에 대한 properties설정을 application-local.yml에 지정하여 Spring Boot가 application을 구동할 때 자동으로 로딩되게 설정함.

[yeollow]

local spring batch job 실행 방법

kafka가 이미 local에 다운로드 되어있으며 kafka가 다운로드 되어있는 경로에서 아래 과정을 진행한다고 가정

• bin/zookeeper-server-start.sh config/zookeeper.properties를 통해 zookeeper를 먼저 구동
• bin/kafka-server-start.sh config/server.properties를 통해 kafka server 구동
  • config/server.properties에서 advertised.listeners=PLAINTTEXT://localhost:9092, zookeeper.connect=localhost:2181 확인.
    • advertised.listeners는 producer/consumer를 위해 노출할 주소를 의미함.
• ./gradlew :batch:bootRun -Dspring.profiles.active=local를 통해 spring batch job 실행
  • .csv파일을 읽어 KafkaItemWriter를 통해 kafka의 broker로 데이터 전달
• bin/kafka-console-consumer.sh --topic CCTV --from-beginning --bootstrap-server localhost:9092명령어를 통해 broker의 전달 내용을 consumer로 읽어 확인.

이외 kafka 명령어

• bin/kafka-topics.sh --zookeeper localhost:2181 --list : topic list 확인
• bin/kafka-topics.sh --zookeeper localhost:2181 --topic topicName --delete : topicName이름의 topic을 삭제
  • config/server.properties에서 delete.topic.enable=true지정 필요
• bin/kafka-topics.sh --zookeeper localhost:2181 --topic topicName --describe : topicName이름의 topic 정보 확인\
  • local 개발환경에서는 broker가 1대밖에 없으므로, Topic생성 시 partition()과 replicas()를 모두 1로 지정했음.
  • topic의 partition 수 변경은 bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic topicName --partitions 3와 같이 topicName이름의 topic의 partition을 3개 등으로 바꿀 수 있음. 하지만, Spring Boot에서 Topic생성 시 partition()갯수를 변경하면 되는 부분.

상기 명령어 등은 운영 환경에 따라 달라질 수 있음.