[bug] 종목 종가 갱신 문제

[yonghwankim-dev]

상황

• redis 저장소에 종가 가격이 만료되어 없어졌는데도 불구하고 5초마다 가격 갱신시 종가 가격이 없다면 종가 갱신이 되어야 하는데 이 부분이 안됩니다.

• 종가 가격이 redis에 저장되지 않은 상태에서 예외가 발생하게 되면 sse 요청시 스프링 서버에서 더이상 처리를 하지 않고 끊어야 하는데 계속 스케줄링 테스크가 작동되는 문제가 발생합니다.

원인

SSE task 유지 문제

sse task를 스케줄러 쓰레드풀에 넣고 실행할 때 예외가 발생했음에도 불구하고 테스크를 계속 반복 수행한다.

쓰레드풀로부터 ScheduledFuture<?> future 객체를 받아서 future.get() 메소드를 호출하여 대기한다. 그러다가 future가 타임아웃 발생시 future.cancel(true)를 호출하여 해당 테스크를 취소할 수는 있지만, get 메소드를 통해서 결과를 대기 받게 되면 정상적인 케이스의 경우 한번 결과를 받고 sse 연결을 종료하게 됩니다. 이렇게 되면 sseExecutor.scheduleAtFixedRate 메소드를 사용할 이유가 없게 된다.

한국투자증권 accessToken 값 저장 문제

redis 저장소의 종가 가격이 저장되지 않은 이유는 redis 저장소에 한국 투자 증권 api 서버에 대한 accessToken이 저장되지 않아서입니다.

서버를 처음 실행할 때 한국투자 증권 api 서버로부터 accessToken 발급시 정상적으로 24시간 후의 만료시간을 가진 accessToken을 발급받습니다. 그러나 발급받은 토큰을 폐기하지 않고 계속 accessToken만을 발급하게 된다면 이전에 발급받은 토큰을 발급받게 됩니다.

예를 들어 배포 서버에서 spring과 redis를 실행(2023-12-22 15시 실행)하여 새로운 accessToken을 발급받았지만 "access_token_token_expired" 프로퍼티는 2023-12-23 15:00:00이 아닌 2023-12-23 14:08:26 시간임을 볼 수 있습니다. 이는 한국투자증권 정책에 따라서 일반개인고객이 Acess Token 발급을 일정시간(6시간) 이내에 재 호출시에는 직전 토큰값을 리턴하기 때문입니다.

코드상의 문제는 무엇인가?
서버 실행시 redis에 accessToken이 저장되지 않아서 발급을 요청시 이전에 발급받은 accessToken을 발급(만료시간이 서버 실행시 24시간 후가 아닌 예전 만료시간이 됨)받을 수 있습니다. 이렇게 되면 KisAccessTokenManager 객체의 refreshAccessToken 메소드 실행시 다시 초기화되는 expirationDatetime 필드 멤버와 발급받은 accessToken의 만료시간이 달라질수 있습니다.

public class KisAccessTokenManager {
	private String accessToken;
	private String tokenType;
	private LocalDateTime expirationDatetime;

        // ...
	public void refreshAccessToken(Map<String, Object> accessTokenMap) {
		this.accessToken = (String)accessTokenMap.get("access_token");
		this.tokenType = (String)accessTokenMap.get("token_type");
		this.expirationDatetime = LocalDateTime.now().plusSeconds((int)accessTokenMap.get("expires_in"));
	}
}

예를 들어 메소드에 입력으로 들어온 accessMap이 다음과 같다고 가정합니다.

{
    "access_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzUxMiJ9.eyJzdWIiOiJ0b2tlbiIsImF1ZCI6ImNkOTAzNzY2LWNlNGUtNDA4MC1hNGU4LTYyYWYyODU0MWFkOCIsImlzcyI6InVub2d3IiwiZXhwIjoxNzAzMzA4MTA2LCJpYXQiOjE3MDMyMjE3MDYsImp0aSI6IlBTRGc4WlVJd041eVl5ZkR6bnA0TDM2Z2xhRUpic2RJNGd6biJ9.31rzd3QItxqki1R0DPL54-mS7ER0rBXRXH_p6uKyXAs-hjEKUZLhhUlYpIW68vIVppCiCjMU8yyXPTU2hqKTlQ",
    "access_token_token_expired": "2023-12-23 14:08:26",
    "token_type": "Bearer",
    "expires_in": 86400
}

그러나 LocalDateTime.now() 메소드 실행시 나온 값이 2023-12-22 15:00:00라면 expirationDatetime 필드 멤버에 초기화되는 값은 86400초를 더한 2023-12-23 15:00:00가 됩니다. 그렇게 되면 실제로는 redis 저장소에 있는 accessToken은 만료시간이 되어 삭제되는 반면 aop 메소드에서 토큰이 만료되었는지 객체에서 검사하는 메소드에서는 만료되지 않았다고 하여 종가 갱신이나 현재가 갱신을 한국투자증권 서버에 요청하다가 토큰 만료로 실패하는 것입니다.

CompletableFuture 객체의 join 무한대기 문제

해당 문제가 발생한 핵심적인 원인입니다. 문제가 발생한 원인은 코드 구현 상에서 CompletableFuture 객체의 타임아웃이 발생했을때의 콜백과 종목 현재가를 요청했을때 예외가 발생한 경우에 예외처리를 하지 않아서입니다.

우선 5초에 한번씩 실행되는 주식 현재가를 갱신하는 메소드의 내용은 다음과 같습니다.

	public void refreshStockCurrentPrice(List<String> tickerSymbols) {
		List<CompletableFuture<CurrentPriceResponse>> futures = tickerSymbols.parallelStream()
			.map(tickerSymbol -> {
				CompletableFuture<CurrentPriceResponse> future = new CompletableFuture<>();
				executorService.schedule(createCurrentPriceRequest(tickerSymbol, future), 200, TimeUnit.MILLISECONDS);
				return future;
			}).collect(Collectors.toList());

		futures.parallelStream()
			.map(CompletableFuture::join)
			.filter(Objects::nonNull)
			.forEach(currentPriceManager::addCurrentPrice);
	}

위 코드에서 CompletableFuture 객체(future)를 생성한 다음 createCurrentPriceRequest 메소드에 매개변수로 전달하는 것을 볼 수 있습니다.

createCurrentPriceRequest 메소드의 내용은 다음과 같습니다.

	private Runnable createCurrentPriceRequest(final String tickerSymbol,
		CompletableFuture<CurrentPriceResponse> future) {
		return () -> {
			CurrentPriceResponse response = readRealTimeCurrentPrice(tickerSymbol);
			future.completeOnTimeout(response, 10, TimeUnit.SECONDS);
			future.exceptionally(e -> {
				log.info(e.getMessage(), e);
				return null;
			});
		};
	}

createCurrentPriceRequest 메소드의 내용은 종목 현재가를 한국투자증권 서버로부터 요청하고 받아오는 Runnable 타입의 테스크를 만들어내는 메소드입니다.

그러나 위 코드에서 문제점은 readRealTimeCurrentPrice 메소드에서 예외가 발생할 경우에 readRealTimeCurrentPrice 메소드 이후의 코드(future.completeOnTimeout, future.exceptionally)는 수행되지 않고 Runnable 테스크는 종료됩니다. readRealTimeCurrentPrice 메소드의 내용은 KisClient 객체를 이용하여 한국투자증권 서버로부터 종목 현재가를 가져오는데 간간히 서버로부터 거래건수 요청 초과 응답을 받게 되면 메소드 내에서 NullPointerException이 발생하게 됩니다. 그 내용은 다음과 같습니다.

public long readRealTimeCurrentPrice(String tickerSymbol, String authorization) {
		MultiValueMap<String, String> headerMap = new LinkedMultiValueMap<>();
		headerMap.add("authorization", authorization);
		headerMap.add("appkey", appkey);
		headerMap.add("appsecret", secretkey);
		headerMap.add("tr_id", "FHKST01010100");

		MultiValueMap<String, String> queryParamMap = new LinkedMultiValueMap<>();
		queryParamMap.add("fid_cond_mrkt_div_code", "J");
		queryParamMap.add("fid_input_iscd", tickerSymbol);

		Map<String, Object> responseMap = getPerform(currentPrice, headerMap, queryParamMap);
		Map<String, String> output = (Map<String, String>)responseMap.get("output");
		return Long.parseLong(output.get("stck_prpr"));
	}

한국투자 증권 서버로부터 초당 거래건수 초과 응답을 받게 되면 responseMap은 다음과 같이 반환받습니다.

{"rt_cd":"1","msg_cd":"EGW00201","msg1":"초당 거래건수를 초과하였습니다."}

위와 같이 responseMap에 저장되면 responseMap.get("output") 호출시 null을 반환하게 되며 output Map은 get 호출시 NullPointerException을 일으킵니다.

위 테스크는 Runnable 테스크 이므로 NullPointerException이 발생하게 되면 그 이후의 코드는 실행되지 않고 종료됩니다.

	private Runnable createCurrentPriceRequest(final String tickerSymbol,
		CompletableFuture<CurrentPriceResponse> future) {
		return () -> {
			CurrentPriceResponse response = readRealTimeCurrentPrice(tickerSymbol);
			future.completeOnTimeout(response, 10, TimeUnit.SECONDS);
			future.exceptionally(e -> {
				log.info(e.getMessage(), e);
				return null;
			});
		};
	}

• readRealTimeCurrentPrice() 메소드에서 NullPointerException이 발생하면 별도의 예외처리가 없으므로 테스크가 죽습니다.
• 이후의 future의 설정은 작동하지 않습니다.

Runnable 테스크가 예외가 발생하고 죽게되면 죽은 테스크의 future는 complete 결과도 받지 못하게 되어 join 메소드를 호출하고 계속 대기하게 됩니다.

futures.parallelStream()
	.map(CompletableFuture::join)
	.filter(Objects::nonNull)
	.forEach(currentPriceManager::addCurrentPrice);

• CompleteableFuture::join 람다를 수행하며 별도의 타임아웃이 없으므로 무한대기하게 됩니다.
• 한건의 예외가 발생하기라도 하면 해당 메시지브로커는 계속 대기하게 된다.

위와 같은 이유로 무한 대기 하게 되면 redis에 저장된 액세스 토큰과 종가 데이터는 하루 내에 만료되어 문제 배경의 사진과 같이 currentPrice 데이터만 남게 되는 것입니다. (currentPrice는 별도의 만료시간을 설정하지 않아서 계속 남게됩니다.)

해결방법

expirationDatetime 갱신 부분을 수정하여 만료시간을 redis 저장소의 것과 동일하게 맞추기

우선은 KisAccessTokenManager 객체의 refreshAccessToken 메소드 실행시 expirationDatetime을 갱신하는 부분을 수정합니다.

	public void refreshAccessToken(Map<String, Object> accessTokenMap) {
		this.accessToken = (String)accessTokenMap.get("access_token");
		this.tokenType = (String)accessTokenMap.get("token_type");
		this.expirationDatetime = LocalDateTime.parse((String)accessTokenMap.get("access_token_token_expired"),
			formatter);
	}

CompletableFuture 객체의 타임아웃시 반환값 설정 및 예외 처리

CompletableFuture 객체 생성시 타임아웃, 예외 발생시의 콜백을 등록합니다.

	public void refreshStockCurrentPrice(List<String> tickerSymbols) {
		List<CompletableFuture<CurrentPriceResponse>> futures = tickerSymbols.parallelStream()
			.map(tickerSymbol -> {
				CompletableFuture<CurrentPriceResponse> future = new CompletableFuture<>();
				future.completeOnTimeout(null, 10, TimeUnit.SECONDS);
				future.exceptionally(e -> {
					log.info(e.getMessage(), e);
					return null;
				});
				executorService.schedule(createCurrentPriceRequest(tickerSymbol, future), 1L, TimeUnit.MILLISECONDS);
				return future;
			}).collect(Collectors.toList());

		futures.parallelStream()
			.map(future -> {
				try {
					return future.get(10L, TimeUnit.SECONDS);
				} catch (InterruptedException | ExecutionException | TimeoutException e) {
					return null;
				}
			})
			.filter(Objects::nonNull)
			.forEach(currentPriceManager::addCurrentPrice);
	}

Runnable 테스크에서 예외가 발생하면 �CompleteFuture 객체의 예외 처리합니다.

	private Runnable createCurrentPriceRequest(final String tickerSymbol,
		CompletableFuture<CurrentPriceResponse> future) {
		return () -> {
			try {
				future.complete(readRealTimeCurrentPrice(tickerSymbol));
			} catch (KisException e) {
				future.completeExceptionally(e);
			}
		};
	}