From df23749a600d1ccb9baea56707c15b43fce990e6 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 7 Jun 2016 15:52:18 -0400 Subject: [PATCH] GH-115: S-I-K Documentation Resolves #115 Also fix some minor PDF formatting issues. Fix typo. --- src/reference/asciidoc/kafka.adoc | 7 +- src/reference/asciidoc/quick-tour.adoc | 13 +- src/reference/asciidoc/si-kafka.adoc | 169 +++++++++++++++++++++++++ 3 files changed, 183 insertions(+), 6 deletions(-) diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 35cefe13..f47f645c 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -146,10 +146,11 @@ The following constructors are available. [source, java] ---- public KafkaMessageListenerContainer(ConsumerFactory consumerFactory, - ContainerProperties containerProperties) + ContainerProperties containerProperties) public KafkaMessageListenerContainer(ConsumerFactory consumerFactory, - ContainerProperties containerProperties, TopicPartitionInitialOffset... topicPartitions) + ContainerProperties containerProperties, + TopicPartitionInitialOffset... topicPartitions) ---- @@ -183,7 +184,7 @@ The single constructor is similar to the first `KafkaListenerContainer` construc [source, java] ---- public ConcurrentMessageListenerContainer(ConsumerFactory consumerFactory, - ContainerProperties containerProperties) + ContainerProperties containerProperties) ---- diff --git a/src/reference/asciidoc/quick-tour.adoc b/src/reference/asciidoc/quick-tour.adoc index 824e1285..69f0785d 100644 --- a/src/reference/asciidoc/quick-tour.adoc +++ b/src/reference/asciidoc/quick-tour.adoc @@ -70,13 +70,17 @@ public void testAutoCommit() throws Exception { logger.info("Stop auto"); } +---- -private KafkaMessageListenerContainer createContainer(ContainerProperties containerProps) { +[source, java] +---- +private KafkaMessageListenerContainer createContainer( + ContainerProperties containerProps) { Map props = consumerProps(); DefaultKafkaConsumerFactory cf = - new DefaultKafkaConsumerFactory(props); + new DefaultKafkaConsumerFactory(props); KafkaMessageListenerContainer container = - new KafkaMessageListenerContainer<>(cf, containerProps); + new KafkaMessageListenerContainer<>(cf, containerProps); return container; } @@ -182,7 +186,10 @@ public class Config { } } +---- +[source, java] +---- public class Listener { private final CountDownLatch latch1 = new CountDownLatch(1); diff --git a/src/reference/asciidoc/si-kafka.adoc b/src/reference/asciidoc/si-kafka.adoc index 00b6e8fe..6c2162f8 100644 --- a/src/reference/asciidoc/si-kafka.adoc +++ b/src/reference/asciidoc/si-kafka.adoc @@ -1,2 +1,171 @@ [[si-kafka]] === Spring Integration Kafka + +==== Introduction + +This documentation pertains to versions 2.0.0 and above; for documentation for earlier releases, see the https://github.com/spring-projects/spring-integration-kafka/blob/1.3.x/README.md[1.3.x README]. + +Spring Integration Kafka is now based on the http://projects.spring.io/spring-kafka/[Spring for Apache Kafka project]. +It provides the following components: + +- Outbound Channel Adapter +- Message-Driven Channel Adapter + +These are discussed in the following sections. + +[[si-outbound]] +==== Outbound Channel Adapter + +The Outbound channel adapter is used to publish messages from a Spring Integration channel to Kafka topics. +The channel is defined in the application context and then wired into the application that sends messages to Kafka. +Sender applications can publish to Kafka via Spring Integration messages, which are internally converted +to Kafka messages by the outbound channel adapter, as follows: the payload of the Spring Integration message will be +used to populate the payload of the Kafka message, and (by default) the `kafka_messageKey` header of the Spring +Integration message will be used to populate the key of the Kafka message. + +The target topic and partition for publishing the message can be customized through the `kafka_topic` +and `kafka_partitionId` headers, respectively. + +In addition, the `` provides the ability to extract the key, target topic, and +target partition by applying SpEL expressions on the outbound message. To that end, it supports the mutually exclusive +pairs of attributes `topic`/`topic-expression`, `message-key`/`message-key-expression`, and +`partition-id`/`partition-id-expression`, to allow the specification of `topic`,`message-key` and `partition-id` +respectively as static values on the adapter, or to dynamically evaluate their values at runtime against +the request message. + +IMPORTANT: The `KafkaHeaders` interface (provided by `spring-kafka`) contains constants used for interacting with +headers. +The `messageKey` and `topic` default headers now require a `kafka_` prefix. +When migrating from an earlier version that used the old headers, you need to specify +`message-key-expression="headers.messageKey"` and `topic-expression="headers.topic"` on the +``, or simply change the headers upstream to +the new headers from `KafkaHeaders` using a `` or `MessageBuilder`. +Or, of course, configure them on the adapter using `topic` and `message-key` if you are using constant values. + +NOTE : If the adapter is configured with a topic or message key (either with a constant or expression), those are used +and the corresponding header is ignored. +If you wish the header to override the configuration, you need to configure it in an expression, such as: + +`topic-expression="headers.topic != null ? headers.topic : 'myTopic'"`. + +The adapter requires a `KafkaTemplate`. + +Here is an example of how the Kafka outbound channel adapter is configured with XML: + +[source, xml] +---- + + + + + + + + + + ... + + + + + +---- + +As you can see, the adapter requires a `KafkaTemplate` which, in turn, requires a suitably configured `KafkaProducerFactory`. + +When using Java Configuration: + +[source, java] +---- +@Bean +@ServiceActivator(inputChannel = "toKafka") +public MessageHandler handler() throws Exception { + KafkaProducerMessageHandler handler = + new KafkaProducerMessageHandler<>(kafkaTemplate()); + handler.setTopicExpression(new LiteralExpression("someTopic")); + handler.setMessageKeyExpression(new LiteralExpression("someKey")); + return handler; +} + +@Bean +public KafkaTemplate kafkaTemplate() { + return new KafkaTemplate<>(producerFactory()); +} + +@Bean +public ProducerFactory producerFactory() { + Map props = new HashMap<>(); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddress); + // set more properties + return new DefaultKafkaProducerFactory<>(props); +} +---- + +[[si-inbound]] +==== Message Driven Channel Adapter: + +The `KafkaMessageDrivenChannelAdapter` (``) uses a `spring-kafka` +`KafkaMessageListenerContainer` or `ConcurrentListenerContainer`. + +An example of xml configuration variant is shown here: + +[source, xml] +---- + + + + + + + + + ... + + + + + + +---- + +When using Java Configuration: + +[source, java] +---- +@Bean +public KafkaMessageDrivenChannelAdapter + adapter(KafkaMessageListenerContainer container) { + KafkaMessageDrivenChannelAdapter kafkaMessageDrivenChannelAdapter = + new KafkaMessageDrivenChannelAdapter<>(container); + kafkaMessageDrivenChannelAdapter.setOutputChannel(received()); + return kafkaMessageDrivenChannelAdapter; +} + +@Bean +public KafkaMessageListenerContainer container() throws Exception { + ContainerProperties properties = new ContainerProperties(this.topic); + // set more properties + return new KafkaMessageListenerContainer<>(consumerFactory(), properties); +} + +@Bean +public ConsumerFactory consumerFactory() { + Map props = new HashMap<>(); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddress); + // set more properties + return new DefaultKafkaConsumerFactory<>(props); +} +----