밴쿠버 코로노의 개발일지

2022.02.04 - [Java/Spring boot] - [Spring boot] Kafka 설정하기

Kafka 의 설정을 마무리하고 나면 이제 Kafka로 Message를 보낼 수 있게 된다. 이 포스트에서는 Kafka Topic으로 Message를 Publish 하는 예제를 공유하고자 한다. 

Kafka에서 메세지를 생산하고 publish 하는 주체를 Producer라고 한다. 나의 Project에서 auth service는 JwtToken을 만들고 이 값을 redis에 저장하고 token을 확인할 때마다 redis에서 값을 가져와서 비교하고자 한다.

Kafka 설정시에 2가지의 시나리오가 있어서 KafkaTemplate과 ReplyingKafkaTemplate을 설정하였다. 이 2가지에 대한 Message Publish를 알아보도록 하자. 

KafkaTempate & ReplyingKafkaTemplate

KafkaTemplate 을 이용해서 메세지를 보낼 때 primitive한 type으로 보낼 수 있지만, 각 서비스에서 데이터를 관리하기 위해서는 보통 Obejct로 많이 보내게 된다. 특정 Object 형태로 KafkaTemplate을 설정해서 사용할 수 있지만 메세지의 종류가 많아질 경우 이 설정 또한 방대해져서 관리가 힘들 수도 있다. 

그래서 난 범용 Object type을 Json을 Serialilze를 하여 Consumer 측에서 해당 Message로 Object Mapping을 하여 사용할 수 있도록 하였다. 이렇게 할 경우 공용 설정을 통해서 설정 부분의 코드를 줄일 수 있다. 하지만, Cosumer쪽에서 Message에 따라 Casting을 해줘야 하는 불편함은 존재한다. 선택의 문제니 필요에 따라 사용하면 될 것 같다. 

예제를 위해서는 몇가지 파일이 필요하다. 

1. Topic : 사용하는 Topic 값을 정의

    (이 Topic의 경우 사용하는 Producer와 Consumer가 함께 가지고 있어야 하는데 어떻게 하면 잘 관리할 수 있을지 고민중이다. )

public class JwtTokenTopic {
    public static final String TOPIC_CREATE_JWT_TOKEN = "topic.create.jwt.token";
    public static final String TOPIC_REQUEST_JWT_TOKEN = "topic.request.jwt.token";
    public static final String TOPIC_REPLY_JWT_TOKEN = "topic.reply.jwt.token";
}

2. Message Object : 메세지로 사용되는 Object들이다, Publish에 사용되는 2가지 간단한 메세지와 응답으로 받는 값을 Mapping 해줄 Object를 정의한다.

@Getter
@Setter
@Accessors(chain = true)
@AllArgsConstructor
@NoArgsConstructor
@ToString
@Schema(description = "JwtToken Cache Creation Message")
public static class CreateMessage {
    @Schema(example = "1L")
    private Long accountId;

    @Schema(example = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.....")
    private String jwtToken;

    @Schema(example = "10L")
    private Long expiration = -1L;

    public static CreateMessage from(Account account, Date expireDate) {
        Long expiration = -1L;

        if (expireDate != null) {
            expiration = DateTimeUtils.getSecondsBetweenDates(new Date(), expireDate);
        }

        return new CreateMessage()
                .setAccountId(account.getId())
                .setJwtToken(account.getJwtToken())
                .setExpiration(expiration);
    }
}

@Getter
@Setter
@Accessors(chain = true)
@AllArgsConstructor
@NoArgsConstructor
@ToString
@Schema(description = "JwtToken Cache Request Message")
public static class RequestMessage {
    @Schema(example = "1L")
    private Long accountId;
}

@Getter
@Setter
@Accessors(chain = true)
@AllArgsConstructor
@NoArgsConstructor
@ToString
@JsonInclude(NON_NULL)
@Schema(description = "JwtToken Cache Info")
public static class JwtTokenInfo {
    @Schema(example = "ff6681f0-50f8-4110-bf96-ef6cec45780e")
    private String id;

    @Schema(example = "1L")
    private Long accountId;

    @Schema(example = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.....")
    private String jwtToken;

    @Schema(example = "2021-01-01T00:00:00")
    @JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd'T'HH:mm:ss")
    private LocalDateTime updatedAt;
}

3. Callback 함수 파일 : 이 파일은 Service layer에서 Message을 보내고 난 후 ack를 확인하기 위한 함수를 정의해 두었다.

@Slf4j
public class JwtTokenFutureCallback implements ListenableFutureCallback<SendResult<String, Object>> {
    private final Object message;

    public JwtTokenFutureCallback(Object message) {
        this.message = message;
    }

    @Override
    public void onFailure(Throwable ex) {
        LOGGER.debug("Unable to send message=[{}] due to : {}", message, ex.getMessage());
    }

    @Override
    public void onSuccess(SendResult<String, Object> result) {
        LOGGER.debug("Sent message=[{}] with offset=[{}]", message, result.getRecordMetadata().offset());
    }
}

이제 중요한 Service layer의 코드이다. 

@Slf4j
@RequiredArgsConstructor
@Service
public class JwtTokenMessageServiceImpl implements JwtTokenMessageService {
    private final KafkaTemplate<String, Object> kafkaJwtTokenTemplate;
    private final ReplyingKafkaTemplate<String, Object, String> jwtTokenReplyingKafkaTemplate;
    private final ObjectMapper objectMapper;
    private final JwtTokenProvider jwtTokenProvider;

    @Override
    public void creteJwtTokenCache(Account account) {
        Date expireDate = jwtTokenProvider.getExpiredDate(account.getJwtToken());

        JwtTokenMessage.CreateMessage createMessage = JwtTokenMessage.CreateMessage.from(account, expireDate);

        LOGGER.debug("topic = {}, payload = {}", JwtTokenTopic.TOPIC_CREATE_JWT_TOKEN, createMessage);

        ListenableFuture<SendResult<String, Object>> listenableFuture = kafkaJwtTokenTemplate.send(JwtTokenTopic.TOPIC_CREATE_JWT_TOKEN, createMessage);

        listenableFuture.addCallback(new JwtTokenFutureCallback(createMessage));
    }

    @Override
    public JwtTokenMessage.JwtTokenInfo getJwtTokenCache(JwtTokenMessage.RequestMessage message) throws ExecutionException, InterruptedException, TimeoutException, JsonProcessingException {
        LOGGER.debug("topic = {}, payload = {}", JwtTokenTopic.TOPIC_REQUEST_JWT_TOKEN, message);

        ProducerRecord<String, Object> record = new ProducerRecord<>(JwtTokenTopic.TOPIC_REQUEST_JWT_TOKEN, message);
        RequestReplyFuture<String, Object, String> replyFuture = jwtTokenReplyingKafkaTemplate.sendAndReceive(record);

        SendResult<String, Object> sendResult = replyFuture.getSendFuture().get(10, TimeUnit.SECONDS);
        ConsumerRecord<String, String> consumerRecord = replyFuture.get(10, TimeUnit.SECONDS);

        JwtTokenMessage.JwtTokenInfo jwtTokenInfo = objectMapper.readValue(consumerRecord.value(), JwtTokenMessage.JwtTokenInfo.class);

        LOGGER.debug("JwtTokenInfo : {}", jwtTokenInfo);
        return jwtTokenInfo;
    }
}

우선 KafkaTemplate의 경우 Dependency Injection을 통해 Configuration에서 등록된 Bean이 매핑되고, 이 template을 통해 

Send() 함수를 호출해서 메세지를 보내준다. 필요한 부분은 Topic과 메세지를 넣는 부분이다. 메세지는 각 메세지의 Object이고, Object를 넣어서 보내면 Configuration에서 설정해 둔 Serializer가 Json String 형태로 변경하여 보내게 된다. 

ListenableFuture<SendResult<String, Object>> listenableFuture = kafkaJwtTokenTemplate.send(JwtTokenTopic.TOPIC_CREATE_JWT_TOKEN, createMessage);

그리고 ReplayingKafkaTemplate으로 메세지를 보낼 경우에는 추가적인 작업이 필요하다. 

1. ProducerRecord를 생성하여 Topic과 Message 정보를 추가한다. 

ProducerRecord<String, Object> record = new ProducerRecord<>(JwtTokenTopic.TOPIC_REQUEST_JWT_TOKEN, message);

2. ReplyingKafkaTemplate의 sendAndReceive() 함수를 통해서 생성한 ProducerRecord를 보낸다. 

RequestReplyFuture<String, Object, String> replyFuture = jwtTokenReplyingKafkaTemplate.sendAndReceive(record);

3. 보낸 후에는 Async하게 응답을 기다리도록 ReplyFuture 를 리턴받아서 응답을 확인한다. 

SendResult<String, Object> sendResult = replyFuture.getSendFuture().get(10, TimeUnit.SECONDS);
ConsumerRecord<String, String> consumerRecord = replyFuture.get(10, TimeUnit.SECONDS);

4. 응답으로 받는 데이터는 보내는 형태와 마찬가지로 Object를 Json으로 변경한 값이므로 약속된 형태로 변경하여 사용한다. 

JwtTokenMessage.JwtTokenInfo jwtTokenInfo = objectMapper.readValue(consumerRecord.value(), JwtTokenMessage.JwtTokenInfo.class);

처음에 이 부분을 이해하고 코드를 추가하고 테스트를 하는데 어려움을 겪었다. 양쪽 서비스에 구현을 한 후 테스트를 하면서 문제 부분을 확인 후에 수정을 하다보니 시간이 많이 걸렸다. 처음부터 Embedded kafka를 설정해서 Unit test 만들면서 했으면 더 쉬웠을 것 같다. 하지만, Embedded kafka를 적용한 것도 시간이 걸렸으니 쌤쌤인가? ㅋㅋ 

다음 포스트에서는 이 메세지들을 받아서 처리하는 부분을 공유해보고자 한다. 

Helm repository에서 chart를 download 할 수 있는데 2가지 타입으로 할 수 있다. 

1. Helm chart로 다운로드 하기 

2. Kubernetes manifest file(Yaml) 파일로 다운로드 하기 

1. Helm chart로 다운로드 하기 

간단한 예로, Consul chart를 다운로드 해보겠다. 우선 Hashicorp Helm repository를 등록한다. 

$ helm repo add hashicorp https://helm.releases.hashicorp.com

`helm pull` 명령어를 통해서 Chart를 다운로드 한다. 

// 가장 최신 차트가 다운로드 된다. 
$ helm pull hashicorp/consul

// 다운로드와 동시에 압축해제를 할 때 사용
$ helm pull hashicorp/consul --untar

2. Kubernetes manifest 파일로 다운로드 하기

이 경우에는 helm template 명령어를 사용한다. Manifest 파일로 다운로드 할 경우 values.yaml 파일을 넣어주면 내가 원하는 값으로 생성이 가능하다. 그렇지 않으면 Default 값으로 생성된다. 

values.yaml

client:
  enabled: true

server:
  replicas: 1
  bootstrapExpect: 1
  disruptionBudget:
    maxUnavailable: 0

$ helm template consul hashicorp/consul -f values.yaml --namespace default --version 0.40.0 > consul.yaml

위와 같이 실행하면 consul.yaml에 모든 Manifest가 추가되어 생성되게 된다. 

Intellij에서 Java project를 진행하다보면 가끔 @Autowired inspection의 오류가 있는 경우가 있다. 

나의 경우는 Spring-kafka-test를 Kafka Unit test를 만들때 발생하였다. 

Embedded kafka를 활용하여 Kafka Unit test를 진행하게 되는데 EmbeddedKafkaBroker의 Dependency를 추가하려고 하는데 Intellij 에서 아래와 같이 Inspection 오류를 표시하였다. 빨갛게 표시를 하고 있어서 여간 귀찮은 것이 아니다. 

실제 동작은 잘되기 때문에 보이지 않게 무시하고 싶었다. 

그래서 @Autowired inspection 표시 등급을 Error에서 Warning으로 변경하여 무시하기로 하였다. 

위와 같이 메뉴를 보면 Edit Inspection profile setting 이라는 메뉴가 있고 이곳으로 들어가자. 

저 부분은 다른 Severtity로 변경해주면 된다. 난 Warning으로 변경하였다. 

빨간 줄이 사라졌다. 마음이 놓인다. 

굿!

Unit test를 구현할 때에 간혹 특정 Function의 동작을 무시하고 싶을 때가 있다. 

나의 경우를 예를 들면 Restful API로 들어온 요청이 있고, 이 중 특정 데이터를 Message Broker를 통해 다른 Server로 전달 후 Redis에 저장을 하는 경우에 필요하였다. 

Message Broker 동작은 단순히 전달만 하고 끝나는 작업이기도 하고 Message Broker에 대한 Unit test는 별도로 수행하기 때문에 Restful API 동작에 대한 Unit test에서는 Message Broker의 동작은 무시하여도 되는 상황이었다. 

이 때 사용할 수 있는 것이 doNothing()이다. 사용법은 아래와 같다. 

doNothing().when(jwtTokenMessageService).creteJwtTokenCache(any());

JwtTokenMessageService는 Message Broker(kafka)로 메세지를 전달하는 역할을 하고 이 서비스는 별도의 Unit test로 테스트가 된다. (이 부분은 차후에 포스팅 하려고 한다.)

미래의 나를 위해 남겨두는 팁이다.