GH-31: Initial Docs
Resolves #31 Use Constants for Producer/Consumer Config Polishing and fix typos
This commit is contained in:
Gary Russell
committed by
Artem Bilan
Artem Bilan
parent
3e4c145f8b
commit
17d8f081b4
@@ -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<RecordMetadata> convertAndSend(V data);
|
||||
|
||||
Future<RecordMetadata> convertAndSend(K key, V data);
|
||||
|
||||
Future<RecordMetadata> convertAndSend(int partition, K key, V data);
|
||||
|
||||
Future<RecordMetadata> convertAndSend(String topic, V data);
|
||||
|
||||
Future<RecordMetadata> convertAndSend(String topic, K key, V data);
|
||||
|
||||
Future<RecordMetadata> 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<Integer, String> producerFactory() {
|
||||
return new DefaultKafkaProducerFactory<>(producerConfigs());
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Map<String, Object> producerConfigs() {
|
||||
Map<String, Object> props = new HashMap<>();
|
||||
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
|
||||
...
|
||||
return props;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public KafkaTemplate<Integer, String> kafkaTemplate() {
|
||||
return new KafkaTemplate<Integer, String>(producerFactory());
|
||||
}
|
||||
----
|
||||
|
||||
The template can also be configured using standard `<bean/>` 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<K, V> consumerFactory,
|
||||
TopicPartition... topicPartitions)
|
||||
|
||||
public KafkaMessageListenerContainer(ConsumerFactory<K, V> consumerFactory, String... topics)
|
||||
|
||||
public KafkaMessageListenerContainer(ConsumerFactory<K, V> 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<K, V> consumerFactory, TopicPartition... topicPartitions)
|
||||
|
||||
public ConcurrentMessageListenerContainer(ConsumerFactory<K, V> consumerFactory, String... topics)
|
||||
|
||||
public ConcurrentMessageListenerContainer(ConsumerFactory<K, V> 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<K, V> {
|
||||
|
||||
void onMessage(ConsumerRecord<K, V> 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<ConcurrentMessageListenerContainer<Integer, String>>
|
||||
kafkaListenerContainerFactory() {
|
||||
SimpleKafkaListenerContainerFactory<Integer, String> factory =
|
||||
new SimpleKafkaListenerContainerFactory<>();
|
||||
factory.setConsumerFactory(consumerFactory());
|
||||
factory.setConcurrency(3);
|
||||
return factory;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public ConsumerFactory<Integer, String> consumerFactory() {
|
||||
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Map<String, Object> consumerConfigs() {
|
||||
Map<String, Object> 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();
|
||||
}
|
||||
----
|
||||
|
||||
@@ -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].
|
||||
|
||||
@@ -18,7 +18,7 @@ your build tool, e.g. for Maven:
|
||||
</dependency>
|
||||
----
|
||||
|
||||
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<Integer, String> message) {
|
||||
logger.info("auto: " + message);
|
||||
logger.info("received: " + message);
|
||||
latch.countDown();
|
||||
}
|
||||
|
||||
});
|
||||
container.setBeanName("testAuto");
|
||||
container.start();
|
||||
@@ -88,25 +89,29 @@ private KafkaTemplate<Integer, String> createTemplate() {
|
||||
|
||||
private Map<String, Object> consumerProps() {
|
||||
Map<String, Object> 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<String, Object> senderProps() {
|
||||
Map<String, Object> 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<ConcurrentMessageListenerContainer<Integer, String>>
|
||||
SimpleKafkaListenerContainerFactory<Integer, String>
|
||||
kafkaListenerContainerFactory() {
|
||||
SimpleKafkaListenerContainerFactory<Integer, String> factory =
|
||||
new SimpleKafkaListenerContainerFactory<>();
|
||||
@@ -148,10 +153,11 @@ public class Config {
|
||||
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Map<String, Object> consumerConfigs() {
|
||||
Map<String, Object> 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<String, Object> producerConfigs() {
|
||||
Map<String, Object> props = new HashMap<>();
|
||||
props.put("bootstrap.servers", embeddedKafka.getBrokersAsString());
|
||||
......
|
||||
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
|
||||
...
|
||||
return props;
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user