From 17d8f081b4a41df6c82758b470cf50b5d448593b Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 10 Mar 2016 12:40:32 -0500 Subject: [PATCH] GH-31: Initial Docs Resolves #31 Use Constants for Producer/Consumer Config Polishing and fix typos --- src/reference/asciidoc/kafka.adoc | 243 +++++++++++++++++++++++++ src/reference/asciidoc/preface.adoc | 3 - src/reference/asciidoc/quick-tour.adoc | 51 +++--- 3 files changed, 272 insertions(+), 25 deletions(-) diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index a0e5a4a1..ce287d80 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -3,8 +3,251 @@ ==== Sending Messages with the KafkaTemplate +The `KafkaTemplate` wraps a producer and provides convenience methods to send data to kafka topics. +Both asynchronous and synchronous methods are provided, with the async methods returning a `Future`. + +[source, java] +---- +// Async methods + +Future convertAndSend(V data); + +Future convertAndSend(K key, V data); + +Future convertAndSend(int partition, K key, V data); + +Future convertAndSend(String topic, V data); + +Future convertAndSend(String topic, K key, V data); + +Future convertAndSend(String topic, int partition, K key, V data); + + +// Sync methods + + +RecordMetadata syncConvertAndSend(V data) + throws InterruptedException, ExecutionException; + +RecordMetadata syncConvertAndSend(K key, V data) + throws InterruptedException, ExecutionException; + +RecordMetadata syncConvertAndSend(int partition, K key, V data) + throws InterruptedException, ExecutionException; + +RecordMetadata syncConvertAndSend(String topic, V data) + throws InterruptedException, ExecutionException; + +RecordMetadata syncConvertAndSend(String topic, K key, V data) + throws InterruptedException, ExecutionException; + +RecordMetadata syncConvertAndSend(String topic, int partition, K key, V data) + throws InterruptedException, ExecutionException; + +// Flush the producer. + +void flush(); +---- + +To use the template, configure a producer factory and provide it in the template's constructor: + +[source, java] +---- +@Bean +public ProducerFactory producerFactory() { + return new DefaultKafkaProducerFactory<>(producerConfigs()); +} + +@Bean +public Map producerConfigs() { + Map props = new HashMap<>(); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); + ... + return props; +} + +@Bean +public KafkaTemplate kafkaTemplate() { + return new KafkaTemplate(producerFactory()); +} +---- + +The template can also be configured using standard `` definitions. + +Then, to use the template, simply invoke one of its methods. + ==== Receiving Messages +Messages can be received by configuring a `MessageListenerContainer` and providing a `MessageListener`, or by +using the `@KafkaListener` annotation. + ===== Message Listener Containers +Two `MessageListenerContainer` implementations are provided: + +- `KafkaMessageListenerContainer` +- `ConcurrentMessageListenerContainer` + +The `KafkaMessageListenerContainer` receives all message from all topics/partitions on a single thread. +The `ConcurrentMessageListenerContainer` delegates to 1 or more `KafkaMessageListenerContainer` s to provide +multi-threaded consumption. + +====== KafkaMessageListenerContainer + +The following constructors are available. + +[source, java] +---- +public KafkaMessageListenerContainer(ConsumerFactory consumerFactory, + TopicPartition... topicPartitions) + +public KafkaMessageListenerContainer(ConsumerFactory consumerFactory, String... topics) + +public KafkaMessageListenerContainer(ConsumerFactory consumerFactory, + Pattern topicPattern) +---- + +Each takes a `ConsumerFactory` and information about topics and partitions. + +The first takes a list of `TopicPartition` arguments to explicitly instruct the container which partitions to use +(using the consumer `assign()` method). +The second takes a list of topics and Kafka allocates the partitions based on the `group.id` property - distributing +partitions across the group. +The third is similar to the second, but uses a regex `Pattern` to select the topics. + +====== ConcurrentMessageListenerContainer + +The constructors are similar to the `KafkaListenerContainer`: + +[source, java] +---- +public ConcurrentMessageListenerContainer(ConsumerFactory consumerFactory, TopicPartition... topicPartitions) + +public ConcurrentMessageListenerContainer(ConsumerFactory consumerFactory, String... topics) + +public ConcurrentMessageListenerContainer(ConsumerFactory consumerFactory, Pattern topicPattern) +---- + +It also has a property `concurrency`, e.g. `container.setConcurrency(3)` will create 3 +`KafkaMessageListenerContainer` s. + +For the second and third container, kafka will distribute the partitions across the consumers. +For the first constructor, the `ConcurrentMessageListenerContainer` distributes the `TopicPartition` s across the +delegate `KafkaMessageListenerContainer` s. + +If, say, 6 `TopicPartition` s are provided and the `concurrency` is 3; each container will get 2 partitions. +For 5 `TopicPartition` s, 2 containers will get 2 partitions and the third will get 1. +If the `concurrency` is greater than the number of `TopicPartitions`, the `concurrency` will be adjusted down such that +each container will get one partition. + +====== Committing Offsets + +Several options are provided for committing offsets. +If the `enable.auto.commit` consumer property is true, kafka will auto-commit the offsets according to its +configuration. +If it is false, the containers support the following `AckMode` s. + +The consumer `poll()` method will return one or more `ConsumerRecords`; the `MessageListener` is called for each record; +the following describes the action taken by the container for each `AckMode` : + +- RECORD - call `commitAsync()` when the listener returns after processing the record. +- BATCH - call `commitAsync()` when all the records returned by the `poll()` have been processed. +- TIME - call `commitAsync()` when all the records returned by the `poll()` have been processed as long as the `ackTime` +since the last commit has been exceeded. +- COUNT - call `commitAsync()` when all the records returned by the `poll()` have been processed as long as `ackCount` +records have been received since the last commit. +- COUNT_TIME - similar to TIME and COUNT but the commit is performed if either condition is true. +- MANUAL - the message listener (`AcknowledgingMessageListener`) is responsible to `acknowledge()` the `Acknowledgment`; +after which, the same semantics as `COUNT_TIME` are applied. +- MANUAL_IMMEDIATE - call `commitAsync()`` immediately when the `Acknowledgment.acknowledge()` method is called by the +listener - must be executed on the container's thread. + +NOTE: `MANUAL` and `MANUAL_IMMEDIATE` require the listener to be an `AcknowledgingMessageListener`. + +[source, java] +---- +public interface AcknowledgingMessageListener { + + void onMessage(ConsumerRecord record, Acknowledgment acknowledgment); + +} + +public interface Acknowledgment { + + void acknowledge(); + +} +---- + +This gives the listener control over when offsets are committed. + ===== @KafkaListener Annotation + +The `@KafkaListener` annotation provides a mechanism for simple POJO listeners: + +[source, java] +---- +public class Listener { + + @KafkaListener(id = "foo", topics = "myTopic") + public void listen(String data) { + ... + } + +} +---- + +This mechanism requires a listener container factory, which is used to configure the underlying +`ConcurrentMessageListenerContainer`: by default, a bean with name `kafkaListenerContainerFactory` is expected. + +[source, java] +---- +@Bean +KafkaListenerContainerFactory> + kafkaListenerContainerFactory() { + SimpleKafkaListenerContainerFactory factory = + new SimpleKafkaListenerContainerFactory<>(); + factory.setConsumerFactory(consumerFactory()); + factory.setConcurrency(3); + return factory; +} + +@Bean +public ConsumerFactory consumerFactory() { + return new DefaultKafkaConsumerFactory<>(consumerConfigs()); +} + +@Bean +public Map consumerConfigs() { + Map props = new HashMap<>(); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString()); + ... + return props; +} +---- + +You can also configure POJO listeners with explicit topics and partitions: + +[source, java] +---- +@KafkaListener(id = "bar", topicPartitions = + { @TopicPartition(topic = "topic1", partitions = { "0", "1" }), + @TopicPartition(topic = "topic2", partitions = { "0", "1" }) + }) +public void listen(ConsumerRecord record) { + ... +} +---- + +When using manual `AckMode`, the listener can also be provided with the `Acknowledgment`; this example also shows +how to use a different container factory. + +[source, java] +---- +@KafkaListener(id = "baz", topics = "myTopic", + containerFactory = "kafkaManualAckListenerContainerFactory") +public void listen(String data, Acknowledgment ack) { + ... + ack.acknowledge(); +} +---- diff --git a/src/reference/asciidoc/preface.adoc b/src/reference/asciidoc/preface.adoc index baea82a4..284244e6 100644 --- a/src/reference/asciidoc/preface.adoc +++ b/src/reference/asciidoc/preface.adoc @@ -2,6 +2,3 @@ The Spring for Apache Kafka project applies core Spring concepts to the development of Kafka-based messaging solutions. We provide a "template" as a high-level abstraction for sending messages. We also provide support for Message-driven POJOs. - -// TODO: spring-kafka project page? -For other project-related information visit the Spring Integration project http://projects.spring.io/spring-integration/[homepage]. diff --git a/src/reference/asciidoc/quick-tour.adoc b/src/reference/asciidoc/quick-tour.adoc index 9848aa84..20b69916 100644 --- a/src/reference/asciidoc/quick-tour.adoc +++ b/src/reference/asciidoc/quick-tour.adoc @@ -18,7 +18,7 @@ your build tool, e.g. for Maven: ---- -And for gradle: +And for Gradle: [source,groovy,subs="+attributes"] ---- @@ -36,7 +36,7 @@ versions of Spring. ===== Very, Very Quick -Using plain, imperative Java to send and receive a message: +Using plain Java to send and receive a message: [source,java] ---- @@ -49,9 +49,10 @@ public void testAutoCommit() throws Exception { @Override public void onMessage(ConsumerRecord message) { - logger.info("auto: " + message); + logger.info("received: " + message); latch.countDown(); } + }); container.setBeanName("testAuto"); container.start(); @@ -88,25 +89,29 @@ private KafkaTemplate createTemplate() { private Map consumerProps() { Map props = new HashMap<>(); - props.put("bootstrap.servers", "localhost:9092"); - props.put("group.id", "myGroup"); - props.put("enable.auto.commit", "true"); - props.put("auto.commit.interval.ms", "100"); - props.put("session.timeout.ms", "15000"); - props.put("key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer"); - props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); + props.put(ConsumerConfig.GROUP_ID_CONFIG, group); + props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true); + props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100"); + props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000"); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, + "org.apache.kafka.common.serialization.IntegerDeserializer"); + props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, + "org.apache.kafka.common.serialization.StringDeserializer"); return props; } private Map senderProps() { Map props = new HashMap<>(); - props.put("bootstrap.servers", "localhost:9092"); - props.put("retries", 0); - props.put("batch.size", 16384); - props.put("linger.ms", 1); - props.put("buffer.memory", 33554432); - props.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer"); - props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); + props.put(ProducerConfig.RETRIES_CONFIG, 0); + props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384); + props.put(ProducerConfig.LINGER_MS_CONFIG, 1); + props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, + "org.apache.kafka.common.serialization.IntegerSerializer"); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, + "org.apache.kafka.common.serialization.StringSerializer"); return props; } ---- @@ -135,7 +140,7 @@ public void testSimple() throws Exception { public class Config { @Bean - KafkaListenerContainerFactory> + SimpleKafkaListenerContainerFactory kafkaListenerContainerFactory() { SimpleKafkaListenerContainerFactory factory = new SimpleKafkaListenerContainerFactory<>(); @@ -148,10 +153,11 @@ public class Config { return new DefaultKafkaConsumerFactory<>(consumerConfigs()); } + @Bean public Map consumerConfigs() { Map props = new HashMap<>(); - props.put("bootstrap.servers", embeddedKafka.getBrokersAsString()); - ...... + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString()); + ... return props; } @@ -165,10 +171,11 @@ public class Config { return new DefaultKafkaProducerFactory<>(producerConfigs()); } + @Bean public Map producerConfigs() { Map props = new HashMap<>(); - props.put("bootstrap.servers", embeddedKafka.getBrokersAsString()); - ...... + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString()); + ... return props; }