개요 kafkajs를 사용해서 nestjs에서 kafka를 제어한다. nest.js에서 제안하는 방식은 microservice를 이용하는 것인데 kafkajs 공식문서 사용법을 이용하는 방식이 더 직관적이어서 이 방법을 선택하여 구현하였다. [참고1][참고2] kafkajs는 kafka streams는 지원하지 않는다. kafkajs ...

개요 class-validator를 이용하여 custom validator를 생성하고 적용한다. class-validator란? 유효성 검사 라이브러리로 클래스 기반의 객체 유효성을 간편하게 검사할 수 있다. 클래스의 프로퍼티 위에 데코레이터를 선언하여 유효성 검사 규칙을 적용한다. [참고1 - 데코레이터 목록] 설치 np...

개요 nest.js에서 passport-google-oauth20를 이용하여 Google OAuth2.0 인증을 구현한다. express-session을 이용하여 session 인증을 구현한다. 의존성 패키지 설치 npm install --save passport npm install --save passport-google-oaut...

개요 PartialValueStrategy 사용할 때 Custom Post Processor를 사용하는 경우, 키로 사용되는 필드가 Post Processing 대상이 되면 ReplaceOneBusinessKeyStrategy이 동작하지 않는다. id-strategy와 상관없이 Post Processing을 적용하기 쉽도록 메세지 키를 이용...

개요 카프카 스트림에서 string 타입으로 저장된 ID 값을 mongoDB sink connector에 저장할때 objectID 타입으로 저장하고 싶다. mongoDB sink connector의 Custom Post Processor를 만들어서 해결하였다. [참고1 - How to convert a String field to Obje...

개요 카프카 스트림(kafka streams) 데이터를 mongoDB로 보낼때 타입을 명시하고 싶을 때가 있다. 이때 json 스키마를 사용할 수 있다. 정해진 포맷에 맞춰 작성 후 데이터를 담아 보내면 mongodb가 데이터를 지정된 타입으로 인식하고 저장한다. json 스키마는 매번 스키마를 같이 보내야하기 때문에 전송되는 데이터 ...

개요 카프카 스트림의 취합된 결과를 외부 스토리지에 저장하고 싶을 때가 있다. 이때 결과가 담긴 토픽 데이터를 카프카 커넥트(kafka connect)를 이용하여 다른 스토리지에 저장한다. 카프카, mongodb, 카프카 커넥트를 도커 컴포즈로 구축한다. 카프카 커넥트 카프카 커넥트는 카프카 스트림 데이터를 다른 스토리지로 옮...

개요 토픽마다 다른 json을 받는 카프카 스트림을 생성한다. 카프카 스트림은 같은 이름의 StreamsBuilderFactoryBean Bean을 오직 하나만 허용한다. 따라서 StreamsBuilderFactoryBean에 이름을 부여하여 여러 개 생성할 수 있도록 한다. 이름이 다른 StreamsBuilderFactoryBean...

개요 생산자(procuder)와 카프카 스트림(kafka streams)이 서로 Json으로 메세지를 주고 받도록 한다[참고1]. 커스텀 Serde를 선언하고 카프카 스트림 설정의 value_serde에 등록한다. Serde란? Serde란 Serializer<>와 Deserializer<> 담긴 Wrapper...

스트림이란? 한 방향으로 흐르는 연속된 데이터 흐름[참고1[참고2] 카프카 스트림이란? 카프카 스트림은 스트림을 처리하는 라이브러리이다. 카프카 스트림은 소비자(consumer)에서 동작한다. 예를 들면, ‘블로그 포스트 조회 수’ 같은 이벤트와 같이 실시간으로 갱신되는 데이터를 처리할 수 있다. 카프카 스트림 특징 카프...