개요 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)에서 동작한다. 예를 들면, ‘블로그 포스트 조회 수’ 같은 이벤트와 같이 실시간으로 갱신되는 데이터를 처리할 수 있다. 카프카 스트림 특징 카프...

개요 비동기 함수 안에서 MultipartFile.getInputStream() 실행 시 임시 파일이 없다는 에러가 난다. 원인 컨트롤러에서 multipart/form-data 요청을 받으면 해당 파일이 임시 경로에 저장되는 듯 하다. 이때 MultipartFile.getInputStream() 실행하면 임시 파일에서 stream ...

개요 카프카 생산자(producer) value serializer를 JsonSerializer로 설정한다. 카프카 소비자(consumer) value deserializer를 JsonDeserializer로 설정한다. 소비자 그룹마다 다른 json을 message로 받을 수 있도록 한다. 환경 java 21 spring ...

개요 spring security 6을 이용하여 oauth2.0 인증을 구현해본다. 예시에서는 oauth 2.0 인증 프로바이더로 google을 사용한다. Oauth 2.0 이란 표준화된 인증 방식을 의미한다. 한 곳에서 Ouath 인증을 하면 이 인증을 공유하는 애플리케이션끼리 별도의 인증 없이 인증할 수 있다[참고1]. 일...