이번 글에서는 Kafka에서 데이터를 읽은 뒤 JSON 파일로 저장하던 Java 코드를 Elastic Search에 저장하도록 수정하자
지난 글에서는 Spring data elasticsearch 패키지 core 의존성을 찾을 수 없는 문제로 임시로 JSON 파일로 저장하도록 했다.
Cursor에게 계속해서 물어보고 수정하고 반복한 결과 데이터 전송 프로세스는 아래와 같다.
1.ElasticsearchOperations 주입
Spring Data ES 의 핵심 인터페이스로 ES와의 상호작용을 처리한다.
2. Kafka Listener을 통한 메세지 수신
listen 메서드로 topic에서 메세지를 수신한다.
3. 데이터 처리 및 문서 생성
정류장 이름 추가, 정류장 경유 여부 등 데이터를 가공하고 타임 스탬프, 고유 ID를 추가하여 ES에 저장하기 위한
문서 형식으로 새로 생성한다.
4. ES로 데이터 전송
IndexQueryBuilder() 메서드를 통해 현재 날짜 기반으로 인덱스를 생성하고 ElasticsearchOperations을 통해
문서를 ES에 저장한다.
ES에 잘 저장 되는지 로컬에서 Docker 컨테이너로 ES를 띄워서 테스트해보자
컨테이너를 생성할 때 인증서를 사용하지 않는 HTTP 요청으로 진행하기 위해 아래와 같이 환경변수 옵션을 줘서 실행한다.
위와 같이 Elastic Search Image로 컨테이너를 실행하고 자바 모듈을 실행한 후 ES 인덱스 조회 curl을 날려보면 인덱스 count가 증가한 것을 볼 수 있다.
curl -X GET "http://localhost:9200/_cat/indices/bus-location-2025-08-31?v"
curl -X GET "http://localhost:9200/bus-location-2025-08-31/_search?pretty"
현재는 로컬 테스트로 ssl 인증서 없이 통신을 진행했지만 처음 Java -> ES 연결을 시도할 때 인증서 관련 에러를 마주했다.
서비스에 사용되는 모듈은 모두 AWS VPC private subnet 환경에서 실행할 계획이여서 https 통신이 필요한지는 잘 모르겠다.
서버 간 통신에 대해 인증서 관련 정보를 찾아볼 필요가 있다.