Kafka

Kafka 공통 가이드는 자율주행모빌리티 운영 플랫폼에서 제공하는 Kakfa를 사용해 애플리케이션을 개발할 때 미리 알아두어야 하는 내용을 설명합니다.

💡 사전 준비 사항

내 서비스 확인

플랫폼에서 제공하는 Kafka를 사용하기 위해서는 해당 토픽(Topic)에 대한 접근 권한이 필요합니다. Topic 별 접근 권한은 사용자 서비스의 개발자 센터 > 내 서비스에서 확인하실 수 있습니다. 접근 가능한 토픽을 확인하고, 각 토픽 별로 부여받은 역할(ALL/CONSUMER/PRODUCER) 을 확인합니다.

Kafka 역할 분류

역할	설명
ALL	Kakfa 토픽에 데이터를 전송할 수 있으며, 토픽에 전송된 데이터를 읽어올 수 있습니다. (CONSUMER+PRODUCER)
CONSUMER	Kafka 토픽에 전송된 데이터를 읽어올 수 있습니다.
PRODUCER	Kakfa 토픽에 데이터를 전송할 수 있습니다.

💡 Kafka 연결

기본적으로 자율주행모빌리티 운영 플랫폼에서 제공하는 Kafka는 SASL/OAUTHBEARER 인증 방식을 사용합니다.

Kafka 인증은 토픽 권한 신청 시 발급된 Client ID 및 Client Secret과 Token URL을 이용해 진행됩니다.

1. Kafka 기본 정보

부트스트랩 서버 리스트	포트
kafka01.mqnicrnd5.com kafka02.mqnicrnd5.com kafka03.mqnicrnd5.com	9093

2. 참고 사항

Consumer 연결 시에는 Group ID가 필요합니다. Group ID 형식은 아래와 같습니다.

<사용자서비스 로그인 이메일> + "-group"

만약, 로그인 이메일이 [e-mail] 이라면, Group ID는 "[e-mail]-group" 입니다.

3. Consumer 연결 예시

Java

📌 개발 환경

JDK 21, Gradle 8.7

📌 의존성 추가

dependencies {
    ...
    
    implementation 'org.apache.kafka:kafka-clients:3.5.1'
    implementation 'org.apache.kafka:kafka-streams:3.5.1'
    implementation 'org.apache.kafka:kafka_2.13:3.5.1'
}

📌 코드

Change it 이라고 표기되어 있는 부분을 변경하여 사용하세요. 기타 옵션들은 애플리케이션에 맞게 설정하여 사용하시길 바랍니다.

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler;
import org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule;

import java.util.Arrays;
import java.util.Properties;

public class Consumer {

    public static void main(String[] args) {

        Properties configs = new Properties();
        
        // 필수 인증 정보
        configs.put("bootstrap.servers", "kafka01.mqnicrnd5.com:9093,kafka02.mqnicrnd5.com:9093,kafka03.mqnicrnd5.com:9093");
        configs.put("group.id", "<Group ID>"); // Change it
        configs.put("sasl.mechanism", "OAUTHBEARER");
        configs.put("security.protocol", "SASL_PLAINTEXT");
        configs.put("sasl.login.callback.handler.class", OAuthBearerLoginCallbackHandler.class.getName());
        configs.put("sasl.oauthbearer.token.endpoint.url", "<Token URL>"); // Change it
        configs.put("sasl.jaas.config", String.format(
                "%s required clientId=\"%s\" " + "clientSecret=\"%s\";",
                OAuthBearerLoginModule.class.getName(),
                "<Client ID>", // Change it
                "<Client Secret>" // Change it
        ));
        
        // 옵션
        configs.put("auto.offset.reset", "earliest"); 
        configs.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");    
        configs.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(configs); // Consumer 생성
        consumer.subscribe(Arrays.asList("<Topic Name>")); // Change it

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(500);
            for (ConsumerRecord<String, String> record : records) {
                String input = record.topic();
                System.out.println(record.value());
            }
        }
    }
}

Python

from confluent_kafka import Consumer, KafkaException, KafkaError
import json
from requests_oauthlib import OAuth2Session

# Kafka 컨슈머 설정
conf = {
    'bootstrap.servers': 'kafka01.mqnicrnd5.com:9093,kafka02.mqnicrnd5.com:9093,kafka03.mqnicrnd5.com:9093',
    'security.protocol': 'sasl_plaintext',
    'sasl.mechanism': 'OAUTHBEARER',
    'sasl.oauthbearer.method': 'oidc',
    'sasl.oauthbearer.client.id': '<Client ID>',  # Change it
    'sasl.oauthbearer.client.secret': '<Client Secret>',  # Change it
    'sasl.oauthbearer.token.endpoint.url': '<Token URL>', # Change it
    'group.id': '<Group ID>', # Change it
    'auto.offset.reset': 'earliest'
}

consumer = Consumer(conf)

# 구독할 토픽 리스트
topics = ['<Topic Name>'] # Change it

# 토픽 구독
consumer.subscribe(topics)

# 메시지 처리 함수
def handle_message(message):
    try:
        data = json.loads(message.value().decode('utf-8'))
        print(f"Received message: {data}")
    except json.JSONDecodeError as e:
        print(f"Failed to decode JSON: {e}")
    except Exception as e:
        print(f"Error handling message: {e}")


try:
    while True:
        msg = consumer.poll(1.0)  # 1초 동안 폴링

        if msg is None:
            continue

        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                # End of partition event
                print(f"{msg.topic()} [{msg.partition()}] reached end at offset {msg.offset()}")
            elif msg.error():
                raise KafkaException(msg.error())
        else:
            # 메시지 처리
            handle_message(msg)

except KeyboardInterrupt:
    pass
finally:
    # 컨슈머 종료
    consumer.close()
    print("Consumer closed")

Node.js

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

const kafka = new Kafka({
  clientId: '<Client ID>', // Change it
  brokers: ['kafka01.mqnicrnd5.com:9093', 'kafka02.mqnicrnd5.com:9093', 'kafka03.mqnicrnd5.com:9093'],
  ssl: true,
  sasl: {
    mechanism: 'plain',
    username: '<Client ID>', // Change it
    password: '<Client Secret>', // Change it
  },
});

const consumer = kafka.consumer({ groupId: '<Group ID>' }); // Change it

const run = async () => {
  await consumer.connect();
  await consumer.subscribe({ topic: '<Topic Name>' }); // Change it

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      console.log(message.value.toString());
    },
  });
};

run().catch(console.error);

C#

using Confluent.Kafka;
using System;
using System.Collections.Generic;
using System.Threading;

class Program
{
    static void Main(string[] args)
    {
        var conf = new ConsumerConfig
        {
            BootstrapServers = "kafka01.mqnicrnd5.com:9093,kafka02.mqnicrnd5.com:9093,kafka03.mqnicrnd5.com:9093",
            GroupId = "<Group ID>", // Change it
            SaslMechanism = SaslMechanism.OAuthBearer,
            SecurityProtocol = SecurityProtocol.SaslPlaintext,
            SaslOAuthBearerConfig = "<Token URL>", // Change it
            SaslOAuthBearerTokenRefreshHandler = OAuthBearerTokenRefreshHandlerFactory.Get(), // Implement your token refresh handler if needed
            AutoOffsetReset = AutoOffsetReset.Earliest
        };

        using (var consumer = new ConsumerBuilder<Ignore, string>(conf).Build())
        {
            consumer.Subscribe("<Topic Name>"); // Change it

            var cancellationTokenSource = new CancellationTokenSource();

            Console.CancelKeyPress += (_, e) =>
            {
                e.Cancel = true; // prevent the process from terminating.
                cancellationTokenSource.Cancel();
            };

            try
            {
                while (true)
                {
                    try
                    {
                        var result = consumer.Consume(cancellationTokenSource.Token);
                        Console.WriteLine($"Received message: {result.Value}");
                    }
                    catch (ConsumeException e)
                    {
                        Console.WriteLine($"Error occurred: {e.Error.Reason}");
                    }
                }
            }
            catch (OperationCanceledException)
            {
                // This is thrown when the cancellation token is signaled.
            }
            finally
            {
                consumer.Close();
            }
        }
    }
}

4. Producer 연결 예시

Java

📌 개발 환경

JDK 21, Gradle 8.7

📌 의존성 추가

dependencies {
    ...
    
    implementation 'org.apache.kafka:kafka-clients:3.5.1'
    implementation 'org.apache.kafka:kafka-streams:3.5.1'
    implementation 'org.apache.kafka:kafka_2.13:3.5.1'
}

📌 코드

Change it 이라고 표기되어 있는 부분을 변경하여 사용하세요. 기타 옵션들은 애플리케이션에 맞게 설정하여 사용하시길 바랍니다.

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler;
import org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule;

import java.util.Properties;

public class Producer {

    public static void main(String[] args) {

        Properties configs = new Properties();
        // 필수 인증 정보
        configs.put("bootstrap.servers", "kafka01.mqnicrnd5.com:9093,kafka02.mqnicrnd5.com:9093,kafka03.mqnicrnd5.com:9093");
        configs.put("sasl.mechanism", "OAUTHBEARER");
        configs.put("security.protocol", "SASL_PLAINTEXT");
        configs.put("sasl.login.callback.handler.class", OAuthBearerLoginCallbackHandler.class.getName());
        configs.put("sasl.oauthbearer.token.endpoint.url", "<Token URL>"); // Change it
        configs.put("sasl.jaas.config", String.format(
                "%s required clientId=\"%s\" " + "clientSecret=\"%s\";",
                OAuthBearerLoginModule.class.getName(),
                "<Client ID>", // Change it
                "<Client Secret>" // Change it
        ));
        
        // 옵션
        configs.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");  
        configs.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // producer 생성
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configs);

        // message 전달
        for (int i = 0; i < 5; i++) {
            String v = "hello"+i;
            producer.send(new ProducerRecord<String, String>("<Topic Name>", v)); // Change it
        }

        // 종료
        producer.flush();
        producer.close();
    }
}

Python

from confluent_kafka import Producer, KafkaException
import json
from requests_oauthlib import OAuth2Session

# Kafka 프로듀서 설정
conf = {
    'bootstrap.servers': 'kafka01.mqnicrnd5.com:9093,kafka02.mqnicrnd5.com:9093,kafka03.mqnicrnd5.com:9093',
    'security.protocol': 'sasl_plaintext',
    'sasl.mechanism': 'OAUTHBEARER',
    'sasl.oauthbearer.method': 'oidc',
    'sasl.oauthbearer.client.id': '<Client ID>',  # Change it
    'sasl.oauthbearer.client.secret': '<Client Secret>',  # Change it
    'sasl.oauthbearer.token.endpoint.url': '<Token URL>', # Change it
}

producer = Producer(conf)

# 데이터 객체 정의. 알맞게 정의하여 사용하세요.
data_object = {
    'field1': 'value1',
    'field2': 'value2'
}

# JSON 형식의 데이터 생성
json_data = json.dumps(data_object)

# 메시지 전송
try:
    producer.produce('<Topic Name>', json_data.encode('utf-8'))  # Change it
    print('Message sent successfully')
except KafkaException as e:
    print(f'Failed to send message: {e}')

finally:
    producer.flush()

Node.js

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

const kafka = new Kafka({
  clientId: '<Client ID>', // Change it
  brokers: ['kafka01.mqnicrnd5.com:9093', 'kafka02.mqnicrnd5.com:9093', 'kafka03.mqnicrnd5.com:9093'],
  ssl: true,
  sasl: {
    mechanism: 'plain',
    username: '<Client ID>', // Change it
    password: '<Client Secret>', // Change it
  },
});

const producer = kafka.producer();

const run = async () => {
  await producer.connect();

  for (let i = 0; i < 5; i++) {
    const value = `hello${i}`;
    await producer.send({
      topic: '<Topic Name>', // Change it
      messages: [{ value }],
    });
  }

  await producer.disconnect();
};

run().catch(console.error);

C#

using Confluent.Kafka;
using System;

class Program
{
    static void Main(string[] args)
    {
        var config = new ProducerConfig
        {
            BootstrapServers = "kafka01.mqnicrnd5.com:9093,kafka02.mqnicrnd5.com:9093,kafka03.mqnicrnd5.com:9093",
            SaslMechanism = SaslMechanism.OAuthBearer,
            SecurityProtocol = SecurityProtocol.SaslPlaintext,
            SaslOAuthBearerConfig = "<Token URL>", // Change it
            ClientId = "<Client ID>" // Change it
        };

        using (var producer = new ProducerBuilder<Null, string>(config).Build())
        {
            var topic = "<Topic Name>"; // Change it

            for (int i = 0; i < 5; i++)
            {
                var value = $"hello{i}";
                var deliveryReport = producer.ProduceAsync(topic, new Message<Null, string> { Value = value }).Result;
                Console.WriteLine($"Delivered message to {deliveryReport.TopicPartitionOffset}");
            }
        }
    }
}

💡 주요 옵션

애플리케이션의 성능 향상을 위해서는 알맞는 옵션을 적절히 사용하는 것이 중요합니다. 자율주행모빌리티 운영 플랫폼의 Kafka 서버 자체 성능도 중요하지만, Consumer, Producer의 옵션을 잘 이해하고 적용해야만 애플리케이션의 성능을 최적화 시킬 수 있습니다.

아래에 나열된 옵션은 일반적으로 사용되는 주요 옵션들이므로, 더 많은 옵션을 확인하고 싶으시다면 Apache Kafka 공식 문서를 확인해주세요.

Consumer 주요 옵션

옵션명	설명
bootstrap.servers	카프카 클러스터는 클러스터 마스터라는 개념이 없으므로, 클러스터 내 모든 서버가 클라이언트의 요청을 받을 수 있습니다. 클라이언트가 카프카 클러스터에 처음 연결하기 위한 호스트와 포트 정보를 나타냅니다. 정의된 포맷은 호스트이름 :포트,호스트이름:포트,호스트이름:포트,... 입니다. 호스트 하나만 입력해 사용할 수 있지만, 장애가 발생하는 경우 접속이 불가능하기에 클러스터에 있는 호스트를 모두 입력하는것을 권장합니다.
fetch.min.bytes	한 번에 가져올 수 있는 최소 데이터 크기입니다. 만약 지정한 크기 보다 작은 경우, 요청에 응답하지 않고 데이터가 누적될 때까지 기다립니다.
fetch.max.wait.ms	fetch.min.bytes에 의해 설정된 데이터보다 적은 경우 요청에 응답을 기다리는 최대 시간입니다.
group.id	컨슈머가 속한 컨슈머 그룹을 식별하는 식별자입니다. 동일한 그룹 내의 컨슈머 정보는 모두 공유됩니다.
heartbeat.interval.ms	하트비트가 있다는 것은 컨슈머 상태가 active 임을 의미합니다. 그룹 코디네이터에게 얼마나 자주 KafkaConsumer poll()메소드로 하트비트를 보낼 것인지 조정합니다. session.timeout.ms와 밀접한 관계가 있으며 session.timeout.ms보다 낮은 값으로 설정해야 하며, 일반적으로 session.timeout.ms의 3분의 1정도로 설정합니다.
max.partition.fetch.bytes	파티션당 가져올 수 있는 최대 크기를 의미합니다.
session.timeout.ms	컨슈머와 브로커 사이의 세션 타임아웃 시간입니다. 이 시간을 이용해 컨슈머가 종료된 것인지를 판단합니다. 컨슈머는 주기적으로 하트비트를 보내야 하고, 만약 이 시간 전까지 하트비트를 보내지 않았다면 해당 컨슈머는 종료된 것으로 간주하고 컨슈머 그룹에서 제외하고 리밸런싱을 시작합니다.
enable.auto.commit	백그라운드로 주기적으로 오프셋을 커밋합니다.
auto.offset.reset	카프카에서 초기 오프셋이 없거나 현재 오프셋이 더 이상 존재하지 않은 경우에 다음 옵션으로 리셋합니다. earliest: 가장 초기의 오프셋값으로 설정합니다. latest : 가장 마지막의 오프셋값으로 설정합니다. none : 이전 오프셋값을 찾지 못하면 에러를 나타냅니다.
auto.commit.interval.ms	주기적으로 오프셋을 커밋하는 시간입니다.
fetch.max.bytes	한번에 가져올 수 있는 최대 데이터 사이즈입니다.
max.poll.records	한번의 poll() 요청으로 가져올 수 있는 최대 레코드 수를 조정합니다.
max.poll.interval.ms	컨슈머가 하트비트만 보내고 메시지를 가져오지 않을 경우 무한정 해당 파티션을 점유할 수 없도록 주기적으로 poll을 호출하지 않으면 장애라고 판단하고 컨슈머 그룹에서 제외한 후 다른 컨슈머가 해당 파티션에서 메시지를 가져갈 수 있게 합니다.
request.timeout.ms	요청에 대해 응답을 기다리는 최대 시간입니다. 이 시간이 지나면 컨슈머는 요청을 다시 보내거나 재시도 횟수가 모두 소진되면 요청이 실패합니다.

Producer 주요 옵션

옵션명	설명
bootstrap.servers	컨슈머와 동일합니다.
acks	프로듀서가 카프카 토픽의 리더에게 메세지를 전송한 후 요청을 완료하기 전 ack(승인)의 수입니다. 이는 메세지의 내구성을 제어합니다. 수가 크면 메세지 손실 가능성이 낮아지지만 속도가 줄어들고 수가 작으면 손실 가능성이 높아지지만 속도가 높아집니다. 0 : 프로듀서는 카프카 서버로부터 어떠한 ack도 기다리지 않습니다. 1 : 리더는 데이터를 기록하지만, 모든 팔로워는 확인하지 않습니다. 이 경우 데이터 손실 가능성이 있습니다. all 혹은 -1 : ISR의 팔로워로부터 데이터에 대한 acks를 기다립니다. 다소 느릴 수 있지만 메세지가 손실되지 않음을 보장합니다.
buffer.memory	프로듀서가 카프카 서버로 데이터를 보내기 위해 잠시 대기(배치 전송이나 딜레이 등)할 수 있는 전체 메모리 바이트입니다.
compression.type	프로듀서가 메시지 전송 시 선택할 수 있는 압축 타입입니다. none, gzip, snappy, lz4, zstd 중 원하는 타입을 선택할 수 있습니다.
enable.idempotence	이 값을 true로 하는 경우 프로듀서가 각 메세지가 정확히 한번 전송하도록 보장합니다. 즉 중복 없는 전송이 가능합니다. 단 프로듀서 애플리케이션에 장애가 발생할 경우 보장하지 않습니다. true로 할 경우 max.in.flight.requests.per.connection이 5 이하여야 하고, retries는 0 이상, acks는 'all' 로 설정해야 합니다.
retries	일시적인 오류로 인해 전송에 실패한 데이터를 다시 보내는 횟수입니다.
batch.size	프로듀서는 동일한 파티션으로 보내는 여러 데이터를 모아서 배치로 보내려고 시도합니다. 적절한 배치 크기 설정은 성능에 도움을 줍니다.
linger.ms	배치형태의 메시지를 보내기 전에 추가적인 메시지를 위해 기다리는 시간을 조정합니다. 배치 사이즈에 도달하지 못한 상황에서 linger.ms 제한 시간에 도달했을때 메시지를 전송합니다. 즉, 이 시간이 경과하면 누적된 배치 크기가 batch.size 미만이더라도 즉시 기록을 보냅니다.
max.request.size	프로듀서가 단일 요청으로 보낼수 있는 최대 메시지 바이트입니다.