개발자가 된 핏짜

QueryDSL의 최신 버전(2021.07.22 release)인 5.0 버전을 사용했습니다. 블로그 자료들에는 outdated된 것들이 많아서 최대한 공식 문서와 github을 참조하려고 했습니다. QueryDSL의 github에는 데이터 타입 별 튜토리얼도 있습니다. 저도 언젠가는 이런 오픈소스에 기여할 수 있는 개발자가 되고 싶습니다. 제가 개발자라는 직업에 매력을 느끼는 이유기도 합니다. 

QueryDSL 도입 이유

현재 JPQL을 사용하고 있는데 3가지 문제점을 발견하였습니다.

1. 문자열(String)로 처리가 되다 보니 띄어쓰기 하나에도 오류가 날 수 있음
2. Compile 단계에서 오류 체크가 불가능함
3. Runtime 단계에서 오류를 발견할 수 있어 비효율적임

포트폴리오적으로도, 새로운 기술을 써본다는 측면으로도 QueryDSL를 사용해보고 싶었지만 위에 말한 3가지 문제를 해결할 수 있어서 QueryDSL를 도입하기로 결정했습니다. QueryDSL은 아래와 같은 장점을 가지고 있습니다.

1. QueryDSL은 모든 쿼리에 대한 내용이 함수 형태로 제공 됨
2. 위의 이유 덕분에, complie 단계에서 오류 체크(Type-check)가 가능함
3. 커스터마이징하기 쉽고 유연한 코드를 작성할 수 있음

QueryDSL의 단점은 코드 라인 수가 길어진다는 것이 있지만, 함수 형태이기 때문에 가독성이 좋고 IDE의 도움(코드 자동완성)을 받을 수 있다는 게 장점입니다. 2번 이유에 대해서 멘토님께 발표할 때 human error라고 했는데 무슨 말인지 못알아들으셔서 "오타요!" 라고 외치고 말았습니다.

build.gradle 설정하기

buildscript {
    ext {
        queryDslVersion = "5.0.0"
    }
}
plugins {
 	...
 	id "com.ewerk.gradle.plugins.querydsl" version "1.0.10"
 	...
 }
 ...
 dependencies {
 	// querydsl 추가
 	implementation "com.querydsl:querydsl-jpa:5.0.0"
 	implementation "com.querydsl:querydsl-apt:5.0.0"
     ...
 }
 // Qtype 생성 경로
 def querydslDir = "$buildDir/generated/querydsl"
 querydsl {
 	jpa = true
 	querydslSourcesDir = querydslDir
 }
 sourceSets {
 	main.java.srcDir querydslDir
 }
 compileQuerydsl{
 	options.annotationProcessorPath = configurations.querydsl
 }
 configurations {
 	compileOnly {
 		extendsFrom annotationProcessor
 	}
 	querydsl.extendsFrom compileClasspath
 }

최신 버전인 5.0.0 버전을 사용하기로 했습니다.  다른 블로그를 참조하실 때 조심하세요! 버전이 다 다릅니다. 저는 지마켓 기술 블로그를 참조했습니다. 

낮은 버전이라도 버전 숫자만 바꾸면 되겠지 하다가 엄청 해맸습니다. 기술에 대해서 완전히 이해하고 코드 쓰기로 다짐했으면서 바로 질러버리는 행동을 해서 반성합니다....

Q 클래스

QueryDSL로 쿼리를 작성할 때에는 QType을 이용해서 쿼리를 Type-safe하게 작성합니다.

사이드 바에 있는 Gradle을 클릭하면 아래처럼 윈도우가 열립니다. 

Tasks-> other -> complieQuerydsl 을 실행합니다.

Q클래스가 생성된 것을  확인 할 수 있습니다.

Configuration 설정하기

@Configuration
public class QuerydslConfig {
        @PersistenceContext
        private EntityManager entityManager;

        @Bean
        public JPAQueryFactory jpaQueryFactory() {
            return new JPAQueryFactory(entityManager);
        }

}

JPAQueryFactory를 Bean으로 등록해서 프로젝트 전역에서 QueryDSL을 작성할 수 있습니다.

 참조문서

일요일에 오랜만에 노트북을 들지 않고 외출을 했습니다. 몸은 가벼운데 마음은 무거워져서 내일은 더 열심히 해야지 다짐해봅니다.

오늘은 멘토님의 멘토링이 있는 날입니다. 

그동안 사용한 기술스택을 정리하고 회고하는 시간이었습니다.

배포를 하고 유저 피드백을 받고 있습니다. 좋은 칭찬도 많고 도움이 되는 피드백도 많이 오고 있습니다. 시간을 내어 피드백을 써주시는 분들께 감사하지만 한 피드백을 보고 삐딱한(?) 맘이 들었습니다. 아 삐딱한 이라기보다는 "개발자 빨리 할걸...."이라는 맘입니다. 

회원 가입시, abc@def.ghi로 가입이 됩니다

위와 같은 피드백이 왔을 때 든 맘은

"잉?" 

이었습니다. ghi 라는 도메인이 없다고 어떻게 장담해서 이걸 정규식을 도입을 안 했다고 말씀하시는 거지? 세상에 이메일 주소들이 무한대로 많은지언데! 

그래도 피드백이 들어왔으니 머리를 맡대고 조원들이랑 고민을 해봤는데 저희는 이미 프론트에서 정규식으로 한번, 백에서도 정규식 메서드로 한번 총 두번을 검증하고 있습니다. 골뱅이가 들어가고 알파벳과 숫자로만 이루어지는 이메일인지 확인을 하는 정규식 입니다. 백엔드에서는 아래와 같은 코드를 사용해서 유효성검사를 합니다.

public class Validator {
    public static boolean isValidEmail(String email){
        final String REGEX = "^[0-9a-zA-Z]([-_.]?[0-9a-zA-Z])*@[0-9a-zA-Z]([-_.]?[0-9a-zA-Z])*.[a-zA-Z]{2,3}$";

        return Pattern.matches(REGEX, email);
    }

그래서 정규식을 도입하지 않았다고 하는 피드백은 오해이고 이건 어쩔수 없는 문제라고 결론을 내리려다가 든 생각은,

"아, 이래서 이메일 인증을 하는구나!"

였습니다.

이메일 인증이 사용자가 입력한 이메일 값이 정확한 건지 거짓으로 입력한 건지를 알 수 있는 유일한 수단이었던 것입니다. 

아아, 안탑깝게도 저희의 프로젝트는 이제 1주일 밖에 남지 않았고 부트캠프도 2주 밖에 남지 않은 상태입니다. 당장 도입을 한다고 해도 물리적으로 시간이 가능할지 (백은 가능하지만 프론트는 계속해서 들어오는 피드백과 에러를 고치기에도 시간이 모자란 상태입니다.) 의문이 들었습니다. 그래서 제가 제안한 것은, 프로젝트 기한이 끝나고 부트캠프가 끝나도 계속해서 만나서 프로젝트의 완성도를 높혀나가자 였습니다. 고맙게도 열정과 욕심히 그득그득한 조원들 대부분이 찬성해주었습니다. 다들 6주라는 시간으로 인한 서비스의 완성도에 아쉬움을 가지고 있었거든요. 부트캠프 끝나고도 계속 할일이 생겼네요.

오늘 포스팅의 제목은 '모든 건 개발자 마음대로'인데요, 정말 개발자의 경력과 능력치에 따라서 서비스가 얼마나 완성도가 있어지는지를 새삼 더욱 더 많이~ 깨닫고 있는 요즘입니다. 이메일 인증같은 다소 사소해보이는 이 기능도 개발자가 생각해냈을테니까요. 저는 경영학과 출신이라 기획만 하고 아이디어를 생각만 하고 펀드만 끌어오지 구체적인 구현을 할 줄 몰라서 답답해 했었는데 이제는 '타이탄의 도구' 하나를 획득했습니다. 전세계 갑부들을 보면 일론 머스크, 마크 주커버그 등 대부분 본인이 프로그래밍을 직접할 줄 알아서 아이디어 구현까지 할 수 있었던 엔지니어들이 대부분입니다. 물론 워런버핏도 있지만, 워런버핏이 전세계 사람들의 삶에 영향을 끼칠만한 지대한 기여는 하지 않았으니까요. 

시간이 갈수록 엔지니어를 왜 진작하지 않았을까 하는 아쉬움이 밀려들지만, 앞으로 90년은 더 살아야하는 21세기 인간이기에 지금이라도 열심히 하려고 합니다. 

드디어 배포를 했습니다.

피드백이 들어왔는데 

데이터 최적화는 어떻게 하는 걸까요?

ㅠㅠ

SSE(Server-Sent-Event)를 사용해서 알림기능을 만들기로 결정을 하고 약 2주동안 고생하면서 결국에는 성공(?)을 시켰습니다. 완전하다고는 할 수 없지만 이제는 왠만하면 SSE를 사용하지는 않고 다른 옵션을 시도해볼 것 같습니다. 로컬에서는 잘 돌아갔는데 프론트랑 연결하면서 팡팡팡팡팡! 에러가 하루가 멀다하고 터졌거든요. 산넘어 산이었습니다. 이번 포스팅에서는 다른 블로그들은 언급하지 않고 넘어가는 SSE 알림 기능 관련 에러 총정리를 하려고 합니다. 저처럼 헤매지 마시길.

SSE 알림 기능 관련 주요 TIL

230110 알림 기능 구현 뭘로 할까?

230111 SSE로 알림기능 구현하기

230120 SSE 프론트와 연결하기 (feat. 비관적락)

230124 SSE 에러: DB connection leak, open-in-view 설정

230125 SSE 또 connection leak triggered

230126 @Async 비동기처리

230126 읽지 않은 알림 갯수 반환하기

230127 @EventListener 알림 기능 강결합 제거

230131 AsyncConfigurerSupport로 비동기 설정하기

위 처럼 다양한 에러를 만나고 결국엔 해결했습니다. TIL에서 언급하지 않은 부분과 그래서 결국 최종 코드가 무엇인지에 대해 알려드리도록 하겠습니다. 하나 걸러 하나가 나오는 에러라 다 기록하지는 못했지만 앞으로는 더 꼼꼼히 기록해야겠다는 다짐을 하면서 우선 SSE 설정을 위한 헤더에 대해 알아봅시다.

NotificationController

@Tag(name = "SSE")
@ApiResponses(value = {
        @ApiResponse(responseCode = "2000", description = "SSE 연결 성공"),
        @ApiResponse(responseCode = "5000", description = "SSE 연결 실패")
})
@Operation(summary = "SSE 연결")
@GetMapping(value="/api/subscribe/{nickname}", produces = "text/event-stream")
public SseEmitter subscribe(
        @PathVariable String nickname,
        @RequestHeader(value="Last-Event-ID", required = false, defaultValue = "") String lastEventId,
        HttpServletResponse response){
    response.addHeader("X-Accel-Buffering", "no");
    response.addHeader("Content-Type", "text/event-stream");
    response.setHeader("Connection", "keep-alive");
    response.setHeader("Cache-Control", "no-cache");

    String encodedNickname = URLDecoder.decode(nickname, StandardCharsets.UTF_8);
    return notificationService.subscribe(lastEventId, encodedNickname);
}

Ngnix를 적용하고 에러가 나서,

response.addHeader("X-Accel-Buffering", "no");

를 적용해 주었고

아래와 같이 에러가 나서

EventSource's response has a Content-type specifying an unsupported type: application/json. Aborting the connection

response.addHeader("Content-Type", "text/event-stream");

를 적용해주었습니다.

ngnix 세팅 때문에 토큰을 헤더에 담아오지 못하게 되어서 nickname으로 파라미터를 변경해주었는데, 닉네임에 한글도 있어서 encoding관련해서 아래처럼 설정을 넣어주었습니다.

String encodedNickname = URLDecoder.decode(nickname, StandardCharsets.UTF_8);

NotificationService

    public SseEmitter subscribe(String lastEventId, String nickname) {
        Member member = memberRepository.findByNickname(nickname)
                .orElseThrow(()-> new NotFoundException(SSE, SERVICE, MEMBER_NOT_FOUND, "Nickname : " + nickname));
        Long memberId = member.getId();
        String emitterId = memberId + "_" + System.currentTimeMillis();
        SseEmitter emitter = emitterRepository.save(emitterId, new SseEmitter(DEFAULT_TIMEOUT));

        log.info("emitter created");

        emitter.onCompletion(() -> {
            synchronized (emitter){
            emitterRepository.deleteById(emitterId);}});
        emitter.onTimeout(() -> {
            emitter.complete();
            emitterRepository.deleteById(emitterId);});

        sendToClient(emitter, emitterId, "EventStream Created. [memberId=" + memberId + "]");

        if (!lastEventId.isEmpty()) {
            Map<String, Object> events = emitterRepository.findAllEventCacheStartWithByMemberId(String.valueOf(memberId));
            events.entrySet().stream()
                    .filter(entry -> lastEventId.compareTo(entry.getKey()) < 0)
                    .forEach(entry -> sendToClient(emitter, entry.getKey(), entry.getValue()));
        }

        return emitter;
    }

    @Async // 비동기 처리를 위한 어노테이션
    @Transactional
    public void send(Member receiver, Member sender, NotificationType notificationType, String content, RedirectionType type, Long typeId, Long postId) {
        Notification notification = notificationRepository.save(new Notification(receiver, notificationType, content, type, typeId, postId, sender));
        String memberId = String.valueOf(receiver.getId());

        Map<String, SseEmitter> sseEmitters = emitterRepository.findAllEmitterStartWithByMemberId(memberId);
        sseEmitters.forEach(
                (key, emitter) -> {
                    emitterRepository.saveEventCache(key, notification);
                    sendToClient(emitter, key, notification.getContent());
                }
        );
    }

    private void sendToClient(SseEmitter emitter, String emitterId, Object data) {
        try {
            log.warn("emitterId : " + emitterId);
            log.warn("data : " + data.toString());
            emitter.send(SseEmitter.event()
                    .id(emitterId)
                    .data(data));
            log.info(emitterId+"-emitter has been sent and completed");
        } catch (IOException exception) {
            log.error("Unable to emit");
            emitter.completeWithError(exception);
            emitterRepository.deleteById(emitterId);
        }
    }

    @Transactional
    public List<ResponseNotificationDto> getNotificationList(Member member) {
        List<Notification> notificationList = notificationRepository.findAllByReceiverOrderByCreatedAtDesc(member);
        List<ResponseNotificationDto> responseNotificationDtoList= new ArrayList<>();
        for (Notification notification: notificationList) {
            responseNotificationDtoList.add(SSE_MAPPER.NotificationtoResponseNotificationDto(notification));
        }
        return responseNotificationDtoList;
    }

    @Transactional
    public void readNotification(Long notificationid, Member member) {
        Notification notification = notificationRepository.findById(notificationid)
                .orElseThrow(()-> new NotFoundException(SSE, SERVICE, NOTIFICATION_NOT_FOUND, "Notification ID : " + notificationid));
        if(member.getId().equals(notification.getReceiver().getId())) {
            notification.read();
        }
        notificationRepository.save(notification);
    }

    @Transactional
    public ResponseCountNotificationDto countUnreadNotifications(Member member) {
        String nickname = member.getNickname();
        Long count = notificationRepository.countUnreadNotifications(nickname);

        return new ResponseCountNotificationDto(count);

    }
}

여러가지 메소드가 들어있어서 좀 깁니다. 

중요한 수정사항들을 몇가지 언급하자면,

synchornized() : 비동기를 사용했던 친구들이 완료되면 동기화 시키는 메서드

emitter.completeWithError(exception);

에미터가 발신 되지 않았을 경우 에러와 함께 종료 시키기

대댓글 알림 서비스 메서드 예시

제가 EventListener를 사용했다는 사실 알고 계시죠? (@EventListener 사용)  

아래와 같이 EventListener를 구현하였습니다.

@Component
@RequiredArgsConstructor
@Slf4j
public class NotificationListener {

    private final NotificationService notificationService;

    @TransactionalEventListener
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    @Async
    public void handleNotification(RequestNotificationDto requestNotificationDto){
        notificationService.send(requestNotificationDto.getReceiver(), requestNotificationDto.getSender(),requestNotificationDto.getNotificationType(),
                requestNotificationDto.getContent(), requestNotificationDto.getType(), requestNotificationDto.getTypeId(), requestNotificationDto.getPostId());
        log.info("EventListener has been operated. Sender Id: " + requestNotificationDto.getSender().getId() + " NotificationType: " +requestNotificationDto.getNotificationType());
    }

}

예시로는 대댓글을 작성하면 가는 알림입니다.

@Transactional
public void createComment(Long postId,Long parentId, CommentDto commentDto, String nickname) {

    Post post = postRepository.findById(postId).orElseThrow(
            () -> new NotFoundException(Domain.COMMENT, Layer.SERVICE, POST_NOT_FOUND, "Post ID : " + postId)
    );
    Comment parentComment = null;
    if (parentId != null) {
        parentComment = commentRepository.findById(parentId).orElseThrow(
                () -> new NotFoundException(Domain.COMMENT, Layer.SERVICE, COMMENT_NOT_FOUND, "Parent Comment ID : " + parentId)
        );
    }
    Member member = memberRepository.findByNickname(nickname)
            .orElseThrow(() -> new NotFoundException(COMMENT, SERVICE, MEMBER_NOT_FOUND, "Nickname : " + nickname));

    Comment comment = COMMENT_MAPPER.commentDtoToComment(commentDto, member, post, parentId);
    commentRepository.save(comment);

    //post 작성자에게 댓글 알림
    Member postMember = memberRepository.findByNickname(post.getNickname())
            .orElseThrow(() -> new NotFoundException(COMMENT, SERVICE, MEMBER_NOT_FOUND, "Nickname : " + nickname));

    if (!postMember.getNickname().equals(member.getNickname())) {
        String content = post.getTitle() + "에 " + nickname + "님이 댓글을 남겼습니다.";
        notify(postMember, member, NotificationType.COMMENT, content, RedirectionType.detail, postId, null);
    }

    //댓글 작성자에게 댓글 알림
    if (parentComment != null) {
        Member commentMember = memberRepository.findByNickname(parentComment.getNickname())
                .orElseThrow(() -> new NotFoundException(COMMENT, SERVICE, MEMBER_NOT_FOUND, "Nickname : " + comment.getNickname()));
        if (!commentMember.getNickname().equals(member.getNickname())) {
            String content = commentMember.getNickname() + "님의 댓글에 " + nickname + "님이 댓글을 남겼습니다.";
            notify(commentMember, member, NotificationType.COMMENT, content, RedirectionType.detail, postId, null);
        }
    }
}


private void notify(Member postMember, Member sender, NotificationType notificationType,
                    String content, RedirectionType type, Long typeId, Long postId){
    eventPublisher.publishEvent(new RequestNotificationDto(postMember,sender, notificationType,content,type, typeId, postId));

}

위처럼 notify 라는 메서드를 만들어서 event를 발행합니다. 해당 event가 발행되면 listener가 듣고 NotificationService의 send 메서드를 실행시켜줍니다. 이때, Eventlistener의 파라미터는 1개이어야 하기 때문에 RequestNotificationDto를 생성해서 하나의 dto에 정보를 담아서 보내주었습니다.

SSE 에러를 해결하면서 느낀 점

인터넷에 나오는 모든 정보가 정확한 것은 아닙니다. 에러가 있어도 언급을 하지 않는 경우가 있고, 내 컴퓨터에서는 안되는 경우가 더러 있습니다. 그리고 동료들이 있어서 함께 에러를 해결해나갈수 있었습니다. github의 다른 개발자들도 같은 고민을 한 적이 있고 각기 다른 방법으로 에러를 해결했다는 사실도 재미있습니다. 

에러는 로컬에서 발생하는 에러(error factor : ONLY myself), 프론트와 연결하면서 발생하는 에러(error factor: still ONLY ME)가 있습니다. ngnix 설정을 하면서 또 한번 다수의 에러가 생겼고, 강결합으로 인한 에러도 있었습니다. 이번을 계기로 http 통신에 대해서 좀 더 알게되었습니다. 

아직도 미해결 에러가 있습니다. 해협!!!

많은 시간 공들여서 찾아보았지만 아직 해결되지 않은 에러가 한가지 남아있습니다.

바로바로.....

net:: ERR_INCOMPLETE_CHUNKED_ENCODING 200

입니다.

구글링으로 나와있는 방법은 대부분 시도해보았습니다. 혹시 해결방안을 알고 계시는 분들은 댓글달아주세요...!!

라고 하자마자 해결방안이 나와서 에러가 이제 안납니다.

ngnix 설정을 추가해 줍니다.

오늘은 계획된 배포일이지만 아무도 오늘 가능할 거라고 생각하지 못합니다.

일단은 @Async 의 Configuration 설정을 통해서 서버가 터지는 현상은 막았습니다. 

AsyncConfigurerSupport 

@Configuration
@EnableAsync
public class AsyncConfig extends AsyncConfigurerSupport {

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(30);
        executor.setQueueCapacity(50);
        executor.setThreadNamePrefix("NOTIFICATION-ASYNC");
        executor.initialize();
        return executor;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return super.getAsyncUncaughtExceptionHandler();
    }
}

위와 같이 설정해주었는데 나중에 Chatgpt한테 물어보니 코드를 그대로 짜주더라고요. 살짝 개발자 커리어에 대한 불안감이 엄습해왔지만 당장 닥친 미래가 더 중요해서 집중을 했습니다. 무서운 세상입니다... 취업하면 MS 주식 풀매수... 

에러 넘어 에러!! ERROR!!

대댓글에는 알림이 가지 않는 현상을 발견했습니다. 생각을 깊게 못한 제 탓인가요. 기능이 하루걸러 추가되는 프로젝트였다 보니 그런거 같기도 합니다. 

에러가 마구마구 터지는 중.

해결이 안되는데

일단 새벽 3시가 넘어서 잠.

어느새 최종프로젝트도 배포일이 2일 남았습니다. MVP 기능은 예전에 끝이 났는데 프론트와 연결하고 배포하면서 일어나는 자잘한 에러들과 디자인 변경으로 인한 새로운 api들 여러개를 구현하다보니 시간이 훌쩍 지나갔습니다. SSE 파트를 맡은 저는 하루 걸러 새로 나오는 버그들에 정신이 없었던거 같습니다. SSE 관련해서 타임라인을 보면 아래와 같습니다. 관련해서 잘 정리해놓고 면접때 꼭 얘기하라는 멘토님의 말씀이 있기도 했습니다. 

덕분에 비동기, 동기 관련 개념을 알게 되었고 @EventListener라는 어노테이션을 활용할 수 있게 되었으며 Gson이라는 구글의 라이브러라리를 알게 되었습니다. 

팀원들과 소통하며 각자의 파트에서 배운 것들을 공유했습니다. 또 facade 패턴을 도입하는 것도 고려 했었는데 멘토님이 SAGA 패턴이라는 것을 알려주셔서 패턴에 대해서도 공부를 했습니다. 하루하루 배울 것이 천지이고 알게 될수록 개발은 흥미롭기만 합니다.

(영어 2배속으로 들었는데 한번 더 들어야 이해가 될 듯 합니다. 영어는 영어이지만 외계어인듯.... Speak English Ma'am!!)

영어를 아는 것과 프로그래밍 언어를 아는 것은 다르고 컴퓨터 용어를 아는 것도 다릅니다... 배울게 넘  많아요!!

이제 2주 남짓한 시간이 남았는데 최종 프로젝트 끝나면 알고리즘 공부를 빡세게 해야겠습니다.... 마지막까지 화이팅🙏🙏🙏🙏

에러

EventSource's response has a Content-Type specifying an unsupported type;

이라는 에러가 발생했습니다. 프론트로부터 위와 같은 에러를 전달 받았습니다.

이벤트 데이터를 저장할 때 원본 객체를 문자열로 변환하지 않고 그대로 저장할 경우 EventSteam 연결이 즉시 종료되는 현상이 확인되었습니다. 그래서 여기저기 물어보고 구글링 해서 json 객체를 문자열로 변환하는 방법을 찾았습니다.

json 마샬링 언마샬링 하기에 대해서 아래글을 참조하세요.

객체직렬화 하기

@Async // 비동기 처리를 위한 어노테이션
@Transactional
public void send(Member receiver, Member sender, NotificationType notificationType, String content, RedirectionType type, Long typeId, Long postId) {
    Notification notification = notificationRepository.save(new Notification(receiver, notificationType, content, type, typeId, postId, sender));
    String memberId = String.valueOf(receiver.getId());

    Map<String, SseEmitter> sseEmitters = emitterRepository.findAllEmitterStartWithByMemberId(memberId);
    sseEmitters.forEach(
            (key, emitter) -> {
                emitterRepository.saveEventCache(key, notification);
                Gson gson = new Gson();
                String data = gson.toJson(notification);
                sendToClient(emitter, key, data);
            }
    );
}

Gson을 사용해서 객체직렬화를 하였습니다.

Gson gson = new Gson();
String data = gson.toJson(notification);

아직 프론트랑 연결은 안해봤는데 일단 string 형태로 보내니까 json 형태라는 에러는 안뜰거라고 예상합니다.

참고 자료