Kafka Share

원문 보기 위키 갱신 2026.10.11

Kafka Share 컴포넌트는 Kafka 토픽의 레코드를 share group(KIP-932, Queues for Kafka) 방식으로 소비하는 컴포넌트예요. 여러 consumer가 같은 파티션을 읽고, 레코드마다 개별 승인이 이뤄지는 등 큐처럼 동작하는 게 특징이에요. 오프셋 커밋이나 순서 보장이 필요 없는 워크로드에서 consumer group 대신 쓸 수 있어요.

출처: 문서

본문

Camel 4.23부터 지원

consumer만 지원해요

Kafka Share 컴포넌트는 Apache Kafka 토픽의 레코드를 share group(KIP-932, Queues for Kafka)을 통해 소비해요.

share group의 consumer들은 큐처럼 토픽의 레코드를 처리해요:

여러 consumer가 같은 파티션을 읽어요. 그래서 consumer 수가 파티션 수에 제한되지 않아요.
모든 레코드는 각자 따로 승인(acknowledge)돼요. 실패한 레코드가 뒤의 레코드를 막지 않죠.
브로커는 release된 레코드를, 다른 consumer에게일 수도 있지만, 다시 전달하고 레코드별로 전달 횟수를 셉니다.

share group에는 커밋하거나 seek할 오프셋이 없고, 순서 보장도, 트랜잭션도, 토픽 패턴도 없어요. 레코드를 생산하거나 consumer group으로 순서대로 소비하려면 Kafka 컴포넌트를 사용하세요.

Maven 사용자라면 이 컴포넌트를 쓰려고 pom.xml에 다음 의존성을 추가해야 해요.

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

URI 형식 (URI format)

kafka-share:topic[?options]

topic은 쉼표로 구분한 토픽 목록일 수 있어요. share group의 이름인 groupId 옵션은 필수예요.

옵션 설정 방법 (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 Share 컴포넌트가 지원하는 옵션은 아래와 같아요.

Name	Description	Default	Type
additionalProperties (common)	Camel 설정에 직접 넣을 수 없는 값(예: 아직 Camel 설정에 반영되지 않은 새 Kafka 프로퍼티)을 kafka consumer나 kafka producer에 추가로 설정해요. 프로퍼티는 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 프로퍼티를 대괄호로 감싸는 식이죠. additionalProperties. 접두어를 쓰는 멀티 값 옵션이에요.		Map
brokers (common)	사용할 Kafka 브로커 URL이에요. 형식은 host1:port1,host2:port2이고, 목록은 브로커의 일부분이거나 그 일부분을 가리키는 VIP일 수 있어요. Kafka 문서에서는 bootstrap.servers로 알려져 있어요.		String
clientId (common)	클라이언트 id는 호출 추적을 돕고자 각 요청에 담아 보내는 사용자 지정 문자열이에요. 요청을 보내는 애플리케이션을 논리적으로 식별하는 용도여야 해요.		String
configuration (consumer)	엔드포인트가 재사용할 공통 옵션으로 Kafka share 컴포넌트를 미리 설정할 수 있게 해줘요.		KafkaShareConfiguration
connectionMaxIdleMs (common)	이 설정이 지정한 밀리초 동안 유휴 상태인 커넥션을 닫아요.	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)	반복해서 접속에 실패한 브로커에 재접속할 때 기다릴 최대 시간(밀리초)이에요. 설정하면 호스트별 백오프가 연속 접속 실패마다 지수적으로 늘어나 이 최댓값까지 커져요. 백오프 증가를 계산한 뒤에는 커넥션 스톰을 피하려고 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
bridgeErrorHandler (consumer)	consumer를 Camel 라우팅 에러 핸들러로 브리지해요. Camel consumer가 들어오는 메시지를 집어 올리는 동안 (가능하다면) 발생한 예외가 메시지로 처리되어 라우팅 에러 핸들러에서 다뤄져요. 중요: 3rd party 컴포넌트가 예외가 던져졌을 때 Camel에 알릴 수 있어야만 가능해요. 일부 컴포넌트는 내부에서만 처리하기 때문에 bridgeErrorHandler를 쓸 수 없어요. 다른 상황에서는 향후 릴리스에서 Camel 컴포넌트가 3rd party 컴포넌트에 연결되도록 개선할 수 있어요. 기본적으로 consumer는 org.apache.camel.spi.ExceptionHandler로 예외를 다루며, WARN이나 ERROR 레벨로 로그를 남기고 무시해요.	false	boolean
checkCrcs (consumer)	소비한 레코드의 CRC32를 자동으로 확인해요. 전송 중이나 디스크에서 메시지가 훼손되지 않았음을 보장해요.	true	Boolean
commitMode (consumer)
poll의 승인(acknowledgement)을 브로커에 커밋하는 방식이에요. SYNC는 poll의 레코드들이 처리된 뒤 커밋하고 결과를 기다려서, 커밋 실패가 exception handler로 보고돼요. ASYNC는 기다리지 않고 커밋하고, 커밋 실패는 로그로 남겨요.

Enum values:

SYNC
ASYNC

exchange가 실패하거나 롤백된 레코드를 승인하는 방식이에요. RELEASE는 브로커의 전달 횟수 한도에 도달할 때까지 이 레코드를 이 consumer나 다른 consumer에게 다시 사용할 수 있게 해요. REJECT는 레코드를 버려요. ACCEPT는 레코드를 소비된 것으로 표시합니다.

Enum values:

ACCEPT
RELEASE
REJECT

share consumer가 새 레코드를 poll하는 중 예외를 던지면 어떻게 할지 정해요. DISCARD와 RETRY는 예외를 로그로 남기고 다시 poll해요: kafka 컴포넌트와 달리 poll 자체가 실패했기 때문에 건너뛰거나 재시도할 레코드가 없고, 라우트에서 실패한 레코드는 onFailure로 처리돼요. ERROR_HANDLER는 consumer의 exception handler가 예외를 처리하게 한 뒤 다시 poll해요. RECONNECT는 share consumer를 닫고 새로 만들어요. STOP은 소비를 멈춰요. 인증이나 권한 실패는 항상 소비를 멈춥니다.

Enum values:

DISCARD
ERROR_HANDLER
RECONNECT
RETRY
STOP

share consumer가 레코드를 획득하는 방식이에요. record_limit이면 poll()이 최대 maxPollRecords개의 레코드를 반환해요. batch_optimized면 poll()이 토픽의 배치에 맞추려고 maxPollRecords보다 더 많은 레코드를 반환할 수 있어요.

Enum values:

batch_optimized
record_limit

| batch_optimized | 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 | | pollExceptionStrategy (consumer (advanced)) | Autowired 메시지를 poll하는 동안 Kafka 브로커가 던지는 예외를 어떻게 처리할지 제어하는 커스텀 전략을 consumer에 사용해요. | | PollExceptionStrategy | | autowiredEnabled (advanced) | 오토와이어링을 활성화할지 여부예요. (autowired로 표시된 옵션을) 레지스트리에서 일치하는 타입의 인스턴스가 하나뿐인지 찾아 자동으로 오토와이어링하는 데 쓰여요. JDBC 데이터 소스, JMS 커넥션 팩토리, AWS 클라이언트 등을 자동 설정할 수 있어요. | true | boolean | | kafkaShareClientFactory (advanced) | org.apache.kafka.clients.consumer.KafkaShareConsumer 인스턴스를 만드는 팩토리예요. 순수 Kafka 클라이언트의 로직을 확장한 로직으로 인스턴스를 만드는 커스텀 팩토리를 설정할 수 있어요. | | KafkaShareClientFactory | | healthCheckConsumerEnabled (health) | 이 컴포넌트의 consumer 기반 헬스 체크를 모두 켜거나 끄는 데 쓰여요. | true | boolean | | healthCheckProducerEnabled (health) | 이 컴포넌트의 producer 기반 헬스 체크를 모두 켜거나 끄는 데 쓰여요. 참고: Camel은 기본적으로 producer 기반 헬스 체크를 모두 꺼 두었어요. camel.health.producersEnabled=true로 설정하면 producer 체크를 전역으로 켤 수 있어요. | true | boolean | | schemaRegistryURL (schema) | 사용할 스키마 레지스트리 서버 URL이에요. 형식은 host1:port1,host2:port2예요. 여러 Schema 레지스트리 문서에서는 schema.registry.url로 알려져 있어요. 이 옵션은 외부에서만 제공돼요(표준 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 클라이언트 ID예요. saslAuthType이 OAUTH로 설정됐을 때 쓰여요. | | String | | oauthClientSecret (security) | OAuth 클라이언트 시크릿이에요. saslAuthType이 OAUTH로 설정됐을 때 쓰여요. | | String | | oauthScope (security) | OAuth 스코프예요. 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) | 사용되는 SASL(Simple Authentication and Security Layer) 메커니즘이에요. 유효한 값은 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) | 암호 스위트 목록이에요. TLS나 SSL 네트워크 프로토콜로 네트워크 연결의 보안 설정을 협상할 때 쓰는 인증·암호화·MAC·키 교환 알고리즘의 명명된 조합이에요. 기본적으로 사용 가능한 모든 암호 스위트를 지원해요. | | String | | sslContextParameters (security) | Camel SSLContextParameters 객체를 쓰는 SSL 설정이에요. 설정하면 다른 SSL 엔드포인트 파라미터보다 먼저 적용돼요. 참고: Kafka는 키스토어를 파일 위치에서만 로드할 수 있어서, 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 연결의 키 매니저 팩토리가 쓰는 알고리즘이에요. 기본값은 Java 가상 머신에 설정된 키 매니저 팩토리 알고리즘이에요. | SunX509 | String | | sslKeyPassword (security) | 키 스토어 파일의 개인 키 비밀번호나 sslKeystoreKey에 지정한 PEM 키의 비밀번호예요. 양방향 인증을 설정한 클라이언트에만 필요해요. | | 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 연결에 쓰는 보안 프로바이더 이름이에요. 기본값은 JVM의 기본 보안 프로바이더예요. | | String | | sslTrustmanagerAlgorithm (security) | SSL 연결의 트러스트 매니저 팩토리가 쓰는 알고리즘이에요. 기본값은 Java 가상 머신에 설정된 트러스트 매니저 팩토리 알고리즘이에요. | PKIX | String | | sslTruststoreLocation (security) | 트러스트 스토어 파일의 위치예요. | | String | | sslTruststorePassword (security) | 트러스트 스토어 파일의 비밀번호예요. 비밀번호를 설정하지 않으면 설정된 트러스트 스토어 파일이 그래도 쓰이지만 무결성 검사는 꺼져요. PEM 형식은 트러스트 스토어 비밀번호를 지원하지 않아요. | | String | | sslTruststoreType (security) | 트러스트 스토어 파일의 파일 형식이에요. 기본값은 JKS예요. | JKS | String | | useGlobalSslContextParameters (security) | 전역 SSL 컨텍스트 파라미터 사용을 활성화해요. | false | boolean |

엔드포인트 옵션 (Endpoint Options)

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

kafka-share:topic

다음 _path_와 query 파라미터를 사용해요:

경로 파라미터 (Path Parameters)

Name	Description	Default	Type
topic (common)	필수 소비할 토픽 이름이에요. 여러 토픽은 쉼표로 구분하세요. 토픽 패턴은 share group이 지원하지 않아요.		String

쿼리 파라미터 (Query Parameters)

Name	Description	Default	Type
additionalProperties (common)	Camel 설정에 직접 넣을 수 없는 값(예: 아직 Camel 설정에 반영되지 않은 새 Kafka 프로퍼티)을 kafka consumer나 kafka producer에 추가로 설정해요. 프로퍼티는 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 프로퍼티를 대괄호로 감싸는 식이죠. additionalProperties. 접두어를 쓰는 멀티 값 옵션이에요.		Map
brokers (common)	사용할 Kafka 브로커 URL이에요. 형식은 host1:port1,host2:port2이고, 목록은 브로커의 일부분이거나 그 일부분을 가리키는 VIP일 수 있어요. Kafka 문서에서는 bootstrap.servers로 알려져 있어요.		String
clientId (common)	클라이언트 id는 호출 추적을 돕고자 각 요청에 담아 보내는 사용자 지정 문자열이에요. 요청을 보내는 애플리케이션을 논리적으로 식별하는 용도여야 해요.		String
connectionMaxIdleMs (common)	이 설정이 지정한 밀리초 동안 유휴 상태인 커넥션을 닫아요.	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)	반복해서 접속에 실패한 브로커에 재접속할 때 기다릴 최대 시간(밀리초)이에요. 설정하면 호스트별 백오프가 연속 접속 실패마다 지수적으로 늘어나 이 최댓값까지 커져요. 백오프 증가를 계산한 뒤에는 커넥션 스톰을 피하려고 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
checkCrcs (consumer)	소비한 레코드의 CRC32를 자동으로 확인해요. 전송 중이나 디스크에서 메시지가 훼손되지 않았음을 보장해요.	true	Boolean
commitMode (consumer)
poll의 승인(acknowledgement)을 브로커에 커밋하는 방식이에요. SYNC는 poll의 레코드들이 처리된 뒤 커밋하고 결과를 기다려서, 커밋 실패가 exception handler로 보고돼요. ASYNC는 기다리지 않고 커밋하고, 커밋 실패는 로그로 남겨요.

Enum values:

SYNC
ASYNC

Enum values:

ACCEPT
RELEASE
REJECT

Enum values:

DISCARD
ERROR_HANDLER
RECONNECT
RETRY
STOP

Enum values:

batch_optimized
record_limit

| batch_optimized | String | | bridgeErrorHandler (consumer (advanced)) | consumer를 Camel 라우팅 에러 핸들러로 브리지해요. Camel consumer가 들어오는 메시지를 집어 올리는 동안 (가능하다면) 발생한 예외가 메시지로 처리되어 라우팅 에러 핸들러에서 다뤄져요. 중요: 3rd party 컴포넌트가 예외가 던져졌을 때 Camel에 알릴 수 있어야만 가능해요. 일부 컴포넌트는 내부에서만 처리하기 때문에 bridgeErrorHandler를 쓸 수 없어요. 다른 상황에서는 향후 릴리스에서 Camel 컴포넌트가 3rd party 컴포넌트에 연결되도록 개선할 수 있어요. 기본적으로 consumer는 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

| | ExchangePattern | | kafkaShareClientFactory (advanced) | org.apache.kafka.clients.consumer.KafkaShareConsumer 인스턴스를 만드는 팩토리예요. 순수 Kafka 클라이언트의 로직을 확장한 로직으로 인스턴스를 만드는 커스텀 팩토리를 설정할 수 있어요. | | KafkaShareClientFactory | | schemaRegistryURL (schema) | 사용할 스키마 레지스트리 서버 URL이에요. 형식은 host1:port1,host2:port2예요. 여러 Schema 레지스트리 문서에서는 schema.registry.url로 알려져 있어요. 이 옵션은 외부에서만 제공돼요(표준 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 클라이언트 ID예요. saslAuthType이 OAUTH로 설정됐을 때 쓰여요. | | String | | oauthClientSecret (security) | OAuth 클라이언트 시크릿이에요. saslAuthType이 OAUTH로 설정됐을 때 쓰여요. | | String | | oauthScope (security) | OAuth 스코프예요. 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 Share 컴포넌트가 지원하는 메시지 헤더는 아래와 같아요:

Name	Description	Type
CamelKafkaTopic (consumer) Constant: `TOPIC`	메시지가 나온 토픽이에요.	String
CamelKafkaPartition (consumer) Constant: `PARTITION`	메시지가 저장된 파티션이에요.	Integer
CamelKafkaOffset (consumer) Constant: `OFFSET`	메시지의 오프셋이에요.	Long
CamelKafkaKey (consumer) Constant: `KEY`	설정했다면 메시지의 키예요.	Object
CamelKafkaTimestamp (consumer) Constant: `TIMESTAMP`	메시지의 타임스탬프예요.	Long
CamelKafkaHeaders (consumer) Constant: `HEADERS`	레코드 헤더예요.	Headers
CamelKafkaShareDeliveryCount (consumer) Constant: `DELIVERY_COUNT`	이 전달을 포함해 레코드가 몇 번 전달됐는지예요. 브로커가 전달을 세지 않으면 이 헤더는 설정되지 않아요.	Short
CamelKafkaShareAcknowledge (consumer) Constant: `ACKNOWLEDGE`
exchange 결과에서 도출하는 대신 레코드를 승인하는 방식(ACCEPT, RELEASE, REJECT)을 정해요. 라우트에서 설정하는데, 예를 들어 너무 많이 전달된 레코드를 REJECT하는 데 쓸 수 있어요.

Enum values:

ACCEPT
RELEASE
REJECT

| | String |

사용 방법 (Usage)

브로커 요구 사항 (Broker requirements)

share group에는 클러스터의 share.version 기능이 레벨 1로 필요해요. Kafka 4.2 이상으로 만든 클러스터에는 기본으로 켜져 있고, 오래된 버전에서 업그레이드한 클러스터는 kafka-features.sh --bootstrap-server <broker> upgrade --feature share.version=1로 켤 수 있어요. 브로커가 하나뿐이라면 share.coordinator.state.topic.replication.factor=1과 share.coordinator.state.topic.min.isr=1도 설정하세요. 그렇지 않으면 브로커가 share group의 상태를 담는 내부 토픽을 만들지 못해서 consumer가 레코드를 받지 못해요.

브로커 설정 group.share.delivery.count.limit(기본값 5)은 레코드가 몇 번 전달될 수 있는지 제한하고, group.share.record.lock.duration.ms(기본값 30초)는 브로커가 레코드를 다른 consumer에게 풀어 주기 전에 consumer가 레코드를 쥐고 있을 수 있는 시간을 제한해요.

새 share group은 자기가 만들어진 뒤에 생산된 레코드부터 시작해요. 토픽에 이미 있는 레코드를 소비하려면 group이 처음 소비하기 전에 그룹 설정 share.auto.offset.reset을 earliest로 설정하세요. 예: kafka-configs.sh --entity-type groups --entity-name my-group --alter --add-config share.auto.offset.reset=earliest.

승인 (Acknowledgements)

각 레코드는 exchange가 처리된 뒤, exchange의 결과에 따라 승인(acknowledge)돼요:

Exchange outcome	Acknowledgement
라우트가 `CamelKafkaShareAcknowledge` 헤더를 설정한 경우(`ACCEPT`, `RELEASE` 또는 `REJECT`)	헤더의 값
완료했거나, 라우트가 처리한 예외로 실패한 경우(예: `onException(…).handled(true)`)	`ACCEPT`
실패했거나 롤백된 경우	`onFailure` 옵션, 기본값은 `RELEASE`
consumer가 멈추는 중이라 처리되지 않은 경우	`RELEASE`

ACCEPT

레코드가 처리됐어요: 다시 전달되지 않아요.

RELEASE

레코드가 처리되지 않았어요: 브로커의 전달 횟수 한도에 도달할 때까지 브로커가 이 consumer나 다른 consumer에게 다시 전달해요.

REJECT

레코드를 처리할 수 없어요: 버려지고 다시 전달되지 않아요.

poll의 레코드 승인(acknowledgement)은 처리된 뒤 브로커에 커밋돼요. commitMode=SYNC(기본값)면 consumer가 결과를 기다리고, 커밋 실패는 exception handler로 보고돼요. commitMode=ASYNC면 기다리지 않고 실패를 로그로 남겨요. 승인을 커밋할 수 없으면 브로커가 레코드를 다시 전달해요.

CamelKafkaShareAcknowledge 헤더는 Camel로 시작해요. 그래서 Kafka 레코드에 같은 이름의 헤더가 있으면 헤더 필터 전략이 제거해요: 이 헤더는 라우트만 설정할 수 있죠.

전달 횟수와 poison 레코드 (Delivery count and poison records)

CamelKafkaShareDeliveryCount 헤더는 이 전달을 포함해 레코드가 몇 번 전달됐는지 담고 있어요. 브로커가 release된 레코드를 다시 전달하기 때문에, 라우트는 계속 실패하는 레코드의 재시도를 멈추고 다른 곳으로 옮길 수 있어요:

Java
YAML

errorHandler(noErrorHandler()); // no local retries: the broker delivers the record again, possibly to another pod

onException(InvalidInvoiceException.class)
    .handled(true)                                       // handled -> ACCEPT
    .to("kafka:invoices-invalid?brokers={{kafka.brokers}}");

from("kafka-share:invoices?brokers={{kafka.brokers}}&groupId=invoice-ocr")
    .filter(header(KafkaShareConstants.DELIVERY_COUNT).isGreaterThanOrEqualTo(4))
        .setHeader(KafkaShareConstants.ACKNOWLEDGE, constant("REJECT"))
        .to("kafka:invoices-dlq?brokers={{kafka.brokers}}")
        .stop()
    .end()
    .to("http://ocr-service/extract")
    .to("kafka:invoices-extracted?brokers={{kafka.brokers}}");

- route:
    from:
      uri: kafka-share:invoices
      parameters:
        brokers: "{{kafka.brokers}}"
        groupId: invoice-ocr
      steps:
        - filter:
            simple: "${header.CamelKafkaShareDeliveryCount} >= 4"
            steps:
              - setHeader:
                  name: CamelKafkaShareAcknowledge
                  constant: REJECT
              - to: "kafka:invoices-dlq?brokers={{kafka.brokers}}"
              - stop: {}
        - to: "http://ocr-service/extract"
        - to: "kafka:invoices-extracted?brokers={{kafka.brokers}}"

consumer 수 (Consumers count)

각 consumer(consumersCount)는 Kafka 클라이언트가 스레드 세이프하지 않기 때문에 자기만의 share consumer와 함께 자기 스레드에서 돌아요. consumer group과 달리 파티션 수를 넘는 consumer는 놀지 않아요: 전부 레코드를 받아요.

Java
YAML

// 20 concurrent consumers on a topic with 3 partitions: a slow ticket does not hold up the others
from("kafka-share:support-tickets?brokers={{kafka.brokers}}&groupId=ticket-triage&consumersCount=20")
    .to("langchain4j-chat:triage")
    .to("kafka:tickets-triaged?brokers={{kafka.brokers}}");

- route:
    from:
      uri: kafka-share:support-tickets
      parameters:
        brokers: "{{kafka.brokers}}"
        groupId: ticket-triage
        consumersCount: 20
      steps:
        - to: "langchain4j-chat:triage"
        - to: "kafka:tickets-triaged?brokers={{kafka.brokers}}"

전달 시맨틱 (Delivery semantics)

전달은 at-least-once예요. 레코드가 다시 처리되는 경우는 이래요:

라우트가 레코드를 처리하는 동안 획득 잠금(acquisition lock)이 만료될 때 (group.share.record.lock.duration.ms 참고);
라우트의 부수 효과가 끝난 뒤지만 승인(acknowledgement)이 브로커에 도착하기 전에 JVM이 멈출 때;
승인(acknowledgement)을 커밋할 수 없을 때;
라우트가 일부만 부수 효과를 낸 뒤 레코드를 release할 때.

중복이 문제 되는 곳에서는 Idempotent Consumer를 사용하세요.

Kafka 컴포넌트와의 차이 (Differences from the Kafka component)

Kafka 컴포넌트의 옵션 중 오프셋, consumer group 멤버십, 순서에 관한 것들은 share group에는 적용되지 않아요: seekTo, autoOffsetReset, autoCommitEnable, allowManualCommit, breakOnFirstError, partitionAssignor, groupProtocol, groupInstanceId, topicIsPattern, isolationLevel, offsetRepository, batching이 그렇죠. share consumer는 일시정지(pause)될 수 없어서, 라우트를 suspend하면 consumer가 멈춰요. brokers, security, deserializer, header, additional properties 옵션은 Kafka 컴포넌트와 같아요.

share consumer가 거절하는 클라이언트 프로퍼티(auto.offset.reset, enable.auto.commit, group.instance.id, isolation.level, partition.assignment.strategy, interceptor.classes, session.timeout.ms, heartbeat.interval.ms, group.protocol, group.remote.assignor)를 additionalProperties로 설정하면 consumer가 시작에 실패하고, 그 이름을 지목하는 에러가 나와요.

헬스 체크 (Health check)

모든 share consumer가 만들어지고 구독(subscribe)되면 consumer는 준비(ready) 상태예요. share consumer를 만들 수 없으면 헬스 체크가 down으로 보고하는데, 에러와 함께 bootstrap.servers, group.id, client.id, route.id, topic 상세도 알려줘요.

더 알아보기 (Learn more)

share group의 브로커 요구 사항과 승인 동작은 위의 브로커 요구 사항(Broker requirements), 승인(Acknowledgements) 섹션이 핵심이에요. Kafka 컴포넌트와의 차이(Differences from the Kafka component) 섹션에서 share group에 적용되지 않는 옵션 목록을 확인하고, poison 레코드 처리는 전달 횟수와 poison 레코드(Delivery count and poison records) 섹션의 Java·YAML 예제를 참고하세요. KIP-932의 배경은 Kafka 위키의 Queues for Kafka 제안에서 읽을 수 있어요.

Kafka Share

본문

URI 형식 (URI format)

옵션 설정 방법 (Configuring Options)

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

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

컴포넌트 옵션 (Component Options)

엔드포인트 옵션 (Endpoint Options)

경로 파라미터 (Path Parameters)

쿼리 파라미터 (Query Parameters)

메시지 헤더 (Message Headers)

사용 방법 (Usage)

브로커 요구 사항 (Broker requirements)

승인 (Acknowledgements)

전달 횟수와 poison 레코드 (Delivery count and poison records)

consumer 수 (Consumers count)

전달 시맨틱 (Delivery semantics)

Kafka 컴포넌트와의 차이 (Differences from the Kafka component)

헬스 체크 (Health check)

더 알아보기 (Learn more)

함께 읽는 실무 이야기