Kafka

원문 보기 위키 갱신 2026.10.11

Kafka 컴포넌트는 Apache Kafka 메시지 브로커와 통신하는 Camel 컴포넌트예요. producer와 consumer를 모두 지원하고, 토픽·파티션·오프셋 관리부터 배치 컨슈밍, 트랜잭션, Kerberos/OAuth 인증까지 폭넓은 옵션을 제공해요. 이 문서는 컴포넌트/엔드포인트 옵션, 메시지 헤더, 사용법과 예제를 한국어로 옮긴 거예요.

출처: 문서

본문

Camel 2.13부터 지원

producer와 consumer 모두 지원

Kafka 컴포넌트는 Apache Kafka 메시지 브로커와 통신하는 데 쓰여요.

<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-kafka</artifactId>
    <version>x.x.x</version>
    <!-- use the same version as your Camel core version -->
</dependency>

URI 형식 (URI Format)

kafka:topic[?options]

옵션 설정 방법 (Configuring Options)

Camel 컴포넌트는 두 단계로 나눠 설정합니다:

컴포넌트(component) 레벨
엔드포인트(endpoint) 레벨

컴포넌트 옵션 설정 (Configuring Component Options)

컴포넌트 레벨에서는 엔드포인트들이 이어서 물려받는 일반적이고 공유되는 설정을 해요. 가장 상위의 설정 레벨이죠.

예를 들어 컴포넌트에는 보안 설정, 인증용 자격 증명, 네트워크 연결 URL 같은 것들이 있을 수 있어요.

컴포넌트에 따라 옵션이 몇 개뿐인 것도 있고 아주 많은 것도 있어요. 컴포넌트는 보통 널리 쓰이는 기본값이 미리 설정되어 있어서, 몇 가지 옵션만 설정하거나 아예 하나도 건드리지 않아도 되는 경우가 많아요.

컴포넌트는 다음 방법으로 설정할 수 있어요:

the Component DSL.
설정 파일(application.properties, *.yaml 파일 등).
Java 코드에서 직접.

엔드포인트 옵션 설정 (Configuring Endpoint Options)

엔드포인트는 옵션이 많아서 설정에 시간을 더 쓰게 될 거예요. 이 옵션들이 엔드포인트가 원하는 일을 하도록 커스터마이즈해 줘요. 옵션은 엔드포인트가 consumer(from)로 쓰이는지, producer(to)로 쓰이는지, 둘 다인지에 따라 분류되기도 해요.

엔드포인트 설정은 대부분 엔드포인트 URI에 _path_와 query 파라미터로 바로 하는 경우가 많아요. Java에서 엔드포인트와 데이터 포맷을 _타입 세이프_하게 설정하려면 Endpoint DSL과 DataFormat DSL을 쓸 수도 있어요.

옵션을 설정할 때 좋은 습관은 Property Placeholders를 사용하는 거예요.

프로퍼티 플레이스홀더는 이런 장점이 있어요:

URL, 포트 번호, 민감한 정보 같은 설정을 하드코딩하는 걸 막아 줘요.
설정을 코드 밖으로 분리할 수 있어요.
코드가 더 유연해지고 재사용하기 좋아져요.

다음 두 섹션에서 모든 옵션을 나열해요. 먼저 컴포넌트 옵션, 그다음 엔드포인트 옵션 순서예요.

컴포넌트 옵션 (Component Options)

Kafka 컴포넌트가 지원하는 옵션은 아래와 같아요.

Name	Description	Default	Type
additionalProperties (common)	kafka consumer나 kafka producer에 Camel 설정에 바로 반영되지 않는 새로운 Kafka 프로퍼티처럼 직접 세팅할 수 없는 값이 있을 때 추가 프로퍼티를 넣어 주는 옵션이에요. 프로퍼티에는 additionalProperties. 접두어를 붙여야 해요. 예: additionalProperties.transactional.id=12345&additionalProperties.schema.registry.url=http://localhost:8811/avro. application.properties 파일에 설정할 때는 camel.component.kafka.additional-properties(Kafka Share 컴포넌트라면 camel.component.kafka-share.additional-properties) 접두어 뒤에 대괄호로 감싼 프로퍼티 이름을 붙여야 해요. 예컨대 delivery.timeout.ms 프로퍼티를 대괄호로 감싸는 식이에요. 여러 값을 넣는 multi-value 옵션이고 접두어는 additionalProperties.예요.		Map
brokers (common)	사용할 Kafka 브로커의 URL이에요. 형식은 host1:port1,host2:port2이고, 이 목록은 브로커 전체의 부분 집합이거나 브로커 부분 집합을 가리키는 VIP일 수도 있어요. Kafka 문서에서는 bootstrap.servers로 알려진 옵션이에요.		String
clientId (common)	client id는 호출 추적을 돕기 위해 매 요청마다 함께 보내는, 사용자가 지정한 문자열이에요. 요청을 보내는 애플리케이션을 논리적으로 식별할 수 있는 값이면 돼요.		String
configuration (common)	엔드포인트들이 재사용할 공통 옵션으로 Kafka 컴포넌트를 미리 설정할 수 있게 해 줘요.		KafkaConfiguration
connectionMaxIdleMs (common)	이 설정이 지정한 밀리초가 지나면 유휴(idle) 연결을 닫아요.	540000	Integer
headerFilterStrategy (common)	Camel 메시지와 주고받는 헤더를 필터링할 커스텀 HeaderFilterStrategy를 쓰려면 이걸 지정해요.		HeaderFilterStrategy
metadataMaxAgeMs (common)	파티션 리더십 변화가 관찰되지 않았더라도 이 밀리초가 지나면 메타데이터를 강제로 새로 고쳐서 새 브로커나 새 파티션을 미리 발견하는 주기예요.	300000	Integer
metricReporters (common)	메트릭 리포터로 쓸 클래스 목록이에요. MetricReporter 인터페이스를 구현하면 새 메트릭이 만들어질 때 알림을 받는 클래스를 끼워 넣을 수 있어요. JMX 통계 등록용 JmxReporter는 항상 포함돼요.		String
metricsSampleWindowMs (common)	메트릭 샘플을 계산하는 시간 창이에요.	30000	Integer
noOfMetricsSample (common)	메트릭 계산에 유지되는 샘플 개수예요.	2	Integer
receiveBufferBytes (common)	데이터를 읽을 때 쓰는 TCP 수신 버퍼(SO_RCVBUF) 크기예요.	65536	Integer
reconnectBackoffMaxMs (common)	접속에 반복적으로 실패한 브로커에 재접속할 때 기다리는 최대 시간(밀리초)이에요. 값을 주면 연속 접속 실패가 이어질 때마다 호스트별 백오프가 이 최댓값까지 지수적으로 늘어나요. 백오프 증가를 계산한 뒤에는 접속 폭주(connection storm)를 막으려고 20%의 무작위 지터를 더해요.	1000	Integer
reconnectBackoffMs (common)	특정 호스트에 재접속을 시도하기 전에 기다리는 시간이에요. 짧은 루프 안에서 같은 호스트에 계속 접속하는 걸 막아 줘요. 이 백오프는 consumer가 브로커에 보내는 모든 요청에 적용돼요.	50	Integer
retryBackoffMaxMs (common)	반복적으로 실패한 브로커에 요청을 재시도할 때 기다리는 최대 시간(밀리초)이에요. 값을 주면 실패한 요청마다 클라이언트별 백오프가 이 최댓값까지 지수적으로 늘어나요. 모든 클라이언트가 재시도 시점에 맞춰 움직이는 걸 막으려고 0.2 배율의 무작위 지터를 백오프에 적용해서, 계산값의 -20%~+20% 범위 안에서 백오프가 결정돼요. retry.backoff.ms를 retry.backoff.max.ms보다 크게 설정하면 지수 증가 없이 retry.backoff.max.ms를 처음부터 고정 백오프로 써요.	1000	Integer
retryBackoffMs (common)	특정 토픽 파티션에 실패한 요청을 재시도하기 전에 기다리는 시간이에요. 일부 장애 상황에서 짧은 루프로 요청을 계속 보내는 걸 막아 줘요. 이 값이 초기 백오프이고, 실패한 요청마다 retry.backoff.max.ms 값까지 지수적으로 늘어나요.	100	Integer
sendBufferBytes (common)	소켓 쓰기 버퍼 크기예요.	131072	Integer
shutdownTimeout (common)	consumer나 producer가 우아하게 종료되고 워커 스레드를 마무리할 때까지 기다리는 밀리초 단위 타임아웃이에요.	30000	int
allowManualCommit (consumer)	KafkaManualCommit을 통한 수동 커밋을 허용할지 여부예요. 이 옵션을 켜면 Exchange 메시지 헤더에 KafkaManualCommit 인스턴스가 저장되고, 사용자가 이 API에 접근해 Kafka consumer로 수동 오프셋 커밋을 할 수 있어요.	false	boolean
autoCommitEnable (consumer)	true면 consumer가 이미 가져온 메시지의 오프셋을 ZooKeeper에 주기적으로 커밋해요. 프로세스가 실패했을 때 이 커밋된 오프셋이 새 consumer가 시작할 위치로 쓰여요.	true	boolean
autoCommitIntervalMs (consumer)	consumer 오프셋을 zookeeper에 커밋하는 주기(밀리초)예요.	5000	Integer
autoOffsetReset (consumer)	커밋된 오프셋이 없거나 커밋된 오프셋이 범위를 벗어났을 때 consumer group이 어디서부터 읽기 시작할지 정해요. 유효한 값은: earliest(가장 이른 사용 가능한 오프셋으로 이동), latest(가장 최신 오프셋으로 이동, 기본값), none(이전 오프셋이 없으면 예외 던짐), by_duration:(ISO-8601 기간을 뒤에 붙임. 예: by_duration:PT5M 또는 by_duration:P1D. Kafka 4.0 이상 필요)이에요.	latest	String
batching (consumer)	처리를 배치(batching)로 할지 스트리밍으로 할지 여부예요. 기본값은 false고, 이때는 스트리밍을 써요. 스트리밍 모드에서는 메시지 본문의 Camel exchange 하나당 kafka 레코드 하나가 처리돼요. 배치 모드에서는 Camel이 많은 kafka 레코드를 모아서 메시지 본문에 List 객체로 묶어 줘요. 배치 모드에서 몇 개의 레코드를 묶을지는 maxPollRecords 옵션으로 정해요.	false	boolean
batchingIntervalMs (consumer)	consumer 배치 모드에서 현재 배치 크기가 maxPollRecords로 정한 최대 크기에 못 미칠 때 배치 완료를 조기에 트리거하기 위한 밀리초 시간을 지정하는 옵션이에요. 트리거는 주어진 간격에 정확히 일어나지 않고 kafka poll 사이에서만 일어날 수 있다는 점에 유의하세요(pollTimeoutMs 옵션 참고). 예컨대 이 값을 10000으로 두면 트리거는 10000 pollTimeoutMs 구간 안에서 일어나요. pollTimeoutMs 기본값이 5000이라서 대략 15초마다 트리거가 일어난다는 뜻이 돼요.		Integer
breakOnFirstError (consumer)	consumer가 exchange를 처리하다 실패했을 때 어떻게 동작할지 조절하는 옵션이에요. false면 consumer가 다음 메시지로 넘어가 계속 처리해요. true면 consumer가 거기서 멈춰요(break out). 기본인 NoopCommitManager를 쓰면 consumer가 오프셋을 커밋하지 않아서 그 메시지를 다시 시도하게 돼요. 이때 consumer는 KafkaManualCommit을 써서 메시지를 처리하는 최선의 방법을 정해야 해요. SyncCommitManager나 AsyncCommitManager를 쓰면 consumer가 실패를 일으킨 메시지의 오프셋으로 되돌아가(seek back) 그 메시지 처리를 다시 시도해요. 다만 매번 실패가 확실한 메시지(예: poison message)라면 같은 메시지를 끝없이 처리하게 될 수 있어요. 그래서 Camel의 에러 핸들러를 쓰는 식으로 이 상황을 다루는 게 권장돼요.	false	boolean
bridgeErrorHandler (consumer)	컨슈머를 Camel 라우팅 에러 핸들러(Error Handler)에 브리지할 수 있게 해 줘요. 즉, Camel 컨슈머가 들어오는 메시지를 집어 올리는 동안 등에서 예외가 발생하면(가능한 경우), 그 예외를 메시지로 처리해서 라우팅 에러 핸들러가 다루게 돼요. 중요: 이것은 서드파티 컴포넌트가 예외가 던져졌을 때 Camel에 알릴 수 있도록 허용하는 경우에만 가능해요. 일부 컴포넌트는 내부에서만 처리하기 때문에 bridgeErrorHandler를 쓸 수 없어요. 다른 상황에서는 Camel 컴포넌트를 개선해서 서드파티 컴포넌트에 연결(hook)해 이를 가능하게 만들 수도 있어요. 기본으로 컨슈머는 org.apache.camel.spi.ExceptionHandler로 예외를 처리하는데, WARN 또는 ERROR 레벨로 로그를 남기고 무시해요.	false	boolean
checkCrcs (consumer)	소비한 레코드의 CRC32를 자동으로 검사해요. 메시지가 전송 중(on-the-wire)이나 디스크에서 손상되지 않았다는 걸 보장해 줘요. 이 검사는 약간의 오버헤드를 더하니 극한의 성능을 추구하는 경우엔 끌 수도 있어요.	true	Boolean
commitTimeoutMs (consumer)	동기 커밋이 끝날 때까지 코드가 기다리는 최대 시간(밀리초)이에요.	5000	Long
consumerRequestTimeoutMs (consumer)	클라이언트가 요청 응답을 기다리는 최대 시간을 조절하는 설정이에요. 타임아웃이 지나도 응답이 오지 않으면 필요하면 요청을 다시 보내고, 재시도를 다 썼으면 요청을 실패로 처리해요.	30000	Integer
consumersCount (consumer)	kafka 서버에 접속하는 consumer 수예요. 각 consumer는 들어오는 데이터를 가져와 처리하는 별도 스레드에서 동작해요.	1	int
fetchMaxBytes (consumer)	fetch 요청에 대해 서버가 돌려줄 최대 데이터 양이에요. 절대 상한은 아니어서, fetch의 첫 번째 비어 있지 않은 파티션의 첫 메시지가 이 값보다 크더라도 consumer가 진행할 수 있도록 그 메시지는 그대로 돌려줘요. 브로커가 받아들이는 최대 메시지 크기는 message.max.bytes(브로커 설정)나 max.message.bytes(토픽 설정)로 정의돼요. consumer는 여러 fetch를 병렬로 수행한다는 점에 유의하세요.	52428800	Integer
fetchMinBytes (consumer)	fetch 요청에 대해 서버가 돌려줄 최소 데이터 양이에요. 데이터가 충분하지 않으면 요청은 그만큼 데이터가 쌓일 때까지 기다렸다가 답해요.	1	Integer
fetchWaitMaxMs (consumer)	fetch.min.bytes를 바로 만족할 데이터가 없을 때 서버가 fetch 요청에 답하기 전까지 블록하는 최대 시간이에요.	500	Integer
groupId (consumer)	이 consumer가 속한 consumer 프로세스 그룹을 유일하게 식별하는 문자열이에요. 같은 group id를 설정하면 여러 프로세스가 같은 consumer group의 일부임을 나타낼 수 있어요. consumer에는 필수인 옵션이에요.		String
groupInstanceId (consumer)	사용자가 제공하는 consumer 인스턴스의 고유 식별자예요. 비어 있지 않은 문자열만 허용돼요. 설정하면 consumer가 정적 멤버(static member)로 취급돼서, 이 ID를 가진 인스턴스는 consumer group에서 어떤 시점에도 하나만 허용돼요. 일시적인 사용 불가(예: 프로세스 재시작)로 인한 그룹 리밸런스를 피하려고 더 긴 세션 타임아웃과 조합해 쓸 수 있어요. 설정하지 않으면 consumer가 전통적인 방식대로 동적 멤버로 그룹에 참여해요.		String
groupProtocol (consumer)
쓸 consumer group 프로토콜이에요. classic 프로토콜은 전통적인 파티션 할당·리밸런싱 방식을 써요. consumer 프로토콜은 더 빠르고 효율적인 리밸런싱을 제공하는 새로운 KIP-848 consumer 리밸런스 프로토콜을 켜요. consumer로 설정하면 classic 전용 프로퍼티(heartbeatIntervalMs, sessionTimeoutMs, partitionAssignor)가 consumer 설정에서 자동으로 제외돼요.

Enum values:

classic
consumer

| classic | String | | groupRemoteAssignor (consumer) | group.protocol이 consumer로 설정됐을 때 쓸 서버 측 assignor 이름이에요. 지정하지 않으면 group coordinator가 브로커에 설정된 기본 assignor(group.consumer.assignors)를 사용해요. | | String | | headerDeserializer (consumer) | kafka 헤더 값을 역직렬화(deserialize)할 커스텀 KafkaHeaderDeserializer를 쓰려면 이걸 지정해요. | | KafkaHeaderDeserializer | | heartbeatIntervalMs (consumer) | Kafka의 그룹 관리 기능을 쓸 때 consumer coordinator에 보내는 하트비트 사이의 예상 시간이에요. 하트비트는 consumer 세션이 활성 상태로 유지되는지 확인하고, 새 consumer가 그룹에 들어오거나 나갈 때 리밸런싱이 원활히 진행되게 하는 데 쓰여요. 이 값은 session.timeout.ms보다 작아야 하고, 보통은 그 값의 1/3을 넘지 않게 설정하는 게 좋아요. 평상시 리밸런스의 예상 시간을 조절하려면 더 낮출 수도 있어요. | 3000 | Integer | | keyDeserializer (consumer) | Deserializer 인터페이스를 구현하는 키(key)용 deserializer 클래스예요. | org.apache.kafka.common.serialization.StringDeserializer | String | | maxPartitionFetchBytes (consumer) | 서버가 파티션당 돌려줄 최대 데이터 양이에요. 요청에 쓰이는 최대 총 메모리는 #partitions max.partition.fetch.bytes가 돼요. 이 크기는 서버가 허용하는 최대 메시지 크기와 같거나 커야 해요. 그렇지 않으면 producer가 consumer가 fetch할 수 있는 것보다 큰 메시지를 보낼 수 있거든요. 그런 일이 생기면 consumer가 특정 파티션의 큰 메시지를 fetch하려다 멈출(stuck) 수 있어요. | 1048576 | Integer | | maxPollIntervalMs (consumer) | consumer group 관리를 쓸 때 poll() 호출 사이의 최대 지연이에요. consumer가 더 많은 레코드를 가져오기 전에 얼마나 오래 유휴 상태로 있을 수 있는지의 상한을 정해요. 이 타임아웃이 끝나기 전에 poll()이 호출되지 않으면 consumer가 실패한 것으로 간주되고, 그룹이 리밸런스를 해서 파티션을 다른 멤버에 재배정해요. | | Integer | | maxPollRecords (consumer) | poll() 한 번의 호출에서 돌려받는 최대 레코드 수예요. | 500 | Integer | | offsetRepository (consumer) | 토픽의 각 파티션 오프셋을 로컬에 저장할 때 쓸 offset repository예요. 하나를 정의하면 autocommit이 꺼져요. | | StateRepository | | partitionAssignor (consumer) | 그룹 관리를 쓸 때 클라이언트가 consumer 인스턴스들 사이에서 파티션 소유권을 나눠 가지는 데 쓸 파티션 할당 전략의 클래스 이름이에요. | org.apache.kafka.clients.consumer.RangeAssignor | String | | pollOnError (consumer) |

kafka가 새 메시지를 poll하는 동안 예외를 던졌을 때 어떻게 할지 정해요. 기본으로는 엔드포인트 레벨에서 명시적인 값이 설정돼 있지 않으면 컴포넌트 설정의 값을 써요. DISCARD는 메시지를 버리고 다음 메시지를 계속 poll해요. ERROR_HANDLER는 Camel의 에러 핸들러로 예외를 처리한 뒤 다음 메시지를 계속 poll해요. RECONNECT는 consumer를 다시 연결하고 그 메시지 poll을 다시 시도해요. RETRY는 consumer가 같은 메시지 poll을 다시 시도하게 해요. STOP은 consumer를 멈춰요(다시 메시지를 소비하려면 수동으로 시작/재시작해야 해요).

Enum values:

DISCARD
ERROR_HANDLER
RECONNECT
RETRY
STOP

시작할 때 KafkaConsumer가 처음부터 읽을지 끝부터 읽을지 정해요. SeekPolicy.BEGINNING: 처음부터 읽어요. SeekPolicy.END: 끝부터 읽어요.

Enum values:

BEGINNING
END

| | SeekPolicy | | sessionTimeoutMs (consumer) | Kafka의 그룹 관리 기능을 쓸 때 장애를 감지하는 데 쓰는 타임아웃이에요. | 45000 | Integer | | specificAvroReader (consumer) | multiple Schema registries documentation with Avro Deserializers 구현과 함께 쓰는 특정 Avro reader 사용을 켜는 옵션이에요. 이 옵션은 외부(external) 구현에서만 사용할 수 있어요(표준 Apache Kafka에는 없음). | false | boolean | | topicIsPattern (consumer) | 토픽이 패턴(정규표현식)인지 여부예요. 패턴에 맞는 동적인 개수의 토픽을 구독하는 데 쓸 수 있어요. | false | boolean | | valueDeserializer (consumer) | Deserializer 인터페이스를 구현하는 값(value)용 deserializer 클래스예요. | org.apache.kafka.common.serialization.StringDeserializer | String | | createConsumerBackoffInterval (consumer (advanced)) | kafka consumer(kafka-client) 생성을 다시 시도하기 전에 기다리는 지연(밀리초)이에요. | 5000 | long | | createConsumerBackoffMaxAttempts (consumer (advanced)) | kafka consumer(kafka-client) 생성을 시도하는 최대 횟수예요. 이 횟수를 넘으면 포기하고 실패해요. consumer 생성 중 오류는 잘못된 설정 때문에 치명적일 수 있어서 복구가 불가능할 수 있어요. 하지만 검증의 일부는 부트스트랩 브로커 호스트명의 DNS 조회인데, 이건 일시적인 네트워크 문제일 수 있고 잠재적으로는 복구 가능해요. 반면 일부 잘못된 kafka 설정처럼 치명적인 오류도 있죠. 안타깝게도 kafka-client는 이런 오류들을 구분해 주지 않아요. Camel은 기본으로 영원히 재시도하므로 절대 포기하지 않아요. 여러 번 시도 후 포기하고 싶다면 이 옵션을 설정하세요. 그러면 Camel은 포기하는 시점에 consumer를 종료해요. 다시 시도하려면 라우트를 멈췄다 시작해서 consumer를 수동으로 재시작할 수 있어요. | | int | | isolationLevel (consumer (advanced)) |

트랜잭션으로 기록된 메시지를 어떻게 읽을지 조절해요. read_committed로 설정하면 consumer.poll()이 커밋된 트랜잭션 메시지만 돌려줘요. read_uncommitted(기본값)로 설정하면 consumer.poll()이 취소된 트랜잭션 메시지까지 전부 돌려줘요. 트랜잭션에 포함되지 않은 메시지는 어느 모드에서든 무조건 돌려줘요. 메시지는 항상 오프셋 순서대로 돌려줘요. 그래서 read_committed 모드에서는 consumer.poll()이 마지막 안정 오프셋(LSO, 처음 열린 트랜잭션 오프셋보다 하나 작은 값)까지만 돌려줘요. 특히 진행 중인 트랜잭션에 속한 메시지 뒤에 나오는 메시지들은 해당 트랜잭션이 완료될 때까지 보류돼요. 그 결과 진행 중인 트랜잭션이 있으면 read_committed consumer는 high watermark까지 읽지 못해요. 또한 read_committed일 때 seekToEnd 메서드는 LSO를 돌려줘요.

Enum values:

read_uncommitted
read_committed

| read_uncommitted | String | | kafkaManualCommitFactory (consumer (advanced)) | Autowired KafkaManualCommit 인스턴스를 만들 때 쓸 팩토리예요. 커스텀 팩토리를 끼워 넣어서, 기본 구현과 다른 특별한 로직이 필요한 수동 커밋을 할 때 커스텀 KafkaManualCommit 인스턴스를 만들 수 있어요. | | KafkaManualCommitFactory | | pollExceptionStrategy (consumer (advanced)) | Autowired 메시지를 poll하는 동안 Kafka 브로커가 던진 예외를 어떻게 처리할지 조절하려고 consumer에 커스텀 전략을 쓰려면 이걸 지정해요. | | PollExceptionStrategy | | subscribeConsumerBackoffInterval (consumer (advanced)) | Deprecated kafka 브로커 구독(subscribe)을 다시 시도하기 전에 기다리는 지연(밀리초)이에요. | 5000 | long | | subscribeConsumerBackoffMaxAttempts (consumer (advanced)) | Deprecated kafka consumer가 kafka 브로커 구독을 시도하는 최대 횟수예요. 이 횟수를 넘으면 포기하고 실패해요. kafka 토픽 구독 중 오류는 네트워크 문제로 인한 일시적 오류일 수 있고 잠재적으로는 복구 가능해요. Camel은 기본으로 영원히 재시도하므로 절대 포기하지 않아요. 여러 번 시도 후 포기하고 싶다면 이 옵션을 설정하세요. 그러면 Camel은 포기하는 시점에 consumer를 종료해요. 다시 시도하려면 라우트를 멈췄다 시작해서 consumer를 수동으로 재시작할 수 있어요. | | int | | subscribeConsumerTopicMustExists (consumer (advanced)) | Camel Kafka consumer가 Kafka 브로커를 구독할 때 토픽이 브로커에 이미 존재하는지 검사하고, 없으면 실패할지 여부예요. 그렇지 않으면 Camel Kafka consumer는 토픽이 Kafka 브로커에 만들어질 때까지 계속 소비를 시도하고, 그때까지는 UNKNOWN_TOPIC_OR_PARTITION 경고(WARN)를 남기며 실패해요. createConsumerBackoffMaxAttempts 옵션을 설정하면 주어진 횟수만큼 시도한 뒤 구독 시도를 포기할 수 있어요. | false | boolean | | batchWithIndividualHeaders (producer) | 이 기능이 켜져 있고 배치의 단일 요소가 Exchange나 Message일 때, producer가 배치 Message를 사용해 값을 결정해서 그 요소 전용의 kafka 헤더 값을 만들어 내요. 평소 동작은 항상 같은 헤더 값을 쓰는 거예요(Iterable이나 Iterator를 담은 부모 Exchange가 그 값을 결정해요). | false | boolean | | bufferMemorySize (producer) | 서버로 보내기를 기다리는 레코드를 버퍼링하는 데 producer가 쓸 수 있는 총 메모리 바이트예요. 레코드가 서버로 전달될 수 있는 속도보다 빨리 쌓이면, producer는 block.on.buffer.full에 지정한 선호에 따라 블록하거나 예외를 던져요. 이 설정은 producer가 쓸 총 메모리와 대략 일치해야 하지만, producer가 쓰는 메모리가 전부 버퍼링에 쓰이는 건 아니라서 엄격한 상한은 아니에요. 압축이 켜져 있으면 압축용과 진행 중인(in-flight) 요청을 유지하는 데 추가 메모리가 좀 더 쓰여요. | 33554432 | Integer | | compressionCodec (producer) |

이 producer가 생성하는 모든 데이터에 사용할 압축 코덱을 지정하게 해 주는 파라미터예요. 유효한 값은 none, gzip, snappy, lz4, zstd예요.

Enum values:

none
gzip
snappy
lz4
zstd

| none | String | | deliveryTimeoutMs (producer) | send() 호출이 반환된 뒤 성공이나 실패를 보고하기까지의 시간 상한이에요. 이 값은 레코드가 전송 전까지 지연되는 총 시간, 브로커의 승인(acknowledgement)을 기다리는 시간(기대한다면), 재시도 가능한 전송 실패에 허용되는 시간을 함께 제한해요. | 120000 | Integer | | enableIdempotence (producer) | 'true'로 설정하면 producer가 각 메시지의 사본이 스트림에 정확히 하나만 기록되도록 보장해요. 'false'면 브로커 장애 등으로 producer가 재시도할 때 재시도한 메시지의 중복이 스트림에 기록될 수 있어요. idempotence를 켜려면 max.in.flight.requests.per.connection이 5 이하(허용되는 어떤 값이든 메시지 순서는 유지)이고, retries가 0보다 커야 하며, acks는 'all'이어야 해요. 충돌하는 설정이 없으면 idempotence는 기본으로 켜져요. 충돌하는 설정이 있고 idempotence를 명시적으로 켜지 않았다면 꺼져요. idempotence를 명시적으로 켰는데 충돌하는 설정이 있으면 ConfigException이 던져져요. | true | boolean | | headerSerializer (producer) | kafka 헤더 값을 직렬화(serialize)할 커스텀 KafkaHeaderSerializer를 쓰려면 이걸 지정해요. | | KafkaHeaderSerializer | | key (producer) | 레코드 키예요(키를 지정하지 않으면 null). 이 옵션이 설정돼 있으면 헤더 KafkaConstants#KEY보다 우선해요. | | String | | keySerializer (producer) | 키용 serializer 클래스예요(아무것도 지정하지 않으면 메시지용과 같은 기본값을 써요). | org.apache.kafka.common.serialization.StringSerializer | String | | lazyStartProducer (producer) | producer를 lazy(첫 메시지 시점)로 시작할지 여부예요. lazy 시작을 쓰면 producer가 시작 중 실패해서 라우트 시작이 실패하는 상황에서도 CamelContext와 라우트를 시작할 수 있고, 시작 실패 처리를 Camel의 라우팅 에러 핸들러로 넘길 수 있어요. 다만 첫 메시지를 처리하는 시점에 producer를 만들고 시작하는 데 시간이 조금 걸려서 전체 처리 시간이 길어질 수 있다는 점은 유의하세요. | false | boolean | | lingerMs (producer) | producer는 요청 전송 사이에 도착한 레코드들을 하나의 배치 요청으로 묶어요. 보통은 레코드가 보낼 수 있는 속도보다 빨리 도착하는 부하 상황에서만 일어나요. 하지만 어떤 상황에서는 클라이언트가 적당한 부하에서도 요청 수를 줄이고 싶을 수 있어요. 이 설정은 약간의 인위적 지연을 더해서 그렇게 해 줘요. 즉 레코드를 즉시 보내는 대신, producer는 주어진 지연까지 기다려서 다른 레코드도 함께 보내 배치로 묶일 수 있게 해요. TCP의 Nagle 알고리즘과 비슷하다고 생각하면 돼요. 이 설정은 배치 지연의 상한을 줘요. 어떤 파티션에 batch.size만큼의 레코드가 모이면 이 설정과 무관하게 즉시 보내지만, 그보다 적은 바이트가 쌓여 있으면 지정한 시간만큼 'linger'(대기)하며 더 많은 레코드를 기다려요. 기본값은 0(즉, 지연 없음)이에요. 예컨대 linger.ms=5로 설정하면 보내는 요청 수가 줄어드는 효과가 있지만 부하가 없을 때 보내는 레코드에는 최대 5ms의 지연이 더해져요. | 5 | Integer | | maxBlockMs (producer) | KafkaProducer의 send(), partitionsFor(), initTransactions(), sendOffsetsToTransaction(), commitTransaction(), abortTransaction() 메서드가 얼마나 오래 블록할지 조절하는 설정이에요. send()에 대해서는 이 타임아웃이 메타데이터 fetch와 버퍼 할당을 기다리는 총 시간을 제한해요(사용자가 제공한 serializer나 partitioner에서의 블록은 이 타임아웃에 포함되지 않아요). partitionsFor()에 대해서는 메타데이터를 사용할 수 없을 때 기다리는 시간을 제한해요. 트랜잭션 관련 메서드는 항상 블록하지만, 트랜잭션 코디네이터를 발견하지 못하거나 타임아웃 안에 응답하지 않으면 타임아웃될 수 있어요. | 60000 | Integer | | maxInFlightRequest (producer) | 블록하기 전에 클라이언트가 단일 연결에서 보낼 미확인(unacknowledged) 요청의 최대 개수예요. 이 설정이 1보다 크고 실패한 전송이 있으면 재시도로 인해(즉, retries가 켜져 있으면) 메시지 순서가 바뀔 위험이 있다는 점에 유의하세요. | 5 | Integer | | maxRequestSize (producer) | 요청의 최대 크기예요. 사실상 최대 레코드 크기의 상한이기도 해요. 서버에는 이 값과 다를 수 있는 자체 레코드 크기 상한이 있다는 점에 유의하세요. 이 설정은 producer가 단일 요청으로 보내는 레코드 배치 수를 제한해서 거대한 요청을 보내는 걸 막아요. | 1048576 | Integer | | partitioner (producer) | 메시지를 하위 토픽들 사이에서 파티셔닝할 partitioner 클래스예요. 기본 partitioner는 키의 해시에 기반해요. | | String | | partitionerIgnoreKeys (producer) | 파티션을 계산할 때 메시지 키를 무시할지 여부예요. 이 설정은 partitioner가 설정되지 않았을 때만 효과가 있어요. | false | boolean | | partitionKey (producer) | 레코드를 보낼 파티션이에요(파티션을 지정하지 않으면 null). 이 옵션이 설정돼 있으면 헤더 KafkaConstants#PARTITION_KEY보다 우선해요. | | Integer | | producerBatchSize (producer) | 같은 파티션으로 여러 레코드를 보낼 때 producer는 레코드를 묶어서 더 적은 수의 요청으로 만들려고 해요. 클라이언트와 서버 양쪽 성능에 도움이 돼요. 이 설정은 기본 배치 크기를 바이트로 조절해요. 이 크기보다 큰 레코드를 배치하려 하지 않아요. 브로커로 보내는 요청에는 여러 개의 배치가 들어가고, 보낼 데이터가 있는 파티션마다 하나씩 배치가 생겨요. 배치 크기가 작으면 배치가 덜 일어나서 처리량이 줄어들 수 있어요(배치 크기 0은 배치를 완전히 끔). 배치 크기가 아주 크면 추가 레코드를 예상해서 항상 지정한 배치 크기의 버퍼를 할당하기 때문에 메모리를 다소 낭비할 수 있어요. | 16384 | Integer | | queueBufferingMaxMessages (producer) | Deprecated 폐기됨: 이 옵션은 효과가 없어요. 대신 bufferMemorySize나 maxBlockMs를 쓰세요. | 10000 | Integer | | requestRequiredAcks (producer) |

요청이 완료됐다고 치기 전에 producer가 리더에게 받아야 하는 승인(acknowledgment) 개수예요. 이 값이 보내는 레코드의 내구성을 조절해요. 허용되는 설정은 다음과 같아요: acks=0 0으로 설정하면 producer가 서버의 승인을 전혀 기다리지 않아요. 레코드는 즉시 소켓 버퍼에 추가되고 전송된 것으로 쳐요. 이 경우 서버가 레코드를 받았다는 보장이 전혀 없고, 재시도 설정도 효과가 없어요(클라이언트는 대체로 실패를 알 수 없으니까요). 각 레코드에 돌려주는 오프셋은 항상 -1로 설정돼요. acks=1 리더가 레코드를 자기 로컬 로그에 기록하지만 모든 follower의 승인을 기다리지 않고 응답한다는 뜻이에요. 이 경우 리더가 레코드를 승인한 직후, follower가 복제하기 전에 죽으면 레코드가 사라져요. acks=all 리더가 in-sync replica 전원이 레코드를 승인할 때까지 기다린다는 뜻이에요. 최소 한 개의 in-sync replica가 살아 있는 한 레코드가 사라지지 않는다는 게 보장돼요. 가장 강한 보장이에요. acks=-1 설정과 동일해요. idempotence를 켜려면 이 설정값이 'all'이어야 해요. 충돌하는 설정이 있고 idempotence를 명시적으로 켜지 않았다면 꺼져요.

Enum values:

all
-1
0
1

| all | String | | requestTimeoutMs (producer) | 브로커가 클라이언트에 오류를 돌려보내기 전에 request.required.acks 요구사항을 만족하려고 기다리는 시간이에요. | 30000 | Integer | | retries (producer) | 0보다 큰 값을 설정하면 일시적 오류로 전송에 실패한 레코드를 클라이언트가 다시 보내요. 이 재시도는 오류를 받고 클라이언트가 레코드를 다시 보내는 것과 다르지 않아요. 성공 승인 전에 delivery.timeout.ms로 설정한 타임아웃이 먼저 만료되면 재시도 횟수를 다 쓰기 전에 produce 요청이 실패해요. 사용자는 보통 이 설정을 비워 두고 delivery.timeout.ms로 재시도 동작을 조절하는 편이 좋아요. idempotence를 켜려면 이 설정값이 0보다 커야 해요. 충돌하는 설정이 있고 idempotence를 명시적으로 켜지 않았다면 꺼져요. enable.idempotence를 false로, max.in.flight.requests.per.connection을 1로 둔 채 재시도를 허용하면 레코드 순서가 바뀔 수 있어요. 두 배치가 같은 파티션으로 보내졌을 때 첫 배치가 실패해 재시도되는 동안 두 번째 배치가 성공하면, 두 번째 배치의 레코드가 먼저 나타날 수 있거든요. | | Integer | | transacted (producer) | 엔드포인트 id와 라우트 id를 사용해서 transactional.id kafka 프로퍼티를 만들라는 표시예요. transactional.id kafka 프로퍼티나 transactionalId 파라미터가 이미 있으면 이 프로퍼티는 무시돼요. | false | boolean | | transactionalId (producer) | transactional.id 프로퍼티를 설정해서 kafka producer를 트랜잭션 producer로 만들어요. 이 프로퍼티를 쓰면 transacted 파라미터는 무시돼요. | | String | | useIterator (producer) | kafka로 보낼 때 메시지 본문을 단일 레코드로 보낼지, java.util.Iterator를 써서 여러 레코드로 보낼지(메시지 본문을 순회할 수 있다면) 정해요. | true | boolean | | valueSerializer (producer) | 메시지용 serializer 클래스예요. | org.apache.kafka.common.serialization.StringSerializer | String | | workerPool (producer) | KafkaProducer가 비동기 논블로킹 방식으로 보낸 메시지를 kafka 서버가 승인한 뒤 Exchange 라우팅을 계속하기 위해 커스텀 워커 풀을 쓰려면 이걸 지정해요. 이 옵션을 쓰면 더 이상 필요 없을 때 풀을 종료하도록 스레드 풀의 생명주기를 직접 관리해야 해요. | | ExecutorService | | workerPoolCoreSize (producer) | KafkaProducer가 비동기 논블로킹 방식으로 보낸 메시지를 kafka 서버가 승인한 뒤 Exchange 라우팅을 계속하기 위한 워커 풀의 코어 스레드 수예요. | 10 | Integer | | workerPoolMaxSize (producer) | KafkaProducer가 비동기 논블로킹 방식으로 보낸 메시지를 kafka 서버가 승인한 뒤 Exchange 라우팅을 계속하기 위한 워커 풀의 최대 스레드 수예요. | 20 | Integer | | recordMetadata (producer (advanced)) | producer가 Kafka로 보낸 결과의 RecordMetadata를 저장할지 여부예요. 결과는 RecordMetadata 메타데이터를 담은 List에 저장되고, 그 List는 KafkaConstants#KAFKA_RECORD_META 키를 가진 헤더에 저장돼요. | false | boolean | | autowiredEnabled (advanced) | 오토와이어링 사용 여부예요. 레지스트리에서 일치하는 타입의 인스턴스가 하나만 있는지 찾아서 컴포넌트에 자동으로 설정하는(옵션이 autowired로 표시돼 있어야 함) 자동 오토와이어링에 쓰입니다. JDBC 데이터 소스, JMS 커넥션 팩토리, AWS 클라이언트 등을 자동 설정할 때 유용해요. | true | boolean | | kafkaClientFactory (advanced) | Autowired org.apache.kafka.clients.consumer.KafkaConsumer와 org.apache.kafka.clients.producer.KafkaProducer 인스턴스를 만들 때 쓸 팩토리예요. 커스텀 팩토리를 설정해서 기본 Kafka 클라이언트를 확장한 로직으로 인스턴스를 만들 수 있어요. | | KafkaClientFactory | | synchronous (advanced) | 동기 처리를 엄격하게 쓸지 여부를 정해요. | false | boolean | | healthCheckConsumerEnabled (health) | 이 컴포넌트의 consumer 기반 헬스 체크를 모두 켜고 끌 때 사용해요. | true | boolean | | healthCheckProducerEnabled (health) | 이 컴포넌트의 producer 기반 헬스 체크를 모두 켜고 끌 때 사용해요. 참고: Camel은 기본으로 producer 기반 헬스 체크를 모두 꺼 둬요. camel.health.producersEnabled=true로 설정하면 producer 체크를 전역으로 켤 수 있습니다. | true | boolean | | interceptorClasses (monitoring) | producer나 consumer에 인터셉터를 설정해요. producer 인터셉터는 org.apache.kafka.clients.producer.ProducerInterceptor를 구현한 클래스여야 하고, consumer 인터셉터는 org.apache.kafka.clients.consumer.ConsumerInterceptor를 구현한 클래스여야 해요. consumer에 Producer 인터셉터를 쓰면 런타임에 class cast 예외가 던져진다는 점에 유의하세요. | | String | | schemaRegistryURL (schema) | 사용할 스키마 레지스트리 서버의 URL이에요. 형식은 host1:port1,host2:port2예요. multiple Schema registries documentation에서는 schema.registry.url로 알려져 있어요. 이 옵션은 외부(external) 구현에서만 사용할 수 있어요(표준 Apache Kafka에는 없음). | | String | | kerberosBeforeReloginMinTime (security) | 갱신 시도 사이에 로그인 스레드가 자는 시간이에요. | 60000 | Integer | | kerberosConfigLocation (security) | kerberos 설정 파일의 위치예요. | | String | | kerberosInitCmd (security) | Kerberos kinit 명령의 경로예요. 기본값은 /usr/bin/kinit이에요. | /usr/bin/kinit | String | | kerberosPrincipalToLocalRules (security) | principal 이름을 짧은 이름(보통 운영체제 사용자 이름)으로 매핑하는 규칙 목록이에요. 규칙은 순서대로 평가되고, principal 이름에 맞는 첫 번째 규칙이 그 이름을 짧은 이름으로 매핑하는 데 쓰여요. 목록의 뒤쪽 규칙은 무시돼요. 기본적으로 {username}/{hostname}{REALM} 형태의 principal 이름은 {username}으로 매핑돼요. 형식에 대한 자세한 내용은(Apache Kafka 프로젝트 웹사이트의) Security Authorization and ACLs 문서를 보세요. 여러 값은 쉼표로 구분할 수 있어요. | DEFAULT | String | | kerberosRenewJitter (security) | 갱신 시간에 더해지는 무작위 지터의 비율이에요. | 0.05 | Double | | kerberosRenewWindowFactor (security) | 로그인 스레드는 마지막 갱신부터 티켓 만료까지 시간의 지정된 창 계수(window factor)에 도달할 때까지 자다가, 그 시점에 티켓 갱신을 시도해요. | 0.8 | Double | | oauthClientId (security) | OAuth client ID예요. saslAuthType이 OAUTH로 설정됐을 때 써요. | | String | | oauthClientSecret (security) | OAuth client secret이에요. saslAuthType이 OAUTH로 설정됐을 때 써요. | | String | | oauthScope (security) | OAuth scope예요. saslAuthType이 OAUTH로 설정됐을 때 써요. | | String | | oauthTokenEndpointUri (security) | OAuth 토큰 엔드포인트 URI예요. saslAuthType이 OAUTH로 설정됐을 때 써요. | | String | | saslAuthType (security) |

간소화된 인증 타입이에요. securityProtocol, saslMechanism, saslJaasConfig를 손으로 하나하나 설정하지 않고도 Kafka 인증을 더 쉽게 구성하게 해 줘요. 값을 설정하면 알맞은 보안 설정이 자동으로 도출돼요. 참고: 이건 선택 사항이에요. 여전히 securityProtocol, saslMechanism, saslJaasConfig 프로퍼티를 명시적으로 쓰는 전통적인 방식도 가능해요.

Enum values:

NONE
PLAIN
SCRAM_SHA_256
SCRAM_SHA_512
SSL
OAUTH
AWS_MSK_IAM
KERBEROS

| | KafkaAuthType | | saslJaasConfig (security) | kafka의 sasl.jaas.config 파라미터를 노출해요. 예: org.apache.kafka.common.security.plain.PlainLoginModule required username=USERNAME password=PASSWORD;. | | String | | saslKerberosServiceName (security) | Kafka가 실행되는 Kerberos principal 이름이에요. Kafka의 JAAS 설정이나 Kafka 설정 어느 쪽에도 정의할 수 있어요. | | String | | saslMechanism (security) | 사용되는 Simple Authentication and Security Layer(SASL) 메커니즘이에요. 유효한 값은 http://www.iana.org/assignments/sasl-mechanisms/sasl-mechanisms.xhtml을 보세요. | GSSAPI | String | | saslPassword (security) | SASL 인증용 비밀번호예요. saslAuthType이 PLAIN, SCRAM_SHA_256, SCRAM_SHA_512로 설정됐을 때 써요. | | String | | saslUsername (security) | SASL 인증용 사용자 이름이에요. saslAuthType이 PLAIN, SCRAM_SHA_256, SCRAM_SHA_512로 설정됐을 때 써요. | | String | | securityProtocol (security) | 브로커와 통신할 때 쓰는 프로토콜이에요. SASL_PLAINTEXT, PLAINTEXT, SASL_SSL, SSL을 지원해요. | PLAINTEXT | String | | sslCipherSuites (security) | 암호 스위트(cipher suites) 목록이에요. TLS나 SSL 네트워크 프로토콜을 쓰는 네트워크 연결의 보안 설정을 협상하는 데 쓰이는, 인증·암호화·MAC·키 교환 알고리즘의 이름 붙은 조합이에요. 기본으로 사용 가능한 모든 암호 스위트를 지원해요. | | String | | sslContextParameters (security) | Camel SSLContextParameters 객체를 사용하는 SSL 설정이에요. 설정돼 있으면 다른 SSL 엔드포인트 파라미터보다 먼저 적용돼요. 참고: Kafka는 keystore를 파일 위치에서만 읽을 수 있어서, KeyStoreParameters.resource 옵션의 위치 앞에 file:을 붙여야 해요. | | SSLContextParameters | | sslEnabledProtocols (security) | SSL 연결에 켜둘 프로토콜 목록이에요. Java 11 이상에서 실행하면 기본값은 TLSv1.2,TLSv1.3이고 그 외에는 TLSv1.2예요. Java 11 기본값에서는 클라이언트와 서버가 둘 다 지원하면 TLSv1.3을 선호하고, 아니면 TLSv1.2로 폴백해요(둘 다 최소한 TLSv1.2는 지원한다고 가정). 이 기본값은 대부분의 경우 문제없어요. SslProtocol의 설정 문서도 함께 보세요. | | String | | sslEndpointAlgorithm (security) | 서버 인증서로 서버 호스트명을 검증할 때 쓰는 엔드포인트 식별 알고리즘이에요. 서버 호스트명 검증을 끄려면 none이나 false를 쓰세요. | https | String | | sslKeymanagerAlgorithm (security) | SSL 연결의 key manager factory가 쓰는 알고리즘이에요. 기본값은 Java Virtual Machine에 설정된 key manager factory 알고리즘이에요. | SunX509 | String | | sslKeyPassword (security) | 키 스토어 파일 안의 개인 키, 또는 sslKeystoreKey에 지정한 PEM 키의 비밀번호예요. 클라이언트에서는 양방향 인증(two-way authentication)을 설정한 경우에만 필요해요. | | String | | sslKeystoreLocation (security) | 키 스토어 파일의 위치예요. 클라이언트에서는 선택 사항이고, 클라이언트의 양방향 인증에 쓸 수 있어요. | | String | | sslKeystorePassword (security) | 키 스토어 파일의 스토어 비밀번호예요. 클라이언트에서는 선택 사항이고 sslKeystoreLocation을 설정했을 때만 필요해요. 키 스토어 비밀번호는 PEM 형식에서는 지원되지 않아요. | | String | | sslKeystoreType (security) | 키 스토어 파일의 파일 형식이에요. 클라이언트에서는 선택 사항이에요. 기본값은 JKS예요. | JKS | String | | sslProtocol (security) | SSLContext를 만들 때 쓰는 SSL 프로토콜이에요. Java 11 이상에서 실행하면 기본값은 TLSv1.3이고 그 외에는 TLSv1.2예요. 이 값은 대부분의 사용 사례에서 문제없어요. 최근 JVM에서 허용되는 값은 TLSv1.2와 TLSv1.3이에요. TLS, TLSv1.1, SSL, SSLv2, SSLv3은 오래된 JVM에서 지원될 수 있지만 알려진 보안 취약점 때문에 쓰지 말라고 권해요. 이 설정과 sslEnabledProtocols의 기본값에서는 서버가 TLSv1.3을 지원하지 않으면 클라이언트가 TLSv1.2로 다운그레이드해요. 이 설정을 TLSv1.2로 두면 sslEnabledProtocols에 TLSv1.3이 있고 서버가 TLSv1.3만 지원하는 경우라도 클라이언트는 TLSv1.3을 쓰지 않아요. | | String | | sslProvider (security) | SSL 연결에 쓰는 security provider의 이름이에요. 기본값은 JVM의 기본 security provider예요. | | String | | sslTrustmanagerAlgorithm (security) | SSL 연결의 trust manager factory가 쓰는 알고리즘이에요. 기본값은 Java Virtual Machine에 설정된 trust manager factory 알고리즘이에요. | PKIX | String | | sslTruststoreLocation (security) | 트러스트 스토어 파일의 위치예요. | | String | | sslTruststorePassword (security) | 트러스트 스토어 파일의 비밀번호예요. 비밀번호를 설정하지 않아도 설정된 트러스트 스토어 파일은 그대로 쓰이지만 무결성 검사는 꺼져요. 트러스트 스토어 비밀번호는 PEM 형식에서는 지원되지 않아요. | | String | | sslTruststoreType (security) | 트러스트 스토어 파일의 파일 형식이에요. 기본값은 JKS예요. | JKS | String | | useGlobalSslContextParameters (security) | 전역 SSL 컨텍스트 파라미터의 사용을 켜요. | false | boolean |

엔드포인트 옵션 (Endpoint Options)

Kafka 엔드포인트는 다음 URI 문법으로 설정해요:

kafka:topic

다음 _path_와 query 파라미터를 쓸 수 있어요:

경로 파라미터 (Path Parameters)

Name	Description	Default	Type
topic (common)	필수 사용할 토픽의 이름이에요. consumer에서는 쉼표로 여러 토픽을 구분할 수 있어요. producer는 하나의 토픽에만 메시지를 보낼 수 있어요.		String

쿼리 파라미터 (Query Parameters)

Name	Description	Default	Type
additionalProperties (common)	kafka consumer나 kafka producer에 Camel 설정에 바로 반영되지 않는 새로운 Kafka 프로퍼티처럼 직접 세팅할 수 없는 값이 있을 때 추가 프로퍼티를 넣어 주는 옵션이에요. 프로퍼티에는 additionalProperties. 접두어를 붙여야 해요. 예: additionalProperties.transactional.id=12345&additionalProperties.schema.registry.url=http://localhost:8811/avro. application.properties 파일에 설정할 때는 camel.component.kafka.additional-properties(Kafka Share 컴포넌트라면 camel.component.kafka-share.additional-properties) 접두어 뒤에 대괄호로 감싼 프로퍼티 이름을 붙여야 해요. 예컨대 delivery.timeout.ms 프로퍼티를 대괄호로 감싸는 식이에요. 여러 값을 넣는 multi-value 옵션이고 접두어는 additionalProperties.예요.		Map
brokers (common)	사용할 Kafka 브로커의 URL이에요. 형식은 host1:port1,host2:port2이고, 이 목록은 브로커 전체의 부분 집합이거나 브로커 부분 집합을 가리키는 VIP일 수도 있어요. Kafka 문서에서는 bootstrap.servers로 알려진 옵션이에요.		String
clientId (common)	client id는 호출 추적을 돕기 위해 매 요청마다 함께 보내는, 사용자가 지정한 문자열이에요. 요청을 보내는 애플리케이션을 논리적으로 식별할 수 있는 값이면 돼요.		String
connectionMaxIdleMs (common)	이 설정이 지정한 밀리초가 지나면 유휴(idle) 연결을 닫아요.	540000	Integer
headerFilterStrategy (common)	Camel 메시지와 주고받는 헤더를 필터링할 커스텀 HeaderFilterStrategy를 쓰려면 이걸 지정해요.		HeaderFilterStrategy
metadataMaxAgeMs (common)	파티션 리더십 변화가 관찰되지 않았더라도 이 밀리초가 지나면 메타데이터를 강제로 새로 고쳐서 새 브로커나 새 파티션을 미리 발견하는 주기예요.	300000	Integer
metricReporters (common)	메트릭 리포터로 쓸 클래스 목록이에요. MetricReporter 인터페이스를 구현하면 새 메트릭이 만들어질 때 알림을 받는 클래스를 끼워 넣을 수 있어요. JMX 통계 등록용 JmxReporter는 항상 포함돼요.		String
metricsSampleWindowMs (common)	메트릭 샘플을 계산하는 시간 창이에요.	30000	Integer
noOfMetricsSample (common)	메트릭 계산에 유지되는 샘플 개수예요.	2	Integer
receiveBufferBytes (common)	데이터를 읽을 때 쓰는 TCP 수신 버퍼(SO_RCVBUF) 크기예요.	65536	Integer
reconnectBackoffMaxMs (common)	접속에 반복적으로 실패한 브로커에 재접속할 때 기다리는 최대 시간(밀리초)이에요. 값을 주면 연속 접속 실패가 이어질 때마다 호스트별 백오프가 이 최댓값까지 지수적으로 늘어나요. 백오프 증가를 계산한 뒤에는 접속 폭주(connection storm)를 막으려고 20%의 무작위 지터를 더해요.	1000	Integer
reconnectBackoffMs (common)	특정 호스트에 재접속을 시도하기 전에 기다리는 시간이에요. 짧은 루프 안에서 같은 호스트에 계속 접속하는 걸 막아 줘요. 이 백오프는 consumer가 브로커에 보내는 모든 요청에 적용돼요.	50	Integer
retryBackoffMaxMs (common)	반복적으로 실패한 브로커에 요청을 재시도할 때 기다리는 최대 시간(밀리초)이에요. 값을 주면 실패한 요청마다 클라이언트별 백오프가 이 최댓값까지 지수적으로 늘어나요. 모든 클라이언트가 재시도 시점에 맞춰 움직이는 걸 막으려고 0.2 배율의 무작위 지터를 백오프에 적용해서, 계산값의 -20%~+20% 범위 안에서 백오프가 결정돼요. retry.backoff.ms를 retry.backoff.max.ms보다 크게 설정하면 지수 증가 없이 retry.backoff.max.ms를 처음부터 고정 백오프로 써요.	1000	Integer
retryBackoffMs (common)	특정 토픽 파티션에 실패한 요청을 재시도하기 전에 기다리는 시간이에요. 일부 장애 상황에서 짧은 루프로 요청을 계속 보내는 걸 막아 줘요. 이 값이 초기 백오프이고, 실패한 요청마다 retry.backoff.max.ms 값까지 지수적으로 늘어나요.	100	Integer
sendBufferBytes (common)	소켓 쓰기 버퍼 크기예요.	131072	Integer
shutdownTimeout (common)	consumer나 producer가 우아하게 종료되고 워커 스레드를 마무리할 때까지 기다리는 밀리초 단위 타임아웃이에요.	30000	int
allowManualCommit (consumer)	KafkaManualCommit을 통한 수동 커밋을 허용할지 여부예요. 이 옵션을 켜면 Exchange 메시지 헤더에 KafkaManualCommit 인스턴스가 저장되고, 사용자가 이 API에 접근해 Kafka consumer로 수동 오프셋 커밋을 할 수 있어요.	false	boolean
autoCommitEnable (consumer)	true면 consumer가 이미 가져온 메시지의 오프셋을 ZooKeeper에 주기적으로 커밋해요. 프로세스가 실패했을 때 이 커밋된 오프셋이 새 consumer가 시작할 위치로 쓰여요.	true	boolean
autoCommitIntervalMs (consumer)	consumer 오프셋을 zookeeper에 커밋하는 주기(밀리초)예요.	5000	Integer
autoOffsetReset (consumer)	커밋된 오프셋이 없거나 커밋된 오프셋이 범위를 벗어났을 때 consumer group이 어디서부터 읽기 시작할지 정해요. 유효한 값은: earliest(가장 이른 사용 가능한 오프셋으로 이동), latest(가장 최신 오프셋으로 이동, 기본값), none(이전 오프셋이 없으면 예외 던짐), by_duration:(ISO-8601 기간을 뒤에 붙임. 예: by_duration:PT5M 또는 by_duration:P1D. Kafka 4.0 이상 필요)이에요.	latest	String
batching (consumer)	처리를 배치(batching)로 할지 스트리밍으로 할지 여부예요. 기본값은 false고, 이때는 스트리밍을 써요. 스트리밍 모드에서는 메시지 본문의 Camel exchange 하나당 kafka 레코드 하나가 처리돼요. 배치 모드에서는 Camel이 많은 kafka 레코드를 모아서 메시지 본문에 List 객체로 묶어 줘요. 배치 모드에서 몇 개의 레코드를 묶을지는 maxPollRecords 옵션으로 정해요.	false	boolean
batchingIntervalMs (consumer)	consumer 배치 모드에서 현재 배치 크기가 maxPollRecords로 정한 최대 크기에 못 미칠 때 배치 완료를 조기에 트리거하기 위한 밀리초 시간을 지정하는 옵션이에요. 트리거는 주어진 간격에 정확히 일어나지 않고 kafka poll 사이에서만 일어날 수 있다는 점에 유의하세요(pollTimeoutMs 옵션 참고). 예컨대 이 값을 10000으로 두면 트리거는 10000 pollTimeoutMs 구간 안에서 일어나요. pollTimeoutMs 기본값이 5000이라서 대략 15초마다 트리거가 일어난다는 뜻이 돼요.		Integer
breakOnFirstError (consumer)	consumer가 exchange를 처리하다 실패했을 때 어떻게 동작할지 조절하는 옵션이에요. false면 consumer가 다음 메시지로 넘어가 계속 처리해요. true면 consumer가 거기서 멈춰요(break out). 기본인 NoopCommitManager를 쓰면 consumer가 오프셋을 커밋하지 않아서 그 메시지를 다시 시도하게 돼요. 이때 consumer는 KafkaManualCommit을 써서 메시지를 처리하는 최선의 방법을 정해야 해요. SyncCommitManager나 AsyncCommitManager를 쓰면 consumer가 실패를 일으킨 메시지의 오프셋으로 되돌아가(seek back) 그 메시지 처리를 다시 시도해요. 다만 매번 실패가 확실한 메시지(예: poison message)라면 같은 메시지를 끝없이 처리하게 될 수 있어요. 그래서 Camel의 에러 핸들러를 쓰는 식으로 이 상황을 다루는 게 권장돼요.	false	boolean
checkCrcs (consumer)	소비한 레코드의 CRC32를 자동으로 검사해요. 메시지가 전송 중(on-the-wire)이나 디스크에서 손상되지 않았다는 걸 보장해 줘요. 이 검사는 약간의 오버헤드를 더하니 극한의 성능을 추구하는 경우엔 끌 수도 있어요.	true	Boolean
commitTimeoutMs (consumer)	동기 커밋이 끝날 때까지 코드가 기다리는 최대 시간(밀리초)이에요.	5000	Long
consumerRequestTimeoutMs (consumer)	클라이언트가 요청 응답을 기다리는 최대 시간을 조절하는 설정이에요. 타임아웃이 지나도 응답이 오지 않으면 필요하면 요청을 다시 보내고, 재시도를 다 썼으면 요청을 실패로 처리해요.	30000	Integer
consumersCount (consumer)	kafka 서버에 접속하는 consumer 수예요. 각 consumer는 들어오는 데이터를 가져와 처리하는 별도 스레드에서 동작해요.	1	int
fetchMaxBytes (consumer)	fetch 요청에 대해 서버가 돌려줄 최대 데이터 양이에요. 절대 상한은 아니어서, fetch의 첫 번째 비어 있지 않은 파티션의 첫 메시지가 이 값보다 크더라도 consumer가 진행할 수 있도록 그 메시지는 그대로 돌려줘요. 브로커가 받아들이는 최대 메시지 크기는 message.max.bytes(브로커 설정)나 max.message.bytes(토픽 설정)로 정의돼요. consumer는 여러 fetch를 병렬로 수행한다는 점에 유의하세요.	52428800	Integer
fetchMinBytes (consumer)	fetch 요청에 대해 서버가 돌려줄 최소 데이터 양이에요. 데이터가 충분하지 않으면 요청은 그만큼 데이터가 쌓일 때까지 기다렸다가 답해요.	1	Integer
fetchWaitMaxMs (consumer)	fetch.min.bytes를 바로 만족할 데이터가 없을 때 서버가 fetch 요청에 답하기 전까지 블록하는 최대 시간이에요.	500	Integer
groupId (consumer)	이 consumer가 속한 consumer 프로세스 그룹을 유일하게 식별하는 문자열이에요. 같은 group id를 설정하면 여러 프로세스가 같은 consumer group의 일부임을 나타낼 수 있어요. consumer에는 필수인 옵션이에요.		String
groupInstanceId (consumer)	사용자가 제공하는 consumer 인스턴스의 고유 식별자예요. 비어 있지 않은 문자열만 허용돼요. 설정하면 consumer가 정적 멤버(static member)로 취급돼서, 이 ID를 가진 인스턴스는 consumer group에서 어떤 시점에도 하나만 허용돼요. 일시적인 사용 불가(예: 프로세스 재시작)로 인한 그룹 리밸런스를 피하려고 더 긴 세션 타임아웃과 조합해 쓸 수 있어요. 설정하지 않으면 consumer가 전통적인 방식대로 동적 멤버로 그룹에 참여해요.		String
groupProtocol (consumer)
쓸 consumer group 프로토콜이에요. classic 프로토콜은 전통적인 파티션 할당·리밸런싱 방식을 써요. consumer 프로토콜은 더 빠르고 효율적인 리밸런싱을 제공하는 새로운 KIP-848 consumer 리밸런스 프로토콜을 켜요. consumer로 설정하면 classic 전용 프로퍼티(heartbeatIntervalMs, sessionTimeoutMs, partitionAssignor)가 consumer 설정에서 자동으로 제외돼요.

Enum values:

classic
consumer

Enum values:

DISCARD
ERROR_HANDLER
RECONNECT
RETRY
STOP

시작할 때 KafkaConsumer가 처음부터 읽을지 끝부터 읽을지 정해요. SeekPolicy.BEGINNING: 처음부터 읽어요. SeekPolicy.END: 끝부터 읽어요.

Enum values:

BEGINNING
END

| | SeekPolicy | | sessionTimeoutMs (consumer) | Kafka의 그룹 관리 기능을 쓸 때 장애를 감지하는 데 쓰는 타임아웃이에요. | 45000 | Integer | | specificAvroReader (consumer) | multiple Schema registries documentation with Avro Deserializers 구현과 함께 쓰는 특정 Avro reader 사용을 켜는 옵션이에요. 이 옵션은 외부(external) 구현에서만 사용할 수 있어요(표준 Apache Kafka에는 없음). | false | boolean | | topicIsPattern (consumer) | 토픽이 패턴(정규표현식)인지 여부예요. 패턴에 맞는 동적인 개수의 토픽을 구독하는 데 쓸 수 있어요. | false | boolean | | valueDeserializer (consumer) | Deserializer 인터페이스를 구현하는 값(value)용 deserializer 클래스예요. | org.apache.kafka.common.serialization.StringDeserializer | String | | bridgeErrorHandler (consumer (advanced)) | 컨슈머를 Camel 라우팅 에러 핸들러(Error Handler)에 브리지할 수 있게 해 줘요. 즉, Camel 컨슈머가 들어오는 메시지를 집어 올리는 동안 등에서 예외가 발생하면(가능한 경우), 그 예외를 메시지로 처리해서 라우팅 에러 핸들러가 다루게 돼요. 중요: 이것은 서드파티 컴포넌트가 예외가 던져졌을 때 Camel에 알릴 수 있도록 허용하는 경우에만 가능해요. 일부 컴포넌트는 내부에서만 처리하기 때문에 bridgeErrorHandler를 쓸 수 없어요. 다른 상황에서는 Camel 컴포넌트를 개선해서 서드파티 컴포넌트에 연결(hook)해 이를 가능하게 만들 수도 있어요. 기본으로 컨슈머는 org.apache.camel.spi.ExceptionHandler로 예외를 처리하는데, WARN 또는 ERROR 레벨로 로그를 남기고 무시해요. | false | boolean | | exceptionHandler (consumer (advanced)) | consumer가 커스텀 ExceptionHandler를 쓰게 해요. bridgeErrorHandler 옵션이 켜져 있으면 이 옵션은 쓰이지 않습니다. 기본적으로 consumer는 예외를 직접 처리하는데, WARN이나 ERROR 레벨로 기록하고 무시해요. | | ExceptionHandler | | exchangePattern (consumer (advanced)) |

consumer가 exchange를 만들 때 exchange 패턴을 정해요.

Enum values:

InOnly
InOut

Enum values:

read_uncommitted
read_committed

| read_uncommitted | String | | kafkaManualCommitFactory (consumer (advanced)) | KafkaManualCommit 인스턴스를 만들 때 쓸 팩토리예요. 커스텀 팩토리를 끼워 넣어서, 기본 구현과 다른 특별한 로직이 필요한 수동 커밋을 할 때 커스텀 KafkaManualCommit 인스턴스를 만들 수 있어요. | | KafkaManualCommitFactory | | batchWithIndividualHeaders (producer) | 이 기능이 켜져 있고 배치의 단일 요소가 Exchange나 Message일 때, producer가 배치 Message를 사용해 값을 결정해서 그 요소 전용의 kafka 헤더 값을 만들어 내요. 평소 동작은 항상 같은 헤더 값을 쓰는 거예요(Iterable이나 Iterator를 담은 부모 Exchange가 그 값을 결정해요). | false | boolean | | bufferMemorySize (producer) | 서버로 보내기를 기다리는 레코드를 버퍼링하는 데 producer가 쓸 수 있는 총 메모리 바이트예요. 레코드가 서버로 전달될 수 있는 속도보다 빨리 쌓이면, producer는 block.on.buffer.full에 지정한 선호에 따라 블록하거나 예외를 던져요. 이 설정은 producer가 쓸 총 메모리와 대략 일치해야 하지만, producer가 쓰는 메모리가 전부 버퍼링에 쓰이는 건 아니라서 엄격한 상한은 아니에요. 압축이 켜져 있으면 압축용과 진행 중인(in-flight) 요청을 유지하는 데 추가 메모리가 좀 더 쓰여요. | 33554432 | Integer | | compressionCodec (producer) |

이 producer가 생성하는 모든 데이터에 사용할 압축 코덱을 지정하게 해 주는 파라미터예요. 유효한 값은 none, gzip, snappy, lz4, zstd예요.

Enum values:

none
gzip
snappy
lz4
zstd

| none | String | | deliveryTimeoutMs (producer) | send() 호출이 반환된 뒤 성공이나 실패를 보고하기까지의 시간 상한이에요. 이 값은 레코드가 전송 전까지 지연되는 총 시간, 브로커의 승인(acknowledgement)을 기다리는 시간(기대한다면), 재시도 가능한 전송 실패에 허용되는 시간을 함께 제한해요. | 120000 | Integer | | enableIdempotence (producer) | 'true'로 설정하면 producer가 각 메시지의 사본이 스트림에 정확히 하나만 기록되도록 보장해요. 'false'면 브로커 장애 등으로 producer가 재시도할 때 재시도한 메시지의 중복이 스트림에 기록될 수 있어요. idempotence를 켜려면 max.in.flight.requests.per.connection이 5 이하(허용되는 어떤 값이든 메시지 순서는 유지)이고, retries가 0보다 커야 하며, acks는 'all'이어야 해요. 충돌하는 설정이 없으면 idempotence는 기본으로 켜져요. 충돌하는 설정이 있고 idempotence를 명시적으로 켜지 않았다면 꺼져요. idempotence를 명시적으로 켰는데 충돌하는 설정이 있으면 ConfigException이 던져져요. | true | boolean | | headerSerializer (producer) | kafka 헤더 값을 직렬화(serialize)할 커스텀 KafkaHeaderSerializer를 쓰려면 이걸 지정해요. | | KafkaHeaderSerializer | | key (producer) | 레코드 키예요(키를 지정하지 않으면 null). 이 옵션이 설정돼 있으면 헤더 KafkaConstants#KEY보다 우선해요. | | String | | keySerializer (producer) | 키용 serializer 클래스예요(아무것도 지정하지 않으면 메시지용과 같은 기본값을 써요). | org.apache.kafka.common.serialization.StringSerializer | String | | lingerMs (producer) | producer는 요청 전송 사이에 도착한 레코드들을 하나의 배치 요청으로 묶어요. 보통은 레코드가 보낼 수 있는 속도보다 빨리 도착하는 부하 상황에서만 일어나요. 하지만 어떤 상황에서는 클라이언트가 적당한 부하에서도 요청 수를 줄이고 싶을 수 있어요. 이 설정은 약간의 인위적 지연을 더해서 그렇게 해 줘요. 즉 레코드를 즉시 보내는 대신, producer는 주어진 지연까지 기다려서 다른 레코드도 함께 보내 배치로 묶일 수 있게 해요. TCP의 Nagle 알고리즘과 비슷하다고 생각하면 돼요. 이 설정은 배치 지연의 상한을 줘요. 어떤 파티션에 batch.size만큼의 레코드가 모이면 이 설정과 무관하게 즉시 보내지만, 그보다 적은 바이트가 쌓여 있으면 지정한 시간만큼 'linger'(대기)하며 더 많은 레코드를 기다려요. 기본값은 0(즉, 지연 없음)이에요. 예컨대 linger.ms=5로 설정하면 보내는 요청 수가 줄어드는 효과가 있지만 부하가 없을 때 보내는 레코드에는 최대 5ms의 지연이 더해져요. | 5 | Integer | | maxBlockMs (producer) | KafkaProducer의 send(), partitionsFor(), initTransactions(), sendOffsetsToTransaction(), commitTransaction(), abortTransaction() 메서드가 얼마나 오래 블록할지 조절하는 설정이에요. send()에 대해서는 이 타임아웃이 메타데이터 fetch와 버퍼 할당을 기다리는 총 시간을 제한해요(사용자가 제공한 serializer나 partitioner에서의 블록은 이 타임아웃에 포함되지 않아요). partitionsFor()에 대해서는 메타데이터를 사용할 수 없을 때 기다리는 시간을 제한해요. 트랜잭션 관련 메서드는 항상 블록하지만, 트랜잭션 코디네이터를 발견하지 못하거나 타임아웃 안에 응답하지 않으면 타임아웃될 수 있어요. | 60000 | Integer | | maxInFlightRequest (producer) | 블록하기 전에 클라이언트가 단일 연결에서 보낼 미확인(unacknowledged) 요청의 최대 개수예요. 이 설정이 1보다 크고 실패한 전송이 있으면 재시도로 인해(즉, retries가 켜져 있으면) 메시지 순서가 바뀔 위험이 있다는 점에 유의하세요. | 5 | Integer | | maxRequestSize (producer) | 요청의 최대 크기예요. 사실상 최대 레코드 크기의 상한이기도 해요. 서버에는 이 값과 다를 수 있는 자체 레코드 크기 상한이 있다는 점에 유의하세요. 이 설정은 producer가 단일 요청으로 보내는 레코드 배치 수를 제한해서 거대한 요청을 보내는 걸 막아요. | 1048576 | Integer | | partitioner (producer) | 메시지를 하위 토픽들 사이에서 파티셔닝할 partitioner 클래스예요. 기본 partitioner는 키의 해시에 기반해요. | | String | | partitionerIgnoreKeys (producer) | 파티션을 계산할 때 메시지 키를 무시할지 여부예요. 이 설정은 partitioner가 설정되지 않았을 때만 효과가 있어요. | false | boolean | | partitionKey (producer) | 레코드를 보낼 파티션이에요(파티션을 지정하지 않으면 null). 이 옵션이 설정돼 있으면 헤더 KafkaConstants#PARTITION_KEY보다 우선해요. | | Integer | | producerBatchSize (producer) | 같은 파티션으로 여러 레코드를 보낼 때 producer는 레코드를 묶어서 더 적은 수의 요청으로 만들려고 해요. 클라이언트와 서버 양쪽 성능에 도움이 돼요. 이 설정은 기본 배치 크기를 바이트로 조절해요. 이 크기보다 큰 레코드를 배치하려 하지 않아요. 브로커로 보내는 요청에는 여러 개의 배치가 들어가고, 보낼 데이터가 있는 파티션마다 하나씩 배치가 생겨요. 배치 크기가 작으면 배치가 덜 일어나서 처리량이 줄어들 수 있어요(배치 크기 0은 배치를 완전히 끔). 배치 크기가 아주 크면 추가 레코드를 예상해서 항상 지정한 배치 크기의 버퍼를 할당하기 때문에 메모리를 다소 낭비할 수 있어요. | 16384 | Integer | | queueBufferingMaxMessages (producer) | Deprecated 폐기됨: 이 옵션은 효과가 없어요. 대신 bufferMemorySize나 maxBlockMs를 쓰세요. | 10000 | Integer | | requestRequiredAcks (producer) |

Enum values:

all
-1
0
1

| all | String | | requestTimeoutMs (producer) | 브로커가 클라이언트에 오류를 돌려보내기 전에 request.required.acks 요구사항을 만족하려고 기다리는 시간이에요. | 30000 | Integer | | retries (producer) | 0보다 큰 값을 설정하면 일시적 오류로 전송에 실패한 레코드를 클라이언트가 다시 보내요. 이 재시도는 오류를 받고 클라이언트가 레코드를 다시 보내는 것과 다르지 않아요. 성공 승인 전에 delivery.timeout.ms로 설정한 타임아웃이 먼저 만료되면 재시도 횟수를 다 쓰기 전에 produce 요청이 실패해요. 사용자는 보통 이 설정을 비워 두고 delivery.timeout.ms로 재시도 동작을 조절하는 편이 좋아요. idempotence를 켜려면 이 설정값이 0보다 커야 해요. 충돌하는 설정이 있고 idempotence를 명시적으로 켜지 않았다면 꺼져요. enable.idempotence를 false로, max.in.flight.requests.per.connection을 1로 둔 채 재시도를 허용하면 레코드 순서가 바뀔 수 있어요. 두 배치가 같은 파티션으로 보내졌을 때 첫 배치가 실패해 재시도되는 동안 두 번째 배치가 성공하면, 두 번째 배치의 레코드가 먼저 나타날 수 있거든요. | | Integer | | transacted (producer) | 엔드포인트 id와 라우트 id를 사용해서 transactional.id kafka 프로퍼티를 만들라는 표시예요. transactional.id kafka 프로퍼티나 transactionalId 파라미터가 이미 있으면 이 프로퍼티는 무시돼요. | false | boolean | | transactionalId (producer) | transactional.id 프로퍼티를 설정해서 kafka producer를 트랜잭션 producer로 만들어요. 이 프로퍼티를 쓰면 transacted 파라미터는 무시돼요. | | String | | useIterator (producer) | kafka로 보낼 때 메시지 본문을 단일 레코드로 보낼지, java.util.Iterator를 써서 여러 레코드로 보낼지(메시지 본문을 순회할 수 있다면) 정해요. | true | boolean | | valueSerializer (producer) | 메시지용 serializer 클래스예요. | org.apache.kafka.common.serialization.StringSerializer | String | | workerPool (producer) | KafkaProducer가 비동기 논블로킹 방식으로 보낸 메시지를 kafka 서버가 승인한 뒤 Exchange 라우팅을 계속하기 위해 커스텀 워커 풀을 쓰려면 이걸 지정해요. 이 옵션을 쓰면 더 이상 필요 없을 때 풀을 종료하도록 스레드 풀의 생명주기를 직접 관리해야 해요. | | ExecutorService | | workerPoolCoreSize (producer) | KafkaProducer가 비동기 논블로킹 방식으로 보낸 메시지를 kafka 서버가 승인한 뒤 Exchange 라우팅을 계속하기 위한 워커 풀의 코어 스레드 수예요. | 10 | Integer | | workerPoolMaxSize (producer) | KafkaProducer가 비동기 논블로킹 방식으로 보낸 메시지를 kafka 서버가 승인한 뒤 Exchange 라우팅을 계속하기 위한 워커 풀의 최대 스레드 수예요. | 20 | Integer | | lazyStartProducer (producer (advanced)) | producer를 lazy(첫 메시지 시점)로 시작할지 여부예요. lazy 시작을 쓰면 producer가 시작 중 실패해서 라우트 시작이 실패하는 상황에서도 CamelContext와 라우트를 시작할 수 있고, 시작 실패 처리를 Camel의 라우팅 에러 핸들러로 넘길 수 있어요. 다만 첫 메시지를 처리하는 시점에 producer를 만들고 시작하는 데 시간이 조금 걸려서 전체 처리 시간이 길어질 수 있다는 점은 유의하세요. | false | boolean | | recordMetadata (producer (advanced)) | producer가 Kafka로 보낸 결과의 RecordMetadata를 저장할지 여부예요. 결과는 RecordMetadata 메타데이터를 담은 List에 저장되고, 그 List는 KafkaConstants#KAFKA_RECORD_META 키를 가진 헤더에 저장돼요. | false | boolean | | kafkaClientFactory (advanced) | org.apache.kafka.clients.consumer.KafkaConsumer와 org.apache.kafka.clients.producer.KafkaProducer 인스턴스를 만들 때 쓸 팩토리예요. 커스텀 팩토리를 설정해서 기본 Kafka 클라이언트를 확장한 로직으로 인스턴스를 만들 수 있어요. | | KafkaClientFactory | | synchronous (advanced) | 동기 처리를 엄격하게 쓸지 여부를 정해요. | false | boolean | | interceptorClasses (monitoring) | producer나 consumer에 인터셉터를 설정해요. producer 인터셉터는 org.apache.kafka.clients.producer.ProducerInterceptor를 구현한 클래스여야 하고, consumer 인터셉터는 org.apache.kafka.clients.consumer.ConsumerInterceptor를 구현한 클래스여야 해요. consumer에 Producer 인터셉터를 쓰면 런타임에 class cast 예외가 던져진다는 점에 유의하세요. | | String | | schemaRegistryURL (schema) | 사용할 스키마 레지스트리 서버의 URL이에요. 형식은 host1:port1,host2:port2예요. multiple Schema registries documentation에서는 schema.registry.url로 알려져 있어요. 이 옵션은 외부(external) 구현에서만 사용할 수 있어요(표준 Apache Kafka에는 없음). | | String | | kerberosBeforeReloginMinTime (security) | 갱신 시도 사이에 로그인 스레드가 자는 시간이에요. | 60000 | Integer | | kerberosConfigLocation (security) | kerberos 설정 파일의 위치예요. | | String | | kerberosInitCmd (security) | Kerberos kinit 명령의 경로예요. 기본값은 /usr/bin/kinit이에요. | /usr/bin/kinit | String | | kerberosPrincipalToLocalRules (security) | principal 이름을 짧은 이름(보통 운영체제 사용자 이름)으로 매핑하는 규칙 목록이에요. 규칙은 순서대로 평가되고, principal 이름에 맞는 첫 번째 규칙이 그 이름을 짧은 이름으로 매핑하는 데 쓰여요. 목록의 뒤쪽 규칙은 무시돼요. 기본적으로 {username}/{hostname}{REALM} 형태의 principal 이름은 {username}으로 매핑돼요. 형식에 대한 자세한 내용은(Apache Kafka 프로젝트 웹사이트의) Security Authorization and ACLs 문서를 보세요. 여러 값은 쉼표로 구분할 수 있어요. | DEFAULT | String | | kerberosRenewJitter (security) | 갱신 시간에 더해지는 무작위 지터의 비율이에요. | 0.05 | Double | | kerberosRenewWindowFactor (security) | 로그인 스레드는 마지막 갱신부터 티켓 만료까지 시간의 지정된 창 계수(window factor)에 도달할 때까지 자다가, 그 시점에 티켓 갱신을 시도해요. | 0.8 | Double | | oauthClientId (security) | OAuth client ID예요. saslAuthType이 OAUTH로 설정됐을 때 써요. | | String | | oauthClientSecret (security) | OAuth client secret이에요. saslAuthType이 OAUTH로 설정됐을 때 써요. | | String | | oauthScope (security) | OAuth scope예요. saslAuthType이 OAUTH로 설정됐을 때 써요. | | String | | oauthTokenEndpointUri (security) | OAuth 토큰 엔드포인트 URI예요. saslAuthType이 OAUTH로 설정됐을 때 써요. | | String | | saslAuthType (security) |

Enum values:

NONE
PLAIN
SCRAM_SHA_256
SCRAM_SHA_512
SSL
OAUTH
AWS_MSK_IAM
KERBEROS

메시지 헤더 (Message Headers)

Kafka 컴포넌트가 지원하는 메시지 헤더는 아래와 같아요:

Name	Description	Type
CamelKafkaPartitionKey (producer) Constant: `PARTITION_KEY`	파티션을 명시적으로 지정해요.	Integer
CamelKafkaPartition (consumer) Constant: `PARTITION`	메시지가 저장된 파티션이에요.	Integer
CamelKafkaKey (common) Constant: `KEY`	필수 Producer: 관련된 메시지가 모두 같은 파티션으로 가도록 해 주는 메시지의 키예요. Consumer: 설정돼 있다면 메시지의 키예요.	Object
CamelKafkaTopic (consumer) Constant: `TOPIC`	메시지가 유래한 토픽이에요.	String
CamelKafkaOverrideTopic (producer) Constant: `OVERRIDE_TOPIC`	메시지를 보낼 토픽이에요(덮어쓰고 우선함). 이 헤더는 보존되지 않아요.	String
CamelKafkaOffset (consumer) Constant: `OFFSET`	메시지의 오프셋이에요.	Long
CamelKafkaHeaders (consumer) Constant: `HEADERS`	레코드의 헤더들이에요.	Headers
CamelKafkaLastRecordBeforeCommit (consumer) Constant: `LAST_RECORD_BEFORE_COMMIT`	커밋 전의 마지막 레코드인지 여부예요(autoCommitEnable 엔드포인트 파라미터가 false일 때만 제공돼요).	Boolean
CamelKafkaLastPollRecord (consumer) Constant: `LAST_POLL_RECORD`	현재 poll 요청 안에서 마지막 레코드인지 나타내요(autoCommitEnable 엔드포인트 파라미터가 false이거나 allowManualCommit이 true일 때만 제공돼요).	Boolean
CamelKafkaTimestamp (consumer) Constant: `TIMESTAMP`	메시지의 타임스탬프예요.	Long
CamelKafkaOverrideTimestamp (producer) Constant: `OVERRIDE_TIMESTAMP`	ProducerRecord에는 연관된 타임스탬프도 있어요. 사용자가 타임스탬프를 제공하면 producer가 제공된 타임스탬프를 레코드에 찍고, 이 헤더는 보존되지 않아요.	Long
CamelKafkaRecordMeta (producer) Constant: `KAFKA_RECORD_META`	메타데이터예요(recordMetadata 엔드포인트 파라미터가 true일 때만 설정돼요).	List
CamelKafkaManualCommit (consumer) Constant: `MANUAL_COMMIT`	Kafka consumer를 쓸 때 수동 오프셋 커밋을 강제하는 데 쓸 수 있어요.	KafkaManualCommit

Producer/Consumer 설정에 대한 자세한 내용:

http://kafka.apache.org/documentation.html#newconsumerconfigs http://kafka.apache.org/documentation.html#producerconfigs

동적 토픽에 메시지를 보내고 싶다면 CamelKafkaOverrideTopic 헤더를 쓰세요. 이 헤더는 메시지와 함께 전송되지 않는 일회성 헤더로, 실제로 producer에서 제거돼요.

사용법 (Usage)

파티션 수보다 consumer가 더 많고 레코드 하나하나 승인(acknowledgement)을 받는 방식으로, 큐처럼 토픽의 레코드를 소비하려면 Kafka Share 컴포넌트와 함께 share group(KIP-932)을 사용하세요.

consumer 에러 처리 (Consumer error handling)

kafka consumer가 kafka 브로커에서 메시지를 poll하는 동안 오류가 생길 수 있어요. 이 섹션에서는 어떤 일이 일어나는지, 무엇을 설정할 수 있는지 설명해요.

consumer는 Kafka poll API를 호출할 때 예외를 던질 수 있어요. 예컨대 데이터가 잘못돼서 메시지를 역직렬화하지 못하는 경우 등 여러 종류의 오류가 있어요. 이런 오류는 retriable 여부에 따라 나뉘는 KafkaException 형태로 나타나요. 재시도 가능한(RetriableException) 예외는 (그 사이에 poll 타임아웃을 두고) 다시 재시도돼요. 그 외 모든 예외는 pollOnError 설정에 따라 처리돼요. 이 설정은 다음 값을 가질 수 있어요:

DISCARD는 메시지를 버리고 다음 메시지를 계속 poll해요.
ERROR_HANDLER는 Camel의 에러 핸들러로 예외를 처리한 뒤, 그다음 다음 메시지를 계속 poll해요.
RECONNECT는 consumer를 다시 연결하고 그 메시지 poll을 다시 시도해요.
RETRY는 consumer가 같은 메시지 poll을 다시 시도하게 해요.
STOP은 consumer를 멈춰요(다시 메시지를 소비하려면 수동으로 시작/재시작해야 해요).

기본값은 ERROR_HANDLER라서, Camel의 에러 핸들러(설정돼 있다면)가 원인 예외를 처리해요. 그다음 다음 메시지를 계속 poll하고요. 이 동작은 Camel 컴포넌트들이 가진 bridgeErrorHandler 옵션과 비슷해요.

더 세밀한 제어가 필요하면 컴포넌트 레벨에 org.apache.camel.component.kafka.PollExceptionStrategy의 커스텀 구현을 설정할 수 있어요. 그러면 예외마다 어떤 전략을 쓸지 조절할 수 있어요.

consumer 에러 처리 (고급) (Consumer error handling (advanced))

기본적으로 Camel은 ERROR_HANDLER로 예외를 처리하면서 poll해요. 예외로 이어지는 메시지를 Camel이 어떻게 다룰지는 설정의 breakOnFirstError 속성으로 바꿀 수 있어요. 이 값을 쓰면 다음 메시지를 계속 poll하는 대신 오프셋을 커밋해서, 예외를 일으킨 메시지가 재시도되게 해요. 위의 RETRY poll 전략과 비슷한 거예요.

컴포넌트 레벨에서 Java 프로그래밍 방식이나 설정 프로퍼티로 이를 설정할 수 있어요:

자바 전용: 프로그래밍 방식의 컴포넌트 설정

KafkaComponent kafka = new KafkaComponent();
kafka.setBreakOnFirstError(true);
camelContext.addComponent("kafka", kafka);

또는 설정 프로퍼티로:

camel.component.kafka.break-on-first-error=true

아래의 "Using manual commit with Kafka consumer" 섹션을 읽어서, 설정된 CommitManager에 따라 breakOnFirstError가 어떻게 동작하는지 이해하기를 권해요.

Kafka 멱등 리포지터리 (The Kafka idempotent repository)

camel-kafka 라이브러리는 Kafka 토픽 기반 멱등(idempotent) 리포지터리를 제공해요. 이 리포지터리는 멱등 상태 변경(add/remove)을 전부 Kafka 토픽으로 브로드캐스트해서 저장하고, 이벤트 소싱(event sourcing) 방식으로 각 리포지터리 프로세스 인스턴스마다 로컬 인메모리 캐시를 채워요. 쓰이는 토픽은 멱등 리포지터리 인스턴스마다 유일해야 해요. 이 방식은 토픽 파티션 수에 어떤 요구사항도 없어요. 리포지터리가 모든 파티션을 동시에 소비하기 때문이에요. 토픽의 복제 계수(replication factor)에도 요구사항이 없어요. 토픽을 쓰는 각 리포지터리 인스턴스(예: 보통 병렬로 돌아가는 여러 머신)는 자기 consumer group을 각자 관리해요. 그래서 같은 토픽을 쓰는 Camel 프로세스 10개 클러스터에서 각각 자기 오프셋을 관리하죠. 시작할 때 인스턴스는 토픽을 구독하고, 오프셋을 처음으로 되감아 최신 상태까지 캐시를 다시 만들어요. pollDurationMs 길이의 poll 한 번이 0개의 레코드를 돌려줘야 캐시가 워밍업됐다고 칩니다. 캐시가 워밍업되거나 30초가 지나기 전까지는 시작이 완료되지 않아요. 후자가 일어나면 consumer가 토픽 끝까지 따라잡을 때까지 멱등 리포지터리가 불일치 상태일 수 있어요. 유니크 검사에 쓰는 헤더의 형식도 신경 쓰세요. 기본으로 데이터 타입은 String을 써요. 원시 숫자 형식을 쓴다면 그에 맞게 헤더를 역직렬화해야 해요. 예제는 아래 샘플을 확인하세요.

KafkaIdempotentRepository에는 다음 프로퍼티들이 있어요:

Property	Default	Description
`topic`		필수 변경 사항을 브로드캐스트할 Kafka 토픽의 이름이에요. (required)
`bootstrapServers`		필수 내부 Kafka producer와 consumer의 `bootstrap.servers` 프로퍼티예요. `consumerConfig`와 `producerConfig`를 설정하지 않을 때 이걸 축약형으로 쓰세요. 이걸 쓰면 이 컴포넌트가 producer와 consumer에 합리적인 기본 설정을 적용해요.
`groupId`		idempotent consumer에 지정할 groupId예요.
`startupOnly`	`false`	시작할 때만 동기화할지, Camel이 실행되는 동안 계속 동기화할지 여부예요.
`maxCacheSize`	`1000`	메모리에 저장할 최근 사용 키의 개수예요(기본 1000).
`pollDurationMs`	`100`	Kafka consumer의 poll 기간이에요. 로컬 캐시는 즉시 갱신돼요. 이 값은 캐시 액션 메시지를 보낸 idempotent consumer 인스턴스 기준으로, 토픽에서 캐시를 갱신하는 다른 피어들이 얼마나 뒤처질지에 영향을 줘요. 기본값은 100ms예요. 이 값을 명시적으로 설정한다면 원격 캐시의 신선함과 이 리포지터리 consumer와 Kafka 브로커 사이의 네트워크 트래픽 양 사이의 트레이드오프를 감안하세요. 캐시 워밍업 과정은 아무것도 가져오지 않는 poll이 하나 있어야 완료돼요. 그게 스트림이 현재 시점까지 소비됐다는 뜻이거든요. 토픽에서 메시지가 전송되는 비율에 비해 poll 기간이 지나치게 길면 캐시가 워밍업되지 못하고, 따라잡을 때까지 피어들과 불일치한 상태로 동작할 가능성이 있어요.
`producerConfig`		변경 사항을 브로드캐스트하는 Kafka producer가 쓸 프로퍼티를 설정해요. `bootstrapServers`를 덮어쓰므로 Kafka `bootstrap.servers` 프로퍼티를 직접 정의해야 해요.
`consumerConfig`		토픽에서 캐시를 채우는 Kafka consumer가 쓸 프로퍼티를 설정해요. `bootstrapServers`를 덮어쓰므로 Kafka `bootstrap.servers` 프로퍼티를 직접 정의해야 해요.

리포지터리는 topic과 bootstrapServers를 정의해서 만들 수도 있고, SSL/SASL 같은 기능을 쓰려고 producerConfig와 consumerConfig 프로퍼티 세트를 명시적으로 정의할 수도 있어요. 쓰려면 이 리포지터리를 Camel 레지스트리에 넣어야 해요. 직접 넣거나 Spring에서 빈으로 등록하면 되고, CamelContext를 알고 있어야 하거든요.

사용 예는 다음과 같아요:

참고 라우트에서 참조하려면 먼저 KafkaIdempotentRepository 빈을 Camel 레지스트리에 등록해야 해요.

자바 전용: 빈 등록하기

KafkaIdempotentRepository kafkaIdempotentRepository = new KafkaIdempotentRepository("idempotent-db-inserts", "localhost:9091");

SimpleRegistry registry = new SimpleRegistry();
registry.put("insertDbIdemRepo", kafkaIdempotentRepository);

XML: 빈 등록하기

<!-- simple -->
<bean id="insertDbIdemRepo"
  class="org.apache.camel.processor.idempotent.kafka.KafkaIdempotentRepository">
  <property name="topic" value="idempotent-db-inserts"/>
  <property name="bootstrapServers" value="localhost:9091"/>
</bean>

<!-- complex -->
<bean id="insertDbIdemRepo"
  class="org.apache.camel.processor.idempotent.kafka.KafkaIdempotentRepository">
  <property name="topic" value="idempotent-db-inserts"/>
  <property name="maxCacheSize" value="10000"/>
  <property name="consumerConfig">
    <props>
      <prop key="bootstrap.servers">localhost:9091</prop>
    </props>
  </property>
  <property name="producerConfig">
    <props>
      <prop key="bootstrap.servers">localhost:9091</prop>
    </props>
  </property>
</bean>

Java
XML
YAML

from("direct:performInsert")
    .idempotentConsumer(header("id")).idempotentRepository("insertDbIdemRepo")
        .to("sql:INSERT INTO ...")
    .end();

<route>
  <from uri="direct:performInsert"/>
  <idempotentConsumer idempotentRepository="insertDbIdemRepo">
    <header>id</header>
    <to uri="sql:INSERT INTO ..."/>
  </idempotentConsumer>
</route>

- route:
    from:
      uri: direct:performInsert
      steps:
        - idempotentConsumer:
            idempotentRepository: "#insertDbIdemRepo"
            expression:
              header: id
            steps:
              - to:
                  uri: "sql:INSERT INTO ..."

숫자 식별자로 멱등성을 쓸 때는 3가지 방법 중 하나를 고를 수 있어요. 첫 번째는 org.apache.camel.component.kafka.serde.KafkaSerdeHelper의 정적 메서드 numericHeader를 써서 변환을 대신 맡기는 거예요:

자바 전용: numericHeader 헬퍼 메서드

from("direct:performInsert")
    .idempotentConsumer(numericHeader("id")).idempotentRepository("insertDbIdemRepo")
        .to("sql:INSERT INTO ...")
    .end();

또는 라우트 URL로 설정한 커스텀 serializer를 써서 변환을 수행할 수도 있어요:

자바 전용: 커스텀 헤더 deserializer 클래스

public class CustomHeaderDeserializer extends DefaultKafkaHeaderDeserializer {
    private static final Logger LOG = LoggerFactory.getLogger(CustomHeaderDeserializer.class);

    @Override
    public Object deserialize(String key, byte[] value) {
        if (key.equals("id")) {
            BigInteger bi = new BigInteger(value);

            return String.valueOf(bi.longValue());
        } else {
            return super.deserialize(key, value);
        }
    }
}

마지막으로 processor에서 처리할 수도 있어요:

자바 전용: 헤더 타입 변환용 processor

from("kafka:my-topic")
    .process(exchange -> {
        byte[] id = exchange.getIn().getHeader("id", byte[].class);
        BigInteger bi = new BigInteger(id);
        exchange.getIn().setHeader("id", String.valueOf(bi.longValue()));
    })
    .idempotentConsumer(header("id"))
    .idempotentRepository("kafkaIdempotentRepository")
    .to("direct:process");

Kafka consumer로 수동 커밋하기 (Manual commits with the Kafka consumer)

기본적으로 Kafka consumer는 자동 커밋(auto commit)을 써요. 주어진 간격으로 백그라운드에서 오프셋이 자동 커밋되는 방식이에요.

수동 커밋을 강제하고 싶다면, 메시지 헤더에 저장된 Camel Exchange의 KafkaManualCommit API를 쓸 수 있어요. 이렇게 하려면 KafkaComponent나 엔드포인트에 allowManualCommit 옵션을 true로 설정해서 수동 커밋을 켜야 해요. 예컨대:

자바 전용: 프로그래밍 방식의 컴포넌트 설정

KafkaComponent kafka = new KafkaComponent();
kafka.setAutoCommitEnable(false);
kafka.setAllowManualCommit(true);
camelContext.addComponent("kafka", kafka);

또는 설정 프로퍼티로:

camel.component.kafka.auto-commit-enable=false
camel.component.kafka.allow-manual-commit=true

기본으로는 내부적으로 NoopCommitManager를 써요. 오프셋을 커밋하려면 Camel Processor 같은 Java 코드에서 KafkaManualCommit을 써야 해요:

자바 전용: 수동 커밋 processor

public void process(Exchange exchange) {
    KafkaManualCommit manual =
        exchange.getIn().getHeader("CamelKafkaManualCommit", KafkaManualCommit.class);
    manual.commit();
}

팁 헤더 이름은 CamelKafkaManualCommit이에요.

KafkaManualCommit은 동기 커밋을 강제해서, 커밋이 Kafka에서 승인될 때까지 블록하고 실패하면 예외를 던져요. KafkaManualCommitFactory에 DefaultKafkaManualAsyncCommitFactory 구현을 설정하면 비동기 커밋을 쓸 수도 있어요.

그러면 커밋이 kafka 비동기 커밋 API를 써서 다음 consumer 루프에서 수행돼요.

KafkaManualCommit의 커스텀 구현을 쓰고 싶다면, 커스텀 구현의 인스턴스를 만드는 커스텀 KafkaManualCommitFactory를 KafkaComponent에 설정하면 돼요.

consumer가 수동 커밋과 특정 CommitManager를 쓰도록 설정할 때는 이것들이 breakOnFirstError 동작에 어떤 영향을 주는지 이해하는 게 중요해요

자바 전용: 프로그래밍 방식의 컴포넌트 설정

KafkaComponent kafka = new KafkaComponent();
kafka.setAutoCommitEnable(false);
kafka.setAllowManualCommit(true);
kafka.setBreakOnFirstError(true);
kafka.setKafkaManualCommitFactory(new DefaultKafkaManualCommitFactory());
camelContext.addComponent("kafka", kafka);

또는 설정 프로퍼티로:

camel.component.kafka.auto-commit-enable=false
camel.component.kafka.allow-manual-commit=true
camel.component.kafka.break-on-first-error=true
camel.component.kafka.kafka-manual-commit-factory=#class:org.apache.camel.component.kafka.consumer.DefaultKafkaManualCommitFactory

CommitManager를 기본값인 NoopCommitManager로 두면 breakOnFirstError가 오류 난 메시지를 재시도하도록 오프셋을 자동 커밋하지 않아요. consumer가 라우트에서 KafkaManualCommit을 써서 직접 관리해야 해요.

CommitManager를 동기 또는 비동기 매니저로 바꾸면 breakOnFirstError가 오류 난 메시지를 재시도하도록 오프셋을 자동으로 커밋해요. 이 메시지는 오류 없이 처리될 수 있을 때까지 계속 재시도돼요.

Note 1: 파티션의 레코드는 consumer와 같은 스레드에서 처리되고 커밋돼야 해요. 그래서 DSL의 특정 EIP나 비동기·동시성 연산은 커밋을 실패하게 만들 수 있어요. 그런 상황에서 트랜잭션 커밋을 시도하면 Kafka 클라이언트가 KafkaConsumer is not safe for multi-threaded access라는 메시지와 함께 java.util.ConcurrentModificationException 예외를 던져요. 이걸 막으려면 그런 연산을 피하도록 라우트를 다시 설계하세요.

*Note 2: 이건 주로 aggregation의 completion timeout 전략에서 유용해요.

새 consumer group이 읽기 시작하는 위치 (Where a new consumer group starts reading)

autoOffsetReset 옵션은 토픽 파티션에 커밋된 오프셋이 없거나, 커밋된 오프셋이 브로커에 더 이상 없을 때(예: retention에 의해 삭제됨) Kafka consumer group이 어디서부터 읽기 시작할지 조절해요.

중요 autoOffsetReset은 새 컨슈머 그룹이나 리셋된 컨슈머 그룹에만 적용돼요. 그룹에 이미 커밋된 오프셋이 있으면 Kafka는 그 오프셋에서 이어 읽고 이 옵션을 완전히 무시해요. by_duration:PT5M을 설정한다고 기존 컨슈머 그룹이 5분 뒤로 되감기지는 않아요 — 브랜드-new 그룹의 첫 읽기나 오프셋이 만료된 그룹에만 영향을 줘요.

사용할 수 있는 값은 다음과 같아요:

latest (기본값)

consumer가 파티션 끝(가장 최신 오프셋)에서 시작해요. consumer가 처음 접속하기 전에 만들어진 메시지는 건너뛰어요.

earliest

consumer가 파티션 처음에서 시작해요. 남아 있는 모든 메시지를 가장 오래된 사용 가능한 오프셋부터 재생(replay)해요.

none

이전 오프셋이 없으면 예외를 던져요. 실수로 그룹 이름을 바꾸거나 오프셋이 만료된 걸 감지하는 데 쓰세요.

by_duration:<ISO-8601 duration>

(Kafka 4.0+) consumer가 now - duration 이후 타임스탬프를 가진 첫 오프셋에서 시작해요. 예컨대 by_duration:PT1H는 1시간 전부터, by_duration:P1D는 하루 전부터 시작해요. 다른 값들과 마찬가지로 그룹에 커밋된 오프셋이 없을 때만 효과가 있어요.

Java
YAML

// New consumer group starting 1 hour back (only on first run)
from("kafka:orders?brokers=localhost:9092&groupId=reporting&autoOffsetReset=by_duration:PT1H")
    .log("Order received: ${body}");

- route:
    from:
      uri: kafka:orders
      parameters:
        brokers: "localhost:9092"
        groupId: reporting
        autoOffsetReset: "by_duration:PT1H"
      steps:
        - log:
            message: "Order received: ${body}"

참고 기존 컨슈머 그룹을 재시작할 때마다 다시 위치시키고 싶다면 대신 seekTo 옵션(BEGINNING 또는 END)을 쓰세요. autoOffsetReset과 달리 seekTo는 커밋된 오프셋이 있든 없든 consumer 시작 때마다 무조건 적용돼요.

일시정지 가능한 컨슈머 (Pausable Consumers)

Kafka 컴포넌트는 일시정지 가능한(pausable) 컨슈머를 지원해요. 이런 컨슈머는 컴포넌트 자체 외부의 조건, 예컨대 외부 시스템이 사용 불가라든가 하는 일시적 조건에 따라 데이터 소비를 멈출 수 있어요.

자바 전용: 람다로 일시정지 가능한 컨슈머 만들기

from("kafka:topic")
    .pausable(new KafkaConsumerListener(), () -> canContinue())
    .routeId("pausable-route")
    .process(this::process)
    .to("some:destination");

이 예제에서는 canContinue의 결과가 false면 (Kafka의 Consumer pause 메서드를 호출해서) 메시지 소비를 멈출 수 있어요.

중요 pausable EIP는 라우트 어딘가에 예외가 있어서 exchange가 처리되지 못할 때 쓰는 보조 수단이에요. 좀 더 정확히 말하면, pausable EIP가 호출하는 검사는 exchange 처리를 막는 일시적 조건을 확인하는 데 써야 해요.

참고 pausable 컨슈머를 쓸 때는 autoCommitEnable=false와 allowManualCommit=true로 설정해서 오프셋이 성공적인 처리 이후에만 커밋되게 하는 게 권장돼요. autoCommitEnable=true면 Kafka가 poll했지만 아직 처리하지 않은 레코드의 오프셋을 자동 커밋할 수 있어서, pause/resume 주기 동안 메시지가 유실될 수 있어요.

참고 대부분의 사용자는 라우트를 더 잘 제어할 수 있는 RoutePolicy를 쓰는 편이 좋아요.

Kafka 헤더 전파 (Kafka Headers propagation)

Kafka에서 메시지를 소비하면 헤더가 camel exchange 헤더로 자동 전파돼요. 보내는 흐름도 같은 동작을 따라요. 특정 exchange의 camel 헤더가 kafka 메시지 헤더로 전파되죠.

kafka 헤더는 byte[] 값만 허용하기 때문에, camel exchange 헤더가 전파되려면 그 값이 bytes[]로 직렬화돼야 하고, 아니면 헤더는 건너뛰어져요. 다음 헤더 값 타입이 지원돼요: String, Integer, Long, Double, Boolean, byte[]. 참고: kafka에서 camel exchange로 전파되는 모든 헤더는 기본으로 byte[] 값을 담아요. 이 기본 동작을 바꾸려면 from 라우트에는 headerDeserializer, to 라우트에는 headerSerializer URI 파라미터를 설정하면 돼요. 예를 들면:

Java
XML
YAML

from("kafka:my_topic?headerDeserializer=#myDeserializer")
    .to("kafka:my_topic?headerSerializer=#mySerializer");

<route>
  <from uri="kafka:my_topic?headerDeserializer=#myDeserializer"/>
  <to uri="kafka:my_topic?headerSerializer=#mySerializer"/>
</route>

- route:
    from:
      uri: kafka:my_topic
      parameters:
        headerDeserializer: "#myDeserializer"
      steps:
        - to:
            uri: kafka:my_topic
            parameters:
              headerSerializer: "#mySerializer"

기본적으로 모든 헤더는 KafkaHeaderFilterStrategy로 필터링돼요. 이 전략은 Camel이나 org.apache.camel 접두어로 시작하는 헤더를 걸러 내요. 기본 전략은 to와 from 라우트 양쪽에서 headerFilterStrategy URI 파라미터로 바꿀 수 있어요:

Java
XML
YAML

from("kafka:my_topic?headerFilterStrategy=#myStrategy")
    .to("kafka:my_topic?headerFilterStrategy=#myStrategy");

<route>
  <from uri="kafka:my_topic?headerFilterStrategy=#myStrategy"/>
  <to uri="kafka:my_topic?headerFilterStrategy=#myStrategy"/>
</route>

- route:
    from:
      uri: kafka:my_topic
      parameters:
        headerFilterStrategy: "#myStrategy"
      steps:
        - to:
            uri: kafka:my_topic
            parameters:
              headerFilterStrategy: "#myStrategy"

myStrategy 객체는 HeaderFilterStrategy의 하위 클래스여야 하고, CamelContext를 알고 있어야 하므로 직접 넣거나 Spring에서 빈으로 등록해서 Camel 레지스트리에 있어야 해요.

Kafka 트랜잭션 (Kafka Transaction)

producer로 kafka 트랜잭션을 켜려면 transactionalId=<tx-id>나 transacted=true를 추가해야 해요.

Java
XML
YAML

from("direct:transaction")
.to("kafka:my_topic?transacted=true");

<route>
  <from uri="direct:transaction"/>
  <to uri="kafka:my_topic?transacted=true"/>
</route>

- route:
    from:
      uri: direct:transaction
      steps:
        - to:
            uri: kafka:my_topic
            parameters:
              transacted: true

exchange 라우팅이 끝나면 kafka producer는 트랜잭션을 커밋하거나, 예외가 던져졌거나 exchange가 RollbackOnly면 트랜잭션을 중단(abort)해요. Kafka는 멀티스레드 트랜잭션을 지원하지 않아서, 같은 transaction.id를 가진 다른 producer가 트랜잭션 요청을 하면 ProducerFencedException을 던져요.

transacted()를 쓰면 JTA camel-jta와도 동작해요. XA를 지원하는 자원(SQL이나 JMS)이 관련돼 있으면 둘이 짝을 이뤄 동작해서, exchange 라우팅 끝에서 둘 다 커밋하거나 둘 다 롤백해요. 어떤 경우에는 JTA 트랜잭션 매니저가 (2PC 처리 중에) 커밋에 실패하지만 kafka 트랜잭션은 이미 커밋된 상태일 수 있고, kafka 트랜잭션은 JTA/XA 스펙을 지원하지 않으므로 변경 사항을 롤백할 방법이 없어요. 그래서 데이터 정합성에는 여전히 위험이 남아요.

transacted=true 파라미터를 쓰면 transactionalId가 엔드포인트 id와 라우트 id로 만들어져요. 더 세밀한 제어가 필요하면 그냥 transactionalId 파라미터를 쓰면 되지만, kafka producer 엔드포인트마다 각자의 transactionalId를 갖도록 하는 걸 잊지 마세요. 참고로 additionalProperties[transactional.id]를 쓰는 옛 방식도 여전히 유효해요.

transacted=true와 transactionalId가 둘 다 있으면 후자가 우선해요.

Kerberos 설정 파일 지정하기 (Setting Kerberos config file)

'krb5.conf' 파일을 API로 바로 설정할 수 있어요:

자바 전용: 정적 Kerberos 설정

static {
    KafkaComponent.setKerberosConfigLocation("path/to/config/file");
}

또는 JVM 시스템 프로퍼티를 설정할 수도 있어요:

java.security.krb5.conf=/path/to/config/file

Kafka 인증 (Authentication to Kafka)

Kafka는 평문, TLS 위의 PKI(인증서) 등 클라이언트를 서버에 인증하는 여러 방법을 지원해요. 지원되는 메커니즘을 자세히 보려면 Kafka documentation을 참고하세요. kafka 인증과 권한 부여는 JAAS 기반이라서, 클라이언트 쪽에서 JAAS Login Module 구현을 써야 해요.

간소화된 인증 설정 (Simplified Authentication Configuration)

JAAS 설정 문자열을 손으로 만드는 대신, saslAuthType 옵션과 전용 자격 증명 프로퍼티를 쓸 수 있어요. 이 방식은 선택한 인증 타입에 따라 올바른 securityProtocol, saslMechanism, saslJaasConfig 값을 자동으로 도출해 줘요.

다음 인증 타입들이 지원돼요:

saslAuthType	Description
`NONE`	인증 없음(안전하지 않음, 개발 전용)
`PLAIN`	사용자 이름과 비밀번호를 쓰는 SASL/PLAIN
`SCRAM_SHA_256`	SASL/SCRAM-SHA-256 인증
`SCRAM_SHA_512`	SASL/SCRAM-SHA-512 인증
`SSL`	클라이언트 인증서를 쓰는 SSL/TLS (mTLS)
`OAUTH`	OAuth 2.0 / OIDC 인증
`AWS_MSK_IAM`	AWS MSK IAM 인증
`KERBEROS`	Kerberos/GSSAPI 인증

SCRAM 인증 예제

Java
XML
YAML

from("kafka:my-topic?brokers=localhost:9092&saslAuthType=SCRAM_SHA_512&saslUsername=myuser&saslPassword=mypassword")
    .to("log:received");

<route>
  <from uri="kafka:my-topic?brokers=localhost:9092&amp;saslAuthType=SCRAM_SHA_512&amp;saslUsername=myuser&amp;saslPassword=mypassword"/>
  <to uri="log:received"/>
</route>

- route:
    from:
      uri: kafka:my-topic
      parameters:
        brokers: "localhost:9092"
        saslAuthType: SCRAM_SHA_512
        saslUsername: myuser
        saslPassword: mypassword
      steps:
        - to:
            uri: log:received

또는 Spring Boot 설정으로:

camel.component.kafka.sasl-auth-type=SCRAM_SHA_512
camel.component.kafka.sasl-username=myuser
camel.component.kafka.sasl-password=mypassword

OAuth 인증 예제

Java
XML
YAML

from("kafka:my-topic?brokers=localhost:9092&saslAuthType=OAUTH&oauthClientId=my-client&oauthClientSecret=my-secret&oauthTokenEndpointUri=https://auth.example.com/oauth/token")
    .to("log:received");

<route>
  <from uri="kafka:my-topic?brokers=localhost:9092&amp;saslAuthType=OAUTH&amp;oauthClientId=my-client&amp;oauthClientSecret=my-secret&amp;oauthTokenEndpointUri=https://auth.example.com/oauth/token"/>
  <to uri="log:received"/>
</route>

- route:
    from:
      uri: kafka:my-topic
      parameters:
        brokers: "localhost:9092"
        saslAuthType: OAUTH
        oauthClientId: my-client
        oauthClientSecret: my-secret
        oauthTokenEndpointUri: "https://auth.example.com/oauth/token"
      steps:
        - to:
            uri: log:received

AWS MSK IAM 인증

AWS MSK를 IAM 인증과 함께 쓸 때는 aws-msk-iam-auth 라이브러리가 클래스패스에 있는지 확인하세요:

Java
XML
YAML

from("kafka:my-topic?brokers=b-1.mycluster.kafka.us-east-1.amazonaws.com:9098&saslAuthType=AWS_MSK_IAM")
    .to("log:received");

<route>
  <from uri="kafka:my-topic?brokers=b-1.mycluster.kafka.us-east-1.amazonaws.com:9098&amp;saslAuthType=AWS_MSK_IAM"/>
  <to uri="log:received"/>
</route>

- route:
    from:
      uri: kafka:my-topic
      parameters:
        brokers: "b-1.mycluster.kafka.us-east-1.amazonaws.com:9098"
        saslAuthType: AWS_MSK_IAM
      steps:
        - to:
            uri: log:received

간소화된 설정은 선택 사항이에요. 더 세밀한 제어가 필요하거나 커스텀 JAAS 로그인 모듈을 쓴다면, 아래처럼 securityProtocol, saslMechanism, saslJaasConfig 프로퍼티를 명시적으로 쓰는 전통적인 방식도 여전히 가능해요.

전통적인 JAAS 설정 (Traditional JAAS Configuration)

이 섹션에서는 Strimzi JAAS Login Module과 평문을 사용하는 인증의 주요 포인트를 정리해요. Strimzi OAuth에는 인증 메커니즘을 세밀하게 조정할 수 있는 프로퍼티가 여럿 있어서, OAuthBearerLoginModule 섹션에 넣으면 돼요.

가장 기본적인 인증 방법은 사용자 이름과 비밀번호로 평문 로그인 모듈을 쓰는 거예요. 이건 안전하지 않다는 점을 유의하고, 완전히 안전한 메커니즘을 원하면 TLS 위의 OAuth를 권해요.

TLS 위의 사용자 이름과 비밀번호

camel.component.kafka.security-protocol = SASL_SSL
camel.component.kafka.sasl-mechanism=PLAIN
camel.component.kafka.sasl-jaas-config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="my_username" \
  password="my_password";

Strimzi OAuth Login Module은 더 안전한 OAuth 메커니즘을 지원하는데, refresh token, 사용자 이름/비밀번호, client secret을 설정할 수 있어요. 클라이언트 보안 설정을 제대로 하려면 Kafka 브로커의 보안 설정을 이해하고 있어야 해요.

client secret을 쓰는 OAuth Bearer Token

camel.component.kafka.security-protocol = SASL_PLAINTEXT
camel.component.kafka.sasl-mechanism = OAUTHBEARER
camel.component.kafka.sasl-jaas-config = org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.client.id="kafka-producer-client" \
  oauth.client.secret="kafka-producer-client-secret" \
  oauth.username.claim="preferred_username" \
  oauth.ssl.truststore.location="docker/certificates/ca-truststore.p12" \
  oauth.ssl.truststore.type="pkcs12" \
  oauth.ssl.truststore.password="changeit" \
  oauth.token.endpoint.uri="https://keycloak:8443/realms/demo/protocol/openid-connect/token" ;
camel.component.kafka.additional-properties[sasl.login.callback.handler.class]=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

refresh token을 쓰는 OAuth Bearer Token

camel.component.kafka.security-protocol = SASL_PLAINTEXT
camel.component.kafka.sasl-mechanism = OAUTHBEARER
camel.component.kafka.sasl-jaas-config = org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.client.id="kafka-producer-client" \
  oauth.refresh.token="my_refresh_token"
  oauth.username.claim="preferred_username" \
  oauth.ssl.truststore.location="docker/certificates/ca-truststore.p12" \
  oauth.ssl.truststore.type="pkcs12" \
  oauth.ssl.truststore.password="changeit" \
  oauth.token.endpoint.uri="https://keycloak:8443/realms/demo/protocol/openid-connect/token" ;
camel.component.kafka.additional-properties[sasl.login.callback.handler.class]=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

배치 컨슈머 (Batching Consumer)

Camel에서 Kafka 배치 컨슈머를 쓰려면 애플리케이션이 batching 설정을 true로 해야 해요.

받은 레코드는 파이프라인에서 쓰는 exchange의 리스트에 저장돼요. 그래서 각 레코드를 따로 커밋하거나, 리스트의 마지막 exchange를 커밋해서 배치 전체를 한 번에 커밋할 수 있어요.

배치 크기는 maxPollRecords 옵션으로 조절해요.

Camel은 배치 크기(maxPollRecords)만큼 레코드가 받아질 때까지 기다려요. 하지만 레코드가 너무 적고 새 레코드가 안 들어오면 아주 오래 걸릴 수 있어요.

배치가 채워질 때까지 너무 오래 블록하는 걸 피하려면 pollTimeoutMs 옵션으로 poll에 타임아웃을 둘 수 있어요. 이 타임아웃은 지정한 시간 동안 새 메시지를 전혀 받지 못했을 때만 트리거돼요. 그래서 예컨대 pollTimeoutMs=10000이면 타임아웃은 10초이고, Camel이 10초 동안 Kafka에서 새 메시지를 받지 못했을 때만 트리거돼요. 메시지를 받으면 타임아웃 트리거는 리셋돼요. 그래서 새 메시지를 천천히 계속 받고 있으면 이 타임아웃은 한동안 트리거되지 않을 수 있어요. 따라서 배치 크기에 도달하려고 더 기다리는 걸 피하려고 배치 완료를 트리거할 간격(밀리초)을 batchingIntervalMs 옵션으로 지정할 수 있어요. 예컨대 batchingIntervalMs=20000으로 설정하면 Camel은 배치 완료를 트리거하기 전까지 최대 pollTimeoutMs + batchingIntervalMs 기다려요. 이 예제에서는 30초가 되죠.

pollTimeoutMs는 크게 설정하지 말아야 해요. 브로커에서 새 메시지를 받는 동안 Kafka가 직접 쓰는 값이라서 그래요. 이 처리 동안 Camel은 활동하지 않고, 이 옵션이 크게 설정돼 있으면 Camel이 배치 타임아웃이나 간격 완료를 미리 트리거할 수 없어요. 그래서 이 값은 기본값으로 두는 게 권장돼요.

배치 헤더 (Batch Headers)

배치를 나르는 exchange는 배치의 모든 레코드가 일치하는 레코드 메타데이터 헤더인 CamelKafkaTopic과 CamelKafkaPartition도 노출해요. 그래서 배치를 나누기 전에 배치 자체에서 이들을 읽을 수 있어요. 배치가 나누어지면 자식 exchange는 각자 자기 레코드의 본문과 헤더를 가지고 가요:

자바 전용: 배치에서 토픽 읽고 각 레코드에서 오프셋 읽기

from("kafka:topic?groupId=myGroup&batching=true&maxPollRecords=10")
    .log("Received a batch of ${body.size} records from ${header.CamelKafkaTopic}")
    .split(body())
        .log("Record ${body} at offset ${header.CamelKafkaOffset}")
    .end();

헤더는 배치의 모든 레코드가 같은 값을 가질 때만 배치에 설정돼요. 배치가 여러 토픽이나 파티션에 걸쳐 있으면 그 헤더는 설정되지 않아요. 배치 전체에 맞는 단일 값이 없거든요. CamelKafkaOffset처럼 원래 레코드마다 다른 헤더는 배치에는 절대 설정되지 않으니, 본문의 개별 exchange에서 읽으세요.

자동 커밋 (Automatic Commits)

기본적으로 Camel은 배치 처리를 할 때 자동 커밋을 써요. 이 경우 Camel은 애플리케이션이 레코드를 성공적으로 처리한 뒤 자동으로 커밋해요.

실패가 있으면 레코드는 커밋되지 않아요.

아래 코드가 이 접근 방식의 예제예요:

자바 전용: 인라인 processor로 배치 처리하기

from("kafka:topic?groupId=myGroup&pollTimeoutMs=1000&batching=true&maxPollRecords=10&autoOffsetReset=earliest")
    .process(e -> {
        final List<?> exchanges = e.getMessage().getBody(List.class);
        if (exchanges == null || exchanges.isEmpty()) {
            return;
        }
        for (Object obj : exchanges) {
            if (obj instanceof Exchange exchange) {
                LOG.info("Processing exchange with body {}", exchange.getMessage().getBody(String.class));
            }
        }
    })
    .to("mock:result");

자동 커밋에서의 에러 처리 (Handling Errors with Automatic Commits)

자동 커밋을 쓰면 처리에 실패했을 때 Camel이 레코드를 커밋하지 않아요. 그래서 레코드가 여러 번 재처리될 위험이 있어요.

실패한 레코드가 처리 진행을 막지 않도록, 적절한 에러 처리 메커니즘과 패턴(예: dead-letter 큐)을 구현하기를 권해요.

아래 코드는 자동 커밋에서 에러를 처리하는 예제예요:

자바 전용: 배치 처리에서 에러 처리하기

onException(IllegalArgumentException.class).process(exchange -> {
    LOG.warn("Failed to process batch {}", exchange.getMessage().getBody());
    LOG.warn("Failed to process due to {}", exchange.getProperty(Exchange.EXCEPTION_CAUGHT, Throwable.class).getMessage());
}).continued(true);

from("kafka:topic?groupId=myGroup&pollTimeoutMs=1000&batching=true&maxPollRecords=10&autoOffsetReset=earliest")
    .process(e -> {
        final List<?> exchanges = e.getMessage().getBody(List.class);
        if (exchanges == null || exchanges.isEmpty()) {
            return;
        }
        for (Object o : exchanges) {
            if (o instanceof Exchange exchange) {
                LOG.info("Processing exchange with body {}", exchange.getMessage().getBody(String.class));
            }
        }
    })
    .to("mock:result");

팁 프로덕션 시나리오에서는 애플리케이션이 실패한 레코드를 별도 토픽(dead-letter 큐)으로 보내거나 실패를 일으킨 조건을 고치는 게 좋아요.

배치 모드에서 첫 에러에서 멈추기 (Break on First Error in Batching Mode)

breakOnFirstError 옵션은 배치 모드에서도 지원돼요. 스트리밍 모드와 같은 에러 처리 동작을 배치 처리에 적용해 줘요.

배치가 켜진 상태에서 breakOnFirstError=true를 설정하면:

자동 커밋 모드: 배치 처리 중 오류가 나면 Camel이 현재 배치 위치를 커밋하고 consumer 재접속을 트리거해요. 그래서 문제의 배치가 다음 poll 주기에 재시도될 수 있어요.
수동 커밋 모드: 오류가 나면 consumer가 재접속하기 전에 (원한다면) 라우트의 에러 핸들러가 수동 커밋을 수행하는 역할을 해요.
재접속: 오류가 breakOnFirstError를 트리거한 뒤 consumer는 자동으로 재접속해서 커밋된 위치부터 poll을 재개해요.

동작은 스트리밍 모드와 일치해서, 에러 처리 패턴이 두 처리 모드에서 비슷하게 동작해요.

자동 커밋 예제:

Java DSL
YAML
Processor Implementation

from("kafka:topic?groupId=myGroup&pollTimeoutMs=1000&batching=true&maxPollRecords=10&breakOnFirstError=true&autoCommitEnable=true")
    .process(e -> {
        final List<?> exchanges = e.getMessage().getBody(List.class);
        // Process batch - any exception will trigger breakOnFirstError behavior
        for (Object exchange : exchanges) {
            // ... processing logic
        }
    }).to("mock:result");

- route:
    from:
      uri: kafka:topic
      parameters:
        groupId: myGroup
        pollTimeoutMs: 1000
        batching: true
        maxPollRecords: 10
        breakOnFirstError: true
        autoCommitEnable: true
      steps:
        - process:
            ref: batchProcessor
        - to:
            uri: mock:result

@Component("batchProcessor")
public class BatchProcessor implements Processor {
    private static final Logger LOG = LoggerFactory.getLogger(BatchProcessor.class);

    @Override
    public void process(Exchange exchange) throws Exception {
        final List<?> exchanges = exchange.getMessage().getBody(List.class);

        if (exchanges == null || exchanges.isEmpty()) {
            return;
        }

        // Process each exchange in the batch
        for (Object obj : exchanges) {
            if (obj instanceof Exchange batchExchange) {
                String body = batchExchange.getMessage().getBody(String.class);
                LOG.info("Processing exchange with body: {}", body);

                // Your business logic here
                // Any exception thrown will trigger breakOnFirstError behavior
                processMessage(body);
            }
        }
    }

    private void processMessage(String message) {
        // Implement your message processing logic
        // Throw exceptions for error conditions that should trigger breakOnFirstError
        if (message.contains("error")) {
            throw new RuntimeException("Processing failed for message: " + message);
        }
    }
}

수동 커밋 예제:

Java DSL
YAML
Processor Implementations

// Error handler that performs manual commit before reconnection
onException(Exception.class)
    .handled(false)
    .process(exchange -> {
        // Commit manually when error occurs
        KafkaManualCommit manual = exchange.getMessage()
            .getHeader("CamelKafkaManualCommit", KafkaManualCommit.class);
        manual.commit();
    });

from("kafka:topic?groupId=myGroup&batching=true&breakOnFirstError=true&autoCommitEnable=false&allowManualCommit=true")
    .process(e -> {
        final List<?> exchanges = e.getMessage().getBody(List.class);
        // Process batch - errors will be handled by the error handler above
        for (Object exchange : exchanges) {
            // ... processing logic
        }
    })
    .process(exchange -> {
        // Manual commit on successful processing
        KafkaManualCommit manual = exchange.getMessage()
            .getHeader("CamelKafkaManualCommit", KafkaManualCommit.class);
        manual.commit();
    })
    .to("mock:result");

- onException:
    exception:
      - java.lang.Exception
    handled:
      constant:
        expression: "false"
    steps:
      - process:
          ref: errorCommitProcessor
- route:
    from:
      uri: kafka:topic
      parameters:
        groupId: myGroup
        batching: true
        breakOnFirstError: true
        autoCommitEnable: false
        allowManualCommit: true
      steps:
        - process:
            ref: batchProcessorManual
        - process:
            ref: successCommitProcessor
        - to:
            uri: mock:result

@Component("batchProcessorManual")
public class BatchProcessorManual implements Processor {
    private static final Logger LOG = LoggerFactory.getLogger(BatchProcessorManual.class);

    @Override
    public void process(Exchange exchange) throws Exception {
        final List<?> exchanges = exchange.getMessage().getBody(List.class);

        if (exchanges == null || exchanges.isEmpty()) {
            return;
        }

        // Process each exchange in the batch
        for (Object obj : exchanges) {
            if (obj instanceof Exchange batchExchange) {
                String body = batchExchange.getMessage().getBody(String.class);
                LOG.info("Processing exchange with body: {}", body);

                // Your business logic here
                // Errors will be handled by the error handler
                processMessage(body);
            }
        }
    }

    private void processMessage(String message) {
        // Implement your message processing logic
        if (message.contains("error")) {
            throw new RuntimeException("Processing failed for message: " + message);
        }
    }
}

@Component("errorCommitProcessor")
public class ErrorCommitProcessor implements Processor {
    private static final Logger LOG = LoggerFactory.getLogger(ErrorCommitProcessor.class);

    @Override
    public void process(Exchange exchange) throws Exception {
        LOG.warn("Error occurred, performing manual commit before reconnection");

        KafkaManualCommit manual = exchange.getMessage()
            .getHeader("CamelKafkaManualCommit", KafkaManualCommit.class);

        if (manual != null) {
            manual.commit();
            LOG.info("Manual commit completed after error");
        }
    }
}

@Component("successCommitProcessor")
public class SuccessCommitProcessor implements Processor {
    private static final Logger LOG = LoggerFactory.getLogger(SuccessCommitProcessor.class);

    @Override
    public void process(Exchange exchange) throws Exception {
        LOG.debug("Batch processed successfully, performing manual commit");

        KafkaManualCommit manual = exchange.getMessage()
            .getHeader("CamelKafkaManualCommit", KafkaManualCommit.class);

        if (manual != null) {
            manual.commit();
            LOG.debug("Manual commit completed after successful processing");
        }
    }
}

수동 커밋 (Manual Commits)

수동 커밋으로 배치 처리를 할 때는 레코드를 커밋하고 잠재적으로 잘못된 레코드의 결과를 다루는 건 애플리케이션의 몫이에요.

아래 코드가 이 접근 방식의 예제예요:

자바 전용: 수동 커밋으로 배치 처리하기

from("kafka:topic?batching=true&allowManualCommit=true&maxPollRecords=100&kafkaManualCommitFactory=#class:org.apache.camel.component.kafka.consumer.DefaultKafkaManualCommitFactory")
    .process(e -> {
        final List<?> exchanges = e.getMessage().getBody(List.class);
        if (exchanges == null || exchanges.isEmpty()) {
            return;
        }
        // Use the last exchange in the list to commit the whole batch
        final Object tmp = exchanges.getLast();
        if (tmp instanceof Exchange exchange) {
            KafkaManualCommit manual =
                    exchange.getMessage().getHeader("CamelKafkaManualCommit", KafkaManualCommit.class);
            LOG.debug("Performing manual commit");
            manual.commit();
            LOG.debug("Done performing manual commit");
        }
    });

긴 polling 타임아웃 다루기 (Dealing with long polling timeouts)

어떤 경우에는 애플리케이션이 polling 과정에 긴 타임아웃을 원할 수 있어요(참고: pollTimeoutMs).

제대로 하려면 먼저 polling 타임아웃보다 큰 최대 polling 간격을 갖춰야 해요(참고: maxPollIntervalMs).

그다음 shutdown 타임아웃을 늘려서 커밋·종료 같은 Kafka 연산이 갑자기 중단되지 않게 하세요. 예컨대:

자바 전용: 프로그래밍 방식의 shutdown 타임아웃 설정

getCamelContext().getShutdownStrategy().setTimeout(10000);

또는 설정 프로퍼티로:

camel.main.shutdown-timeout=10000

커스텀 구독 어댑터 (Custom Subscription Adapters)

복잡한 구독 로직을 가진 애플리케이션은 구독 과정을 다룰 커스텀 빈을 제공할 수 있어요. 그러려면 SubscribeAdapter 인터페이스를 구현해야 해요.

자바 전용: 커스텀 구독 어댑터 클래스

public class CustomSubscribeAdapter implements SubscribeAdapter {
    @Override
    public void subscribe(Consumer<?, ?> consumer, ConsumerRebalanceListener reBalanceListener, TopicInfo topicInfo) {
        if (topicInfo.isPattern()) {
            consumer.subscribe(topicInfo.getPattern(), reBalanceListener);
        } else {
            consumer.subscribe(topicInfo.getTopics(), reBalanceListener);
        }
    }
}

그다음 이름 붙은 빈 인스턴스로 레지스트리에 추가해야 해요:

자바 전용: 빈 등록하기

context.getRegistry().bind("subscribeAdapter", new CustomSubscribeAdapter());

상호 운용성 (Interoperability)

JMS

Kafka와 JMS를 함께 쓸 때는 JMS 헤더를 기대하는 타입으로 변환(coerce)해야 할 수 있어요.

예컨대 JMS 헤더를 가진 메시지를 Kafka에서 소비해서 JMS 브로커로 보낼 때, 그 헤더는 먼저 바이트 배열로 역직렬화돼요. 그러면 camel-jms 컴포넌트가 이 바이트 배열을 자기가 쓰는 특정 타입으로 변환하려고 해요. 하지만 원래 엔드포인트와 코드에서 어떻게 설정했는지에 따라 데이터가 직렬화·역직렬화되는 방식이 달라질 수 있어요. 그래서 바이트 배열의 데이터 타입을 성급히 가정하는 건 불가능해요.

이 문제를 다루려고, Kafka가 JMS 데이터를 JMS 명세대로 역직렬화하게 강제하는 커스텀 헤더 deserializer를 제공해요. 이 방식은 헤더가 camel-jms 컴포넌트에 의해 제대로 해석·처리되게 해 줘요.

가능한 시스템과 엔드포인트 각각의 고유한 복잡성 때문에, 이 deserializer가 모든 시나리오를 다 커버하지 못할 수 있어요. 그래서 각 애플리케이션의 필요에 맞게 수정하거나 적응할 수 있는 모델로 제공돼요.

이 솔루션을 쓰려면 파이프라인 consumer 쪽의 라우트 URI에 headerDeserializer 옵션을 포함하도록 고쳐야 해요. 예를 들면:

Java
XML
YAML

from("kafka:topic?headerDeserializer=#class:org.apache.camel.component.kafka.consumer.support.interop.JMSDeserializer")
    .to("direct:process");

<route>
  <from uri="kafka:topic?headerDeserializer=#class:org.apache.camel.component.kafka.consumer.support.interop.JMSDeserializer"/>
  <to uri="direct:process"/>
</route>

- route:
    from:
      uri: kafka:topic
      parameters:
        headerDeserializer: "#class:org.apache.camel.component.kafka.consumer.support.interop.JMSDeserializer"
      steps:
        - to:
            uri: direct:process

producer 성능 (Producer Performance)

producer가 너무 느리게 동작한다면 보내기 전에 exchange를 모아 두는(aggregate) 걸 고려할 수 있어요.

Java
XML
YAML

from("direct:start")
    .aggregate(constant(true), new GroupedExchangeAggregationStrategy())
        .completionSize(100)
        .completionTimeout(1000)
    .to("kafka:my-topic");

<route>
  <from uri="direct:start"/>
  <aggregate aggregationStrategy="#groupedExchange" completionSize="100" completionTimeout="1000">
    <correlationExpression>
      <constant>true</constant>
    </correlationExpression>
    <to uri="kafka:my-topic"/>
  </aggregate>
</route>

- route:
    from:
      uri: direct:start
      steps:
        - aggregate:
            aggregationStrategy: "#groupedExchange"
            completionSize: 100
            completionTimeout: 1000
            correlationExpression:
              constant: "true"
            steps:
              - to:
                  uri: kafka:my-topic

팁 XML과 YAML에서는 GroupedExchangeAggregationStrategy를 groupedExchange라는 이름의 빈으로 레지스트리에 등록하세요.

이유는 producer가 두 가지 다른 경우를 다루는 방식과 관련 있어요:

aggregrate를 쓰면 배치 크기 청크로 메시지를 반비동기(semi-asynchronously)로 처리해요. 즉 배치의 모든 메시지를 보낸 다음 그 승인을 기다려요.
그렇지 않으면 동기로 보내서 exchange마다 레코드 메타데이터 fetch에서 블록될 수 있어요.

참고 이 접근 방식의 단점은 진행 중인(in-flight) exchange 수가 늘어나고, 그와 관련된 (작고 드물긴 하지만) 잠재적 위험이 생긴다는 거예요.

예제 (Examples)

Kafka에서 메시지 소비하기 (Consuming messages from Kafka)

Kafka에서 메시지를 읽는 데 필요한 최소 라우트는 다음과 같아요.

Java
XML
YAML

from("kafka:test?brokers=localhost:9092")
    .log("Message received from Kafka : ${body}")
    .log("    on the topic ${headers[CamelKafkaTopic]}")
    .log("    on the partition ${headers[CamelKafkaPartition]}")
    .log("    with the offset ${headers[CamelKafkaOffset]}")
    .log("    with the key ${headers[CamelKafkaKey]}")

<route>
  <from uri="kafka:test?brokers=localhost:9092"/>
  <log message="Message received from Kafka : ${body}"/>
  <log message="    on the topic ${headers[CamelKafkaTopic]}"/>
  <log message="    on the partition ${headers[CamelKafkaPartition]}"/>
  <log message="    with the offset ${headers[CamelKafkaOffset]}"/>
  <log message="    with the key ${headers[CamelKafkaKey]}"/>
</route>

- route:
    from:
      uri: kafka:test
      parameters:
        brokers: "localhost:9092"
      steps:
        - log: "Message received from Kafka : ${body}"
        - log: "    on the topic ${headers[CamelKafkaTopic]}"
        - log: "    on the partition ${headers[CamelKafkaPartition]}"
        - log: "    with the offset ${headers[CamelKafkaOffset]}"
        - log: "    with the key ${headers[CamelKafkaKey]}"

여러 토픽에서 메시지를 소비해야 한다면 토픽 이름을 쉼표로 구분한 목록을 쓸 수 있어요.

Java
XML
YAML

from("kafka:test,test1,test2?brokers=localhost:9092")
    .log("Message received from Kafka : ${body}")
    .log("    on the topic ${headers[CamelKafkaTopic]}")
    .log("    on the partition ${headers[CamelKafkaPartition]}")
    .log("    with the offset ${headers[CamelKafkaOffset]}")
    .log("    with the key ${headers[CamelKafkaKey]}")

<route>
  <from uri="kafka:test,test1,test2?brokers=localhost:9092"/>
  <log message="Message received from Kafka : ${body}"/>
  <log message="    on the topic ${headers[CamelKafkaTopic]}"/>
  <log message="    on the partition ${headers[CamelKafkaPartition]}"/>
  <log message="    with the offset ${headers[CamelKafkaOffset]}"/>
  <log message="    with the key ${headers[CamelKafkaKey]}"/>
</route>

- route:
    from:
      uri: "kafka:test,test1,test2"
      parameters:
        brokers: "localhost:9092"
      steps:
        - log: "Message received from Kafka : ${body}"
        - log: "    on the topic ${headers[CamelKafkaTopic]}"
        - log: "    on the partition ${headers[CamelKafkaPartition]}"
        - log: "    with the offset ${headers[CamelKafkaOffset]}"
        - log: "    with the key ${headers[CamelKafkaKey]}"

토픽 이름으로 패턴을 주고 topicIsPattern 옵션을 써서 여러 토픽을 구독할 수도 있어요.

Java
XML
YAML

from("kafka:test.*?brokers=localhost:9092&topicIsPattern=true")
    .log("Message received from Kafka : ${body}")
    .log("    on the topic ${headers[CamelKafkaTopic]}")
    .log("    on the partition ${headers[CamelKafkaPartition]}")
    .log("    with the offset ${headers[CamelKafkaOffset]}")
    .log("    with the key ${headers[CamelKafkaKey]}")

<route>
  <from uri="kafka:test.*?brokers=localhost:9092&amp;topicIsPattern=true"/>
  <log message="Message received from Kafka : ${body}"/>
  <log message="    on the topic ${headers[CamelKafkaTopic]}"/>
  <log message="    on the partition ${headers[CamelKafkaPartition]}"/>
  <log message="    with the offset ${headers[CamelKafkaOffset]}"/>
  <log message="    with the key ${headers[CamelKafkaKey]}"/>
</route>

- route:
    from:
      uri: "kafka:test.*"
      parameters:
        brokers: "localhost:9092"
        topicIsPattern: true
      steps:
        - log: "Message received from Kafka : ${body}"
        - log: "    on the topic ${headers[CamelKafkaTopic]}"
        - log: "    on the partition ${headers[CamelKafkaPartition]}"
        - log: "    with the offset ${headers[CamelKafkaOffset]}"
        - log: "    with the key ${headers[CamelKafkaKey]}"

Kafka에서 메시지를 소비할 때 Kafka에 이 관리를 맡기지 않고 자체 오프셋 관리를 쓸 수 있어요. 오프셋을 유지하려면 컴포넌트에 FileStateRepository 같은 StateRepository 구현이 필요해요. 이 빈은 레지스트리에 있어야 해요. 사용 방법은 다음과 같아요:

참고 라우트에서 참조하려면 먼저 FileStateRepository 빈을 Camel 레지스트리에 등록해야 해요.

자바 전용: 빈 등록하기

FileStateRepository repository = FileStateRepository.fileStateRepository(new File("/path/to/repo.dat"));

// Bind this repository into the Camel registry
Registry registry = createCamelRegistry();
registry.bind("offsetRepo", repository);

Java
XML
YAML

from("kafka:my-topic?brokers=localhost:9092&groupId=A&autoOffsetReset=earliest&offsetRepository=#offsetRepo")
    .to("mock:result");

<route>
  <from uri="kafka:my-topic?brokers=localhost:9092&amp;groupId=A&amp;autoOffsetReset=earliest&amp;offsetRepository=#offsetRepo"/>
  <to uri="mock:result"/>
</route>

- route:
    from:
      uri: kafka:my-topic
      parameters:
        brokers: "localhost:9092"
        groupId: A
        autoOffsetReset: earliest
        offsetRepository: "#offsetRepo"
      steps:
        - to:
            uri: mock:result

Kafka로 메시지 보내기 (Producing messages to Kafka)

Kafka로 메시지를 보내는 데 필요한 최소 라우트는 다음과 같아요.

Java
XML
YAML

from("direct:start")
    .setBody(constant("Message from Camel"))
    .setHeader("CamelKafkaKey", constant("Camel"))
    .to("kafka:test?brokers=localhost:9092");

<route>
  <from uri="direct:start"/>
  <setBody>
    <constant>Message from Camel</constant>
  </setBody>
  <setHeader name="CamelKafkaKey">
    <constant>Camel</constant>
  </setHeader>
  <to uri="kafka:test?brokers=localhost:9092"/>
</route>

- route:
    from:
      uri: direct:start
      steps:
        - setBody:
            constant: "Message from Camel"
        - setHeader:
            name: CamelKafkaKey
            constant: "Camel"
        - to:
            uri: kafka:test
            parameters:
              brokers: "localhost:9092"

팁 헤더 이름은 CamelKafkaKey예요.

SSL 설정 (SSL configuration)

Kafka 컴포넌트에 SSL 통신을 설정하는 방법은 두 가지가 있어요.

첫 번째 방법은 여러 SSL 엔드포인트 파라미터를 쓰는 거예요:

Java
XML
YAML

from("kafka:my-topic?brokers=localhost:9092&groupId=A&sslKeystoreLocation=/path/to/keystore.jks&sslKeystorePassword=changeit&sslKeyPassword=changeit&securityProtocol=SSL")
    .to("mock:result");

<route>
  <from uri="kafka:my-topic?brokers=localhost:9092&amp;groupId=A&amp;sslKeystoreLocation=/path/to/keystore.jks&amp;sslKeystorePassword=changeit&amp;sslKeyPassword=changeit&amp;securityProtocol=SSL"/>
  <to uri="mock:result"/>
</route>

- route:
    from:
      uri: kafka:my-topic
      parameters:
        brokers: "localhost:9092"
        groupId: A
        sslKeystoreLocation: "/path/to/keystore.jks"
        sslKeystorePassword: changeit
        sslKeyPassword: changeit
        securityProtocol: SSL
      steps:
        - to:
            uri: mock:result

두 번째 방법은 sslContextParameters 엔드포인트 파라미터를 쓰는 거예요:

자바 전용: SSLContextParameters 빈 설정

KeyStoreParameters ksp = new KeyStoreParameters();
ksp.setResource("/path/to/keystore.jks");
ksp.setPassword("changeit");
KeyManagersParameters kmp = new KeyManagersParameters();
kmp.setKeyStore(ksp);
kmp.setKeyPassword("changeit");
SSLContextParameters scp = new SSLContextParameters();
scp.setKeyManagers(kmp);

Registry registry = createCamelRegistry();
registry.bind("ssl", scp);

Java
XML
YAML

from("kafka:my-topic?brokers=localhost:9092&groupId=A&sslContextParameters=#ssl&securityProtocol=SSL")
    .to("mock:result");

<route>
  <from uri="kafka:my-topic?brokers=localhost:9092&amp;groupId=A&amp;sslContextParameters=#ssl&amp;securityProtocol=SSL"/>
  <to uri="mock:result"/>
</route>

- route:
    from:
      uri: kafka:my-topic
      parameters:
        brokers: "localhost:9092"
        groupId: A
        sslContextParameters: "#ssl"
        securityProtocol: SSL
      steps:
        - to:
            uri: mock:result

더 알아보기 (Learn more)

브로커 연결 설정은 위의 brokers, securityProtocol, saslAuthType 옵션 설명과 Kafka 공식 문서를 함께 보세요.

Kafka를 큐처럼 쓰고 싶다면 Kafka Share 컴포넌트 문서를 확인하세요.
옵션 설정 방식 일반은 Component DSL과 Property Placeholders 문서를 참고하세요.

Kafka

본문

URI 형식 (URI Format)

옵션 설정 방법 (Configuring Options)

컴포넌트 옵션 설정 (Configuring Component Options)

엔드포인트 옵션 설정 (Configuring Endpoint Options)

컴포넌트 옵션 (Component Options)

엔드포인트 옵션 (Endpoint Options)

경로 파라미터 (Path Parameters)

쿼리 파라미터 (Query Parameters)

메시지 헤더 (Message Headers)

사용법 (Usage)

공유 그룹(큐) (Share groups (queues))

consumer 에러 처리 (Consumer error handling)

consumer 에러 처리 (고급) (Consumer error handling (advanced))

Kafka 멱등 리포지터리 (The Kafka idempotent repository)

Kafka consumer로 수동 커밋하기 (Manual commits with the Kafka consumer)

새 consumer group이 읽기 시작하는 위치 (Where a new consumer group starts reading)

일시정지 가능한 컨슈머 (Pausable Consumers)

Kafka 헤더 전파 (Kafka Headers propagation)

Kafka 트랜잭션 (Kafka Transaction)

Kerberos 설정 파일 지정하기 (Setting Kerberos config file)

Kafka 인증 (Authentication to Kafka)

간소화된 인증 설정 (Simplified Authentication Configuration)

전통적인 JAAS 설정 (Traditional JAAS Configuration)

배치 컨슈머 (Batching Consumer)

배치 헤더 (Batch Headers)

자동 커밋 (Automatic Commits)

자동 커밋에서의 에러 처리 (Handling Errors with Automatic Commits)

배치 모드에서 첫 에러에서 멈추기 (Break on First Error in Batching Mode)

수동 커밋 (Manual Commits)

긴 polling 타임아웃 다루기 (Dealing with long polling timeouts)

커스텀 구독 어댑터 (Custom Subscription Adapters)

상호 운용성 (Interoperability)

JMS

producer 성능 (Producer Performance)

예제 (Examples)

Kafka에서 메시지 소비하기 (Consuming messages from Kafka)

Kafka로 메시지 보내기 (Producing messages to Kafka)

SSL 설정 (SSL configuration)

더 알아보기 (Learn more)

함께 읽는 실무 이야기