diff --git a/samples/sample-01/README.adoc b/samples/sample-01/README.adoc
index 59504694..50c5261f 100644
--- a/samples/sample-01/README.adoc
+++ b/samples/sample-01/README.adoc
@@ -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.
diff --git a/samples/sample-01/pom.xml b/samples/sample-01/pom.xml
index 5db136ca..dfc3dd05 100644
--- a/samples/sample-01/pom.xml
+++ b/samples/sample-01/pom.xml
@@ -5,7 +5,7 @@
com.example
kafka-sample-01
- 2.3.1.RELEASE
+ 2.4.4.RELEASE
jar
kafka-sample-01
@@ -14,7 +14,7 @@
org.springframework.boot
spring-boot-starter-parent
- 2.2.0.RELEASE
+ 2.2.6.RELEASE
diff --git a/samples/sample-01/src/main/java/com/example/Application.java b/samples/sample-01/src/main/java/com/example/Application.java
index 5219c152..a7167809 100644
--- a/samples/sample-01/src/main/java/com/example/Application.java
+++ b/samples/sample-01/src/main/java/com/example/Application.java
@@ -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 kafkaConsumerFactory,
- KafkaTemplate template) {
- ConcurrentKafkaListenerContainerFactory 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 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();
+ };
+ }
+
}
diff --git a/samples/sample-02/pom.xml b/samples/sample-02/pom.xml
index 4da34f4e..56d4ca53 100644
--- a/samples/sample-02/pom.xml
+++ b/samples/sample-02/pom.xml
@@ -5,7 +5,7 @@
com.example
kafka-sample-02
- 2.3.1.RELEASE
+ 2.4.4.RELEASE
jar
kafka-sample-02
@@ -14,7 +14,7 @@
org.springframework.boot
spring-boot-starter-parent
- 2.2.0.RELEASE
+ 2.2.6.RELEASE
diff --git a/samples/sample-02/src/main/java/com/example/Application.java b/samples/sample-02/src/main/java/com/example/Application.java
index 33ae67cf..d679604a 100644
--- a/samples/sample-02/src/main/java/com/example/Application.java
+++ b/samples/sample-02/src/main/java/com/example/Application.java
@@ -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 kafkaConsumerFactory,
- KafkaTemplate template) {
- ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>();
- configurer.configure(factory, kafkaConsumerFactory);
- factory.setErrorHandler(new SeekToCurrentErrorHandler(
- new DeadLetterPublishingRecoverer(template), new FixedBackOff(0L, 2)));
- return factory;
+ public SeekToCurrentErrorHandler errorHandler(KafkaTemplate template) {
+ return new SeekToCurrentErrorHandler(
+ new DeadLetterPublishingRecoverer(template), new FixedBackOff(1000L, 2));
}
@Bean
diff --git a/samples/sample-03/pom.xml b/samples/sample-03/pom.xml
index 886dfbb7..44398ed2 100644
--- a/samples/sample-03/pom.xml
+++ b/samples/sample-03/pom.xml
@@ -5,7 +5,7 @@
com.example
kafka-sample-03
- 2.3.1.RELEASE
+ 2.4.4.RELEASE
jar
kafka-sample-03
@@ -14,7 +14,7 @@
org.springframework.boot
spring-boot-starter-parent
- 2.2.0.RELEASE
+ 2.2.6.RELEASE
diff --git a/samples/sample-03/src/main/java/com/example/Application.java b/samples/sample-03/src/main/java/com/example/Application.java
index b0554a54..ec5aad55 100644
--- a/samples/sample-03/src/main/java/com/example/Application.java
+++ b/samples/sample-03/src/main/java/com/example/Application.java
@@ -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 kafkaConsumerFactory) {
-
- ConcurrentKafkaListenerContainerFactory 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 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 in) {
- logger.info("Received: " + in);
+ LOGGER.info("Received: " + in);
+ LATCH.countDown();
}
@Bean
diff --git a/samples/sample-03/src/main/resources/application.yml b/samples/sample-03/src/main/resources/application.yml
index 11bba335..0fcf4971 100644
--- a/samples/sample-03/src/main/resources/application.yml
+++ b/samples/sample-03/src/main/resources/application.yml
@@ -6,3 +6,5 @@ spring:
consumer:
properties:
isolation.level: read_committed
+ listener:
+ type: batch