Kafka Producer 설정 가이드

Kafka를 처음 구축하면서(MSK) 단순히 Produce/Consume 으로 사용하는 것보다 프로듀서, 컨슈머, 브로커, 토픽 등 설정 값들에 대해 이해하고 적절히 설정하는 것이 꽤나 어려웠는데요,

이와 관련하여 각각 어떤 설정들이 존재하는지 살펴보고, 상황에 맞는 값들을 잘 설정할 수 있도록 정리하고자 합니다.

 

우선 카프카 프로듀서(Producer) 관련 설정부터!

 

• https://kafka.apache.org/documentation/#producerconfigs

Apache Kafka

---

key.serializer

Serializer class for key that implements the org.apache.kafka.common.serialization.Serializer interface.

• key 데이터 직렬화를 위한 Serializer 클래스
• Type: class
• Default: 
• Name: KEY_SERIALIZER_CLASS_CONFIG

String 타입의 직렬화 클래스(StringSerializer)을 일반적으로 사용한다.

 

value.serializer

Serializer class for value that implements the org.apache.kafka.common.serialization.Serializer interface.

• value 데이터 직렬화를 위한 Serializer 클래스
• Type: class
• Default: 
• Name: VALUE_SERIALIZER_CLASS_CONFIG

JSON 타입의 직렬화 클래스(JsonSerializer)를 일반적으로 사용한다.

 

bootstrap.servers

A list of host/port pairs used to establish the initial connection to the Kafka cluster. Clients use this list to bootstrap and discover the full set of Kafka brokers. While the order of servers in the list does not matter, we recommend including more than one server to ensure resilience if any servers are down. This list does not need to contain the entire set of brokers, as Kafka clients automatically manage and update connections to the cluster efficiently. This list must be in the form host1:port1,host2:port2,...

• Kafka 클러스터의 초기 연결을 위한 브로커 서버의 호스트/포트 리스트
• Type: list
• Default: ""
• Name: BOOTSTRAP_SERVERS_CONFIG

고가용성을 위해 보통 한 대 이상의 서버(보통 3대정도로 구축하는 듯)를 권장한다고 한다.

 

compression.type

Enum KafkaProducerProperties.CompressionType

The compression type for all data generated by the producer. The default is none (i.e. no compression).
Valid values are none, gzip, snappy, lz4, or zstd. Compression is of full batches of data, so the efficacy of batching will also impact the compression ratio (more batching means better compression).

• 메시지에 대한 압축 방식
• Type: String
• Default: none (none, gzip, snappy, lz4, zstd)
• Name: COMPRESSION_TYPE_CONFIG

메시지의 크기, 처리량, 네트워크 대역폭, CPU 사용량 등을 고려하여 적절한 압축 타입을 설정하면 될 듯 하다.

(현재 발행하는 메시지가 간단하고, 처리량도 많지 않을 상황이기에 따로 압축 설정을 하지는 않았다.)

https://developer.ibm.com/articles/benefits-compression-kafka-messaging/

 

retries

Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error. Note that this retry is no different than if the client resent the record upon receiving the error.

Produce requests will be failed before the number of retries has been exhausted if the timeout configured by delivery.timeout.ms expires first before successful acknowledgement. Users should generally prefer to leave this config unset and instead use delivery.timeout.ms to control retry behavior.

Enabling idempotence requires this config value to be greater than 0. If conflicting configurations are set and idempotence is not explicitly enabled, idempotence is disabled.

Allowing retries while setting enable.idempotence to false and max.in.flight.requests.per.connection to greater than 1 will potentially change the ordering of records because if two batches are sent to a single partition, and the first fails and is retried but the second succeeds, then the records in the second batch may appear first.

• 프로듀서가 메시지 발행 실패 시 재시도 횟수 설정
• Type: int
• Default: 2_147_483_647 (Integer.MAX_VALUE)
• Name: RETRIES_CONFIG
• 같이 봐야할 설정: delivery.timeout.ms, enable.idempotence, max.in.flight.requests.per.connection

 

위 설명 중 중요하게 봐야 할 문장이 있는데,

Produce requests will be failed before the number of retries has been exhausted if the timeout configured by delivery.timeout.ms expires first before successful acknowledgement.

 

즉 재시도 횟수가 소진되기 전 delivery.timeout.ms에 설정된 타임아웃에 먼저 도달하게 되면 요청이 실패하게 된다.

(retry, timeout 중 하나라도 먼저 도달하게 될 경우 메시지 실패 처리)

 

따라서 실제 재시도 횟수(Integer.MAX_VALUE)만큼 발행에 대해 재시도를 하는것은 아니다.

timeout 설정이 없다면 재시도 횟수를 반드시 설정을 해주어야 할 것 같은데, timeout 설정이 존재하기 때문에 retry에 대한 커스텀 설정은 따로 해주지 않아도 큰 문제가 없지 않을까 싶다.

(delivery.timeout.ms 는 아래에서 살펴보겠지만 기본이 120000, 2분)

 

추가적으로 아래에서 살펴볼 idempotence(멱등성) 설정으로 인해 중복 발행 없이 안전하게 재시도 요청도 가능하다.

 

delivery.timeout.ms

An upper bound on the time to report success or failure after a call to send() returns. This limits the total time that a record will be delayed prior to sending, the time to await acknowledgement from the broker (if expected), and the time allowed for retriable send failures.

The producer may report failure to send a record earlier than this config if either an unrecoverable error is encountered, the retries have been exhausted, or the record is added to a batch which reached an earlier delivery expiration deadline.

The value of this config should be greater than or equal to the sum of request.timeout.ms and linger.ms

• 프로듀서가 메시지를 브로커에 성공적으로 발행하기까지 기다리는 최대 시간
• 즉, 메시지 발행의 성공/실패 응답이 올때까지 얼마나 기다릴것인지? 를 의미하는 값
• Type: int
• Default: 120000 (2 minutes)
• Name: DELIVERY_TIMEOUT_MS_CONFIG

디폴트 값인 2분도 충분히 넉넉한 시간이 아닐까 싶다. 단순히 메시지 발행 자체로 2분동안 실패하는 경우면 메시지 발행에 실패한다기 보다는, 카프카 클러스터 자체의 문제이지 않을까? 싶다.

 

enable.idempotence

KIP-679: Producer will enable the strongest delivery guarantee by default

When set to 'true', the producer will ensure that exactly one copy of each message is written in the stream. If 'false', producer retries due to broker failures, etc., may write duplicates of the retried message in the stream.

Note that enabling idempotence requires max.in.flight.requests.per.connection to be less than or equal to 5 (with message ordering preserved for any allowable value), retries to be greater than 0, and acks must be 'all'.

Idempotence is enabled by default if no conflicting configurations are set. If conflicting configurations are set and idempotence is not explicitly enabled, idempotence is disabled. If idempotence is explicitly enabled and conflicting configurations are set, a ConfigException is thrown.

• 메시지 발행 시 정확히 하나의 메시지(exactly one)만 발송하도록 보장하는 설정
• max.in.flight.requests.pr.connect 이 5보다 작아야 하고, retries는 0보다 커야하고, acks는 'all'이어야 함
• Type: boolean
• Default: true
• Name: ENABLE_IDEMPOTENCE_CONFIG

default가 true(메시지의 멱등성 보장)이기 때문에, 아래 살펴볼 max.in.flight.requests.pr.connect 설정이나 retries 값도 따로 설정하지 않았다면 크게 커스텀이 필요없는 설정

 

Kafka에서 enable.idempotence를 활성화하면 Producer ID + SequenceNumber로 중복 메시지를 필터링한다.

 

acks

The number of acknowledgments the producer requires the leader to have received before considering a request complete. This controls the durability of records that are sent. The following settings are allowed:
  - acks=0 If set to zero then the producer will not wait for any acknowledgment from the server at all. The record will be immediately added to the socket buffer and considered sent. No guarantee can be made that the server has received the record in this case, and the retries configuration will not take effect (as the client won't generally know of any failures). The offset given back for each record will always be set to -1.
  - acks=1 This will mean the leader will write the record to its local log but will respond without awaiting full acknowledgement from all followers. In this case should the leader fail immediately after acknowledging the record but before the followers have replicated it then the record will be lost.
  - acks=all (Default) This means the leader will wait for the full set of in-sync replicas to acknowledge the record. This guarantees that the record will not be lost as long as at least one in-sync replica remains alive. This is the strongest available guarantee. This is equivalent to the acks=-1 setting.

Note that enabling idempotence requires this config value to be 'all'. If conflicting configurations are set and idempotence is not explicitly enabled, idempotence is disabled.

• 프로듀서가 요청 완료를 결정하기 전 리더로부터 받아야하는 응답 횟수로, 전송되는 레코드의 내구성을 제어한다.
  • acks=0 : 프로듀서는 서버로부터 어떤 확인도 기다리지 않고 전송된것으로 간주하기에 서버가 레코드를 받았다는 보장을 할 수 없고, 클라이언트는 실패를 알 수 없기 때문에 재시도가 적용되지 않음
  • acks=1 : 리더의 응답을 받지만 다른 팔로워들의 승인을 기다리지 않고 응답 처리를 하는 설정
  • acks=all : (기본값) 리더가 전체 팔로워들의 승인까지 기다리는 설정, 최소 하나의 in-sync 복제본이 살아있는 한 레코드가 손실되지 않는 것을 보장함 (acks = -1 과 동일)
• Type: String
• Default: all [0, 1, all]
• Name: ACKS_CONFIG

 

max.in.flight.requests.per.connection

The maximum number of unacknowledged requests the client will send on a single connection before blocking. Note that if this configuration is set to be greater than 1 and enable.idempotence is set to false, there is a risk of message reordering after a failed send due to retries (i.e., if retries are enabled); if retries are disabled or if 
enable.idempotence is set to true, ordering will be preserved.

Additionally, enabling idempotence requires the value of this configuration to be less than or equal to 5, because broker only retains at most 5 batches for each producer. If the value is more than 5, previous batches may be removed on broker side.

• 프로듀서가 클러스터로부터 응답을 받지 못한 상황에서 connection당 요청을 보낼 수 있는 최대 레코드 갯수
• Type: int
• Default: 5
• Name: MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION

 

if this configuration is set to be greater than 1 and enable.idempotence is set to false, there is a risk of message reordering after a failed send due to retries

설정 값이 1보다 크고, enable.idempotence가 false라면 재시도로 인해 실패건들에 대한 발송 후 순서가 변경될 위험이 있다.

하지만 idempotence 설정을 true로 하거나 retries를 disabled하면 위와 같은 문제가 발생하지 않는다고 한다.

 

enable.idempotence = false 로 인한 메시지 순서가 바뀔 수 있는 이슈

 

idempotence 설정을 true로 하게되면 브로커가 최대 5개의 producer만 유지하기 때문에 max.in.flight.requests.per.connection 값은 5보다 작아야한다고 한다.

 

추가적으로 살펴보면 좋을 설정들

• linger.ms, batch.size (배치처리 관련)
• mterics
• 인증 (IAM, SASL/SCRAM, SSL 등)

Amazon MSK를 애플리케이션에서 사용하고 접근하기 위해서는 IAM, SASL/SCRAM, TLS 등의 인증 방식들이 존재하는데 인증 방식에 따라 브로커의 포트 번호도 달라지고, 설정도 달라질 수 있다.

https://aws.amazon.com/ko/blogs/tech/amazon-msk-topic-iam-access-control/

 

참고로 IAM을 통한 인증 시 포트번호는 9098이고, 대략적으로 아래와 같은 설정들이 필요하다.

@Bean
fun producerFactory(): ProducerFactory<String, Any> {
    val configProps =
        mutableMapOf<String, Any>().apply {
            this[ProducerConfig.BOOTSTRAP_SERVERS_CONFIG] = bootstrapServers
            
            ...

            // MSK IAM 인증 설정
            this[CommonClientConfigs.SECURITY_PROTOCOL_CONFIG] = "SASL_SSL"
            this[SaslConfigs.SASL_MECHANISM] = "AWS_MSK_IAM"
            this[SaslConfigs.SASL_JAAS_CONFIG] = "software.amazon.msk.auth.iam.IAMLoginModule required;"
            this[SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS] = "software.amazon.msk.auth.iam.IAMClientCallbackHandler"
        }

    return DefaultKafkaProducerFactory(configProps)
}

 

 

Producer 설정 예시 코드

import org.apache.kafka.clients.CommonClientConfigs
import org.apache.kafka.clients.producer.ProducerConfig
import org.apache.kafka.common.config.SaslConfigs
import org.apache.kafka.common.serialization.StringSerializer
import org.springframework.beans.factory.annotation.Value
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.kafka.annotation.EnableKafka
import org.springframework.kafka.core.DefaultKafkaProducerFactory
import org.springframework.kafka.core.KafkaTemplate
import org.springframework.kafka.core.ProducerFactory
import org.springframework.kafka.support.serializer.JsonSerializer

@Configuration
@EnableKafka
class KafkaProducerConfig(
    @Value("\${spring.kafka.bootstrap-servers}") private val bootstrapServers: List<String>,
) {

    @Bean
    fun kafkaTemplate(): KafkaTemplate<String, Any> = KafkaTemplate(producerFactory())

    /**
     * https://kafka.apache.org/documentation.html#producerconfigs
     */
    @Bean
    fun producerFactory(): ProducerFactory<String, Any> {
        val configProps =
            mutableMapOf<String, Any>().apply {
                this[ProducerConfig.BOOTSTRAP_SERVERS_CONFIG] = bootstrapServers
                this[ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG] = StringSerializer::class.java
                this[ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG] = JsonSerializer::class.java
                
                // 아래 설정은 모두 default 값들과 동일함
                this[ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG] = true
                this[ProducerConfig.ACKS_CONFIG] = "all"
                this[ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG] = 120000 // 2 minutes
                this[ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION] = 5

                // MSK IAM 인증 설정
                this[CommonClientConfigs.SECURITY_PROTOCOL_CONFIG] = "SASL_SSL"
                this[SaslConfigs.SASL_MECHANISM] = "AWS_MSK_IAM"
                this[SaslConfigs.SASL_JAAS_CONFIG] = "software.amazon.msk.auth.iam.IAMLoginModule required;"
                this[SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS] = "software.amazon.msk.auth.iam.IAMClientCallbackHandler"
            }

        return DefaultKafkaProducerFactory(configProps)
    }
}