[AWS] MSK (Managed Streaming for Kafka)

MSK는 메세지 큐인 카프카를 제공하는 서비스다.
Managed Streaming for Kafka의 준말이다. 이름 진짜 못짓는거 같다.

동일하게 카프카를 띄워주는 완전관리형 서비스로 키네시스가 있다.
각자 장단점이 있으니 용처에 맞게 선택하면 된다.
https://blog.naver.com/sssang97/222998247667

비용

비용은 기본적으로 클러스터가 올라가있는 시간, 저장소 등으로 부과된다.

프로비저닝과 서버리스 모드가 있다.

프로비저닝 요금표다.

프로비저닝은 가장 작은 사이즈로 해도 한달 40달러는 나오고, 스토리지는 GB당 0.114 달러가 나온다.
문제는, MSK는 인스턴스를 하나만 띄우는게 안된단 것이다. 고가용성을 강제해서 무조건 2개 이상은 띄워야 한다.
그래서 최소 옵션으로 한개 띄웠을 때의 기본비용은 사실상 한달 80달러다.

서버리스의 경우에는 서울 리전이 지원되지 않았..었는데, 2달전부터 지원하기 시작했다.

굉장히 비싼 편이다.
다른거 따지지 않고 클러스터 1개 떠있는 시간만 세어봐도, 한달에 662달러가 나온다.
서버리스 플랜의 장점이 탄력적으로 비용을 절감할 수 있다는건데, 별로 그런 부분에서 메리트가 보이진 않는 것 같다.

자세한건 요금표를 참고하길 바란다.
https://aws.amazon.com/ko/msk/pricing/

클러스터 생성

생성 페이지에 들어가면 이런 창이 뜬다.

빠른 생성으로 대충 생성하면

대충 이렇게 생성이 된다.
그대로 만들었다가는 참으로 곤란한 상황에 처하기 쉬우니, 되도록 사용자 지정 생성으로 만드는 편을 추천한다.

참고로, 이건 온프레미스 kafka와 동일하게 사용할 수 있다고 생각하면 곤란하다.
퍼블릭 엔드포인트를 열 수는 있는데, 그러려면 강제로 IAM 같은 보안 옵션을 켜야 한다.
되도록이면 VPC 안에서만 쓰는 것이 바람직하다.

이름 짓고

요금 폭탄을 맞고 싶지 않다면, 프로비저닝으로 한다.

영역은 2개로만 하자.
굳이 3개로 할 필요가 있나 싶다.

스토리지는 적당히 준다.
스토리지 크기는 늘일 수만 있고 줄일 수는 없으니, 그점에 유의한다.

그리고 VPC와 서브넷을 적당히 깔아준다.
퍼블릭 액세스가 필요하다면 여기서 퍼블릭 서브넷을 골라줘야 한다.

보안그룹도 적당히 골라준다.
퍼블릭 액세스는 초기 생성때는 지정하지 못하고, 생성 후에 업데이트를 직접 해서 활성화를 해줘야 한다.

이게 중요하다.
퍼블릭 액세스를 허용하려면 여기서 인증 방법을 하나 이상 선택해야 한다.
프라이빗 액세스만 할거라면 "인증되지 않은 액세스"를 켜도 좋다.

기본설정

그리고 생성한다.

그러면 최대 15분 안에 생성이 완료된다고 하는데, 다 개뻥이다.
나는 40분쯤 걸렸다.

심지어 뭐 설정 하나 바꿀때마다 업데이트도 그정도 시간이 걸린다.
조금이라도 더 뜯어먹으려고 일부러 이러는건가?

생성이 완료되면 이리 뜬다.

퍼블릭 액세스를 켜려면 편집창으로 진입해서 이렇게 체크를 해주면 된다.

퍼블릭 액세스가 활성화되면 이런 식으로 퍼블릭 엔드포인트가 열리고

퍼블릭 액세스를 쓰지 않는다면 저 프라이빗 엔드포인트만을 쓰면 된다.

사용법 (Node.js)

한번 메세지가 잘 오가나 테스트나 해보자.
나는 퍼블릭 액세스를 꺼놨기 때문에, VPC 내에서 접속을 해야만 했다.

Kafka와 거의 다 호환이 되기 때문에, 사용법도 거의 같다.
아래는 토픽을 명시적으로 생성하는 코드다.

const { Kafka } = require("kafkajs");

async function main() {
  const kafka = new Kafka({
    clientId: "test-producer",
    brokers: [
      "b-1.testcluster.t2n3y6.c3.kafka.ap-northeast-2.amazonaws.com:9092",
      "b-2.testcluster.t2n3y6.c3.kafka.ap-northeast-2.amazonaws.com:9092",
    ],
  });
  console.log("client connected");

  const admin = kafka.admin();
  await admin.connect();
  console.log("admin connected");

  await admin.createTopics({
    waitForLeaders: false,
    topics: [
      {
        topic: "test_topic1",
      },
    ],
  });
  console.log("topic created");
}

main();

이건 컨슈머

const { Kafka } = require("kafkajs");

async function main() {
  const kafka = new Kafka({
    clientId: "test-producer",
    brokers: [
      "b-1.testcluster.t2n3y6.c3.kafka.ap-northeast-2.amazonaws.com:9092",
      "b-2.testcluster.t2n3y6.c3.kafka.ap-northeast-2.amazonaws.com:9092",
    ],
  });

  const consumer = kafka.consumer({ groupId: "test-group" });

  // 접속
  await consumer.connect();

  // 구독
  await consumer.subscribe({ topic: "test_topic1", fromBeginning: false });

  // 실행
  await consumer.run({
    // 메세지가 들어올 때마다 호출됨
    eachMessage: async ({ topic, partition, message }) => {
      console.log(`# topic ${topic}, partition ${partition}`);
      console.log({
        value: message.value.toString(),
      });
    },
  });
}

main();

이건 프로듀서다.

const { Kafka } = require("kafkajs");

async function main() {
  const kafka = new Kafka({
    clientId: "test-producer",
    brokers: [
      "b-1.testcluster.t2n3y6.c3.kafka.ap-northeast-2.amazonaws.com:9092",
      "b-2.testcluster.t2n3y6.c3.kafka.ap-northeast-2.amazonaws.com:9092",
    ],
  });

  const producer = kafka.producer();

  // 접속
  await producer.connect();

  // 메세지 전송
  await producer.send({
    topic: "test_topic1",
    messages: [
      { value: JSON.stringify({ user_id: 1, message: "안녕하세요" }) },
    ],
  });
}

main();

컨슈머 실행하고

프로듀서를 실행해서 메세지를 날리면

잘 날라올 것이다.

참조
https://docs.aws.amazon.com/msk/latest/developerguide/msk-get-bootstrap-brokers.html
https://blog.naver.com/sssang97/222889537359
https://stackoverflow.com/questions/60543419/amazon-kinesis-vs-aws-manage-service-kafka-msk-connect-from-on-prem