Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Update samples to use current boot autoconfigure

Browse Source 
- auto config of error handler and message converter based on listener type.
This commit is contained in:
Gary Russell
2020-03-27 17:47:50 -04:00
committed by Artem Bilan
parent 19485c3d69
commit 682d68f630
8 changed files with 61 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
samples/sample-01/README.adoc
View File

@@ -12,7 +12,9 @@ Console:
`2018-11-05 10:03:40.216 INFO 39766 --- [ fooGroup-0-C-1] com.example.Application : Received: Foo2 [foo=bar]`
The consumer is configured with a `SeekToCurrentErrorHandler` which replays failed messages up to 3 times and, after retries are exhausted, sends a bad message to a dead-letter topic.
The consumer is configured with a `SeekToCurrentErrorHandler` which replays failed messages up to 2 times, each after a 1 second delay and, after retries are exhausted, sends a bad message to a dead-letter topic.
`$ curl -X POST http://localhost:8080/send/foo/fail`
A second `@KafkaListener` consumes the raw JSON from the message.

4
samples/sample-01/pom.xml
View File

@@ -5,7 +5,7 @@
<groupId>com.example</groupId>
<artifactId>kafka-sample-01</artifactId>
<version>2.3.1.RELEASE</version>
<version>2.4.4.RELEASE</version>
<packaging>jar</packaging>
<name>kafka-sample-01</name>
@@ -14,7 +14,7 @@
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.0.RELEASE</version>
<version>2.2.6.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>

36
samples/sample-01/src/main/java/com/example/Application.java
View File

@@ -20,13 +20,13 @@ import org.apache.kafka.clients.admin.NewTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
@@ -37,6 +37,7 @@ import org.springframework.util.backoff.FixedBackOff;
import com.common.Foo2;
/**
* Sample shows use of a dead letter topic.
*
* @author Gary Russell
* @since 2.2.1
@@ -47,20 +48,19 @@ public class Application {
private final Logger logger = LoggerFactory.getLogger(Application.class);
private final TaskExecutor exec = new SimpleAsyncTaskExecutor();
public static void main(String[] args) {
SpringApplication.run(Application.class, args);
SpringApplication.run(Application.class, args).close();
}
/*
* Boot will autowire this into the container factory.
*/
@Bean
public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
ConsumerFactory<Object, Object> kafkaConsumerFactory,
KafkaTemplate<Object, Object> template) {
ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
configurer.configure(factory, kafkaConsumerFactory);
factory.setErrorHandler(new SeekToCurrentErrorHandler(
new DeadLetterPublishingRecoverer(template), new FixedBackOff(0L, 2))); // dead-letter after 3 tries
return factory;
public SeekToCurrentErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
return new SeekToCurrentErrorHandler(
new DeadLetterPublishingRecoverer(template), new FixedBackOff(1000L, 2));
}
@Bean
@@ -74,11 +74,13 @@ public class Application {
if (foo.getFoo().startsWith("fail")) {
throw new RuntimeException("failed");
}
this.exec.execute(() -> System.out.println("Hit Enter to terminate..."));
}
@KafkaListener(id = "dltGroup", topics = "topic1.DLT")
public void dltListen(String in) {
logger.info("Received from DLT: " + in);
this.exec.execute(() -> System.out.println("Hit Enter to terminate..."));
}
@Bean
@@ -91,4 +93,12 @@ public class Application {
return new NewTopic("topic1.DLT", 1, (short) 1);
}
@Bean
public ApplicationRunner runner() {
return args -> {
System.out.println("Hit Enter to terminate...");
System.in.read();
};
}
}

4
samples/sample-02/pom.xml
View File

@@ -5,7 +5,7 @@
<groupId>com.example</groupId>
<artifactId>kafka-sample-02</artifactId>
<version>2.3.1.RELEASE</version>
<version>2.4.4.RELEASE</version>
<packaging>jar</packaging>
<name>kafka-sample-02</name>
@@ -14,7 +14,7 @@
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.0.RELEASE</version>
<version>2.2.6.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>

25
samples/sample-02/src/main/java/com/example/Application.java
View File

@@ -23,10 +23,7 @@ import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
@@ -39,6 +36,13 @@ import org.springframework.util.backoff.FixedBackOff;
import com.common.Bar2;
import com.common.Foo2;
/**
* Sample shows use of a multi-method listener.
*
* @author Gary Russell
* @since 2.2.1
*
*/
@SpringBootApplication
public class Application {
@@ -46,16 +50,13 @@ public class Application {
SpringApplication.run(Application.class, args);
}
/*
* Boot will autowire this into the container factory.
*/
@Bean
public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
ConsumerFactory<Object, Object> kafkaConsumerFactory,
KafkaTemplate<Object, Object> template) {
ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
configurer.configure(factory, kafkaConsumerFactory);
factory.setErrorHandler(new SeekToCurrentErrorHandler(
new DeadLetterPublishingRecoverer(template), new FixedBackOff(0L, 2)));
return factory;
public SeekToCurrentErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
return new SeekToCurrentErrorHandler(
new DeadLetterPublishingRecoverer(template), new FixedBackOff(1000L, 2));
}
@Bean

4
samples/sample-03/pom.xml
View File

@@ -5,7 +5,7 @@
<groupId>com.example</groupId>
<artifactId>kafka-sample-03</artifactId>
<version>2.3.1.RELEASE</version>
<version>2.4.4.RELEASE</version>
<packaging>jar</packaging>
<name>kafka-sample-03</name>
@@ -14,7 +14,7 @@
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.0.RELEASE</version>
<version>2.2.6.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>

35
samples/sample-03/src/main/java/com/example/Application.java
View File

@@ -18,6 +18,7 @@ package com.example;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import org.apache.kafka.clients.admin.NewTopic;
import org.slf4j.Logger;
@@ -26,12 +27,10 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.converter.BatchMessagingMessageConverter;
import org.springframework.kafka.support.converter.RecordMessageConverter;
@@ -40,6 +39,7 @@ import org.springframework.kafka.support.converter.StringJsonMessageConverter;
import com.common.Foo2;
/**
* Sample showing a batch listener and transactions.
*
* @author Gary Russell
* @since 2.2.1
@@ -48,23 +48,15 @@ import com.common.Foo2;
@SpringBootApplication
public class Application {
private final Logger logger = LoggerFactory.getLogger(Application.class);
private final Logger LOGGER = LoggerFactory.getLogger(Application.class);
public static void main(String[] args) {
SpringApplication.run(Application.class, args);
}
private final static CountDownLatch LATCH = new CountDownLatch(1);
@Bean
public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
ConsumerFactory<Object, Object> kafkaConsumerFactory) {
ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
new ConcurrentKafkaListenerContainerFactory<>();
configurer.configure(factory, kafkaConsumerFactory);
factory.setBatchListener(true);
factory.setMessageConverter(batchConverter());
return factory;
public static void main(String[] args) throws InterruptedException {
ConfigurableApplicationContext context = SpringApplication.run(Application.class, args);
LATCH.await();
Thread.sleep(5_000);
context.close();
}
@Bean
@@ -82,15 +74,16 @@ public class Application {
@KafkaListener(id = "fooGroup2", topics = "topic2")
public void listen1(List<Foo2> foos) throws IOException {
logger.info("Received: " + foos);
LOGGER.info("Received: " + foos);
foos.forEach(f -> kafkaTemplate.send("topic3", f.getFoo().toUpperCase()));
logger.info("Messages sent, hit Enter to commit tx");
LOGGER.info("Messages sent, hit Enter to commit tx");
System.in.read();
}
@KafkaListener(id = "fooGroup3", topics = "topic3")
public void listen2(List<String> in) {
logger.info("Received: " + in);
LOGGER.info("Received: " + in);
LATCH.countDown();
}
@Bean

2
samples/sample-03/src/main/resources/application.yml
View File

@@ -6,3 +6,5 @@ spring:
consumer:
properties:
isolation.level: read_committed
listener:
type: batch