Chapter 90. Kafka

Kafka Component
링크 복사

Available as of Camel 2.13

The kafka: component is used for communicating with Apache Kafka message broker.

Maven users will need to add the following dependency to their pom.xml for this component:

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

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

Copy to Clipboard

Toggle word wrap

Camel 2.17 or newer: Scala is no longer used, as we use the Kafka Java client.

Camel 2.16 or older: You must also add a Maven dependency for your chosen Scala library. camel-kafka does not include that dependency, but assumes its provided. For example, to use Scala 2.10.4 add:

<dependency>
    <groupId>org.scala-lang</groupId>
    <artifactId>scala-library</artifactId>
    <version>2.10.4</version>
</dependency>

<dependency>
    <groupId>org.scala-lang</groupId>
    <artifactId>scala-library</artifactId>
    <version>2.10.4</version>
</dependency>

Copy to Clipboard

Toggle word wrap

URI format
링크 복사

kafka:server:port[?options]

kafka:server:port[?options]

Copy to Clipboard

Toggle word wrap

Options
링크 복사

Expand

Property	Default	Description
`zookeeperHost`		The zookeeper host to use
`zookeeperPort`	2181	The zookeeper port to use
`zookeeperConnect`		Camel 2.13.3/2.14.1: If in use, then zookeeperHost/zookeeperPort is not used.
`topic`		The topic to use
`groupId`
`partitioner`
`consumerStreams`	10
`clientId`
`zookeeperSessionTimeoutMs`
`zookeeperConnectionTimeoutMs`
`zookeeperSyncTimeMs`
`consumersCount`	`1`	Camel 2.15.0: The number of consumers that connect to Kafka server.
`batchSize`	`100`	Camel 2.15.0: The `batchSize` that the `BatchingConsumerTask` processes once.
`barrierAwaitTimeoutMs`	`10000`	Camel 2.15.0: If the `BatchingConsumerTask` processes exchange exceed the `batchSize`, it will wait for `barrierAwaitTimeoutMs`.
`bridgeEndpoint`	`false`	Camel 2.16.0: If the `bridgeEndpoint` is `true`, the producer will ignore the topic header setting of the message.

You can append query options to the URI in the following format, ?option=value&option=value&...

Producer Options
링크 복사

Expand

Property	Default	Description
`producerType`	`sync`	Can have the following values: `sync` - send message/batch immediately, and wait until response is received. `async` - queue the message/batch to send. There is a thread per broker (Kafka node) which polls from this queue upon `queueBufferingMaxMs` or `batchNumMessages`.
`compressionCodec`
`compressedTopics`
`messageSendMaxRetries`
`retryBackoffMs`
`topicMetadataRefreshIntervalMs`
`sendBufferBytes`
`requestRequiredAcks`
`requestTimeoutMs`
`queueBufferingMaxMs`
`queueBufferingMaxMessages`
`queueEnqueueTimeoutMs`
`batchNumMessages`
`serializerClass`
`keySerializerClass`

Consumer Options
링크 복사

Expand

Property	Default	Description
`consumerId`
`socketTimeoutMs`
`socketReceiveBufferBytes`
`fetchMessageMaxBytes`
`autoCommitEnable`
`autoCommitIntervalMs`
`queuedMaxMessages`
`rebalanceMaxRetries`
`fetchMinBytes`
`fetchWaitMaxMs`
`rebalanceBackoffMs`
`refreshLeaderBackoffMs`
`autoOffsetReset`
`consumerTimeoutMs`

Samples
링크 복사

Consuming messages:

from("kafka:localhost:9092?topic=test&zookeeperHost=localhost&zookeeperPort=2181&groupId=group1").to("log:input");

from("kafka:localhost:9092?topic=test&zookeeperHost=localhost&zookeeperPort=2181&groupId=group1").to("log:input");

Copy to Clipboard

Toggle word wrap

Producing messages:

See unit tests of camel-kafka for more examples

Endpoints
링크 복사

Camel supports the Message Endpoint pattern using the Endpoint interface. Endpoints are usually created by a Component and Endpoints are usually referred to in the DSL via their URIs.

From an Endpoint you can use the following methods

createProducer() will create a Producer for sending message exchanges to the endpoint
createConsumer() implements the Event Driven Consumer pattern for consuming message exchanges from the endpoint via a Processor when creating a Consumer
createPollingConsumer() implements the Polling Consumer pattern for consuming message exchanges from the endpoint via a PollingConsumer

이 콘텐츠는 선택한 언어로 제공되지 않습니다.

Kafka Component
링크 복사

URI format
링크 복사

Options
링크 복사

Producer Options
링크 복사

Consumer Options
링크 복사

Samples
링크 복사

Endpoints
링크 복사

See Also
링크 복사

자세한 정보

평가판, 구매 및 판매

커뮤니티

Red Hat 문서 정보

보다 포괄적 수용을 위한 오픈 소스 용어 교체

Red Hat 소개

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links

이 콘텐츠는 선택한 언어로 제공되지 않습니다.

Chapter 90. Kafka

Kafka Component링크 복사링크가 클립보드에 복사되었습니다!

URI format링크 복사링크가 클립보드에 복사되었습니다!

Options링크 복사링크가 클립보드에 복사되었습니다!

Producer Options링크 복사링크가 클립보드에 복사되었습니다!

Consumer Options링크 복사링크가 클립보드에 복사되었습니다!

Samples링크 복사링크가 클립보드에 복사되었습니다!

Endpoints링크 복사링크가 클립보드에 복사되었습니다!

See Also링크 복사링크가 클립보드에 복사되었습니다!

자세한 정보

평가판, 구매 및 판매

커뮤니티

Red Hat 문서 정보

보다 포괄적 수용을 위한 오픈 소스 용어 교체

Red Hat 소개

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links

Kafka Component
링크 복사

URI format
링크 복사

Options
링크 복사

Producer Options
링크 복사

Consumer Options
링크 복사

Samples
링크 복사

Endpoints
링크 복사

See Also
링크 복사