2.2.8.RELEASE
Это 6 месяцев.
Вы должны всегда использовать последнюю версию в вспомогательном выпуске, чтобы убедиться, что у вас есть все ошибки исправления; текущая версия 2.2.x - 2.2.12.
Я переработал свой ответ 2.2.x из здесь с примером того, как вы можете протестировать такое приложение. Он использует встроенный тестовый брокер Kafka.
@SpringBootApplication
public class So601723041Application {
public static void main(String[] args) {
SpringApplication.run(So601723041Application.class, args);
}
@Bean
ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
ConsumerFactory<Object, Object> kafkaConsumerFactory,
KafkaTemplate<Object, Object> template) {
ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
new ConcurrentKafkaListenerContainerFactory<Object, Object>() {
@Override
protected void initializeContainer(ConcurrentMessageListenerContainer<Object, Object> instance,
KafkaListenerEndpoint endpoint) {
super.initializeContainer(instance, endpoint);
customize(instance, template);
}
};
configurer.configure(factory, kafkaConsumerFactory);
return factory;
}
@Bean
ConcurrentKafkaListenerContainerFactory<?, ?> retryKafkaListenerContainerFactory(
ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
ConsumerFactory<Object, Object> kafkaConsumerFactory,
KafkaTemplate<Object, Object> template) {
ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
new ConcurrentKafkaListenerContainerFactory<Object, Object>() {
@Override
protected void initializeContainer(ConcurrentMessageListenerContainer<Object, Object> instance,
KafkaListenerEndpoint endpoint) {
super.initializeContainer(instance, endpoint);
customize(instance, template);
}
};
configurer.configure(factory, kafkaConsumerFactory);
RetryTemplate retryTemplate = new RetryTemplate();
retryTemplate.setRetryPolicy(new SimpleRetryPolicy(3));
FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
backOffPolicy.setBackOffPeriod(5000L);
retryTemplate.setBackOffPolicy(backOffPolicy);
factory.setRetryTemplate(retryTemplate);
return factory;
}
private void customize(ConcurrentMessageListenerContainer<Object, Object> container,
KafkaTemplate<Object, Object> template) {
if (container.getContainerProperties().getTopics()[0].equals("topic")) {
container.setErrorHandler(new SeekToCurrentErrorHandler(
new DeadLetterPublishingRecoverer(template,
(cr, ex) -> new TopicPartition("retryTopic", cr.partition())),
0));
}
else if (container.getContainerProperties().getTopics()[0].equals("retryTopic")) {
container.setErrorHandler(new SeekToCurrentErrorHandler(
new DeadLetterPublishingRecoverer(template,
(cr, ex) -> new TopicPartition("errorTopic", cr.partition())),
0)); // no retries here - retry template instead.
}
}
}
@Component
public class Listener {
private final KafkaTemplate<String, String> template;
private SomeService service;
public Listener(KafkaTemplate<String, String> template, SomeService service) {
this.template = template;
this.service = service;
}
public void setService(SomeService service) {
this.service = service;
}
@KafkaListener(id = "so60172304.1", topics = "topic")
public void listen1(String in) {
this.service.process(in);
this.template.send("successTopic", in);
}
@KafkaListener(id = "so60172304.2", topics = "retryTopic", containerFactory = "retryKafkaListenerContainerFactory")
public void listen2(String in) {
this.service.retry(in);
this.template.send("successTopic", in);
}
}
public interface SomeService {
void process(String in);
void retry(String in);
}
@Component
public class DefaultService implements SomeService {
@Override
public void process(String in) {
System.out.println("topic: " + in);
}
@Override
public void retry(String in) {
System.out.println("retryTopic: " + in);
}
}
spring.kafka.consumer.auto-offset-reset=earliest
Контрольный пример:
@RunWith(SpringRunner.class)
@SpringBootTest
@TestPropertySource(properties = "spring.kafka.bootstrap-servers = ${spring.embedded.kafka.brokers}")
public class So601723041ApplicationTests {
@ClassRule
public static EmbeddedKafkaRule embedded = new EmbeddedKafkaRule(1, true, 1,
"topic", "retryTopic", "successTopic", "errorTopic");
@Autowired
private Listener listener;
@Autowired
private KafkaTemplate<String, String> template;
@Test
public void test() {
TestService testService = spy(new TestService());
this.listener.setService(testService);
template.send("topic", "failAlways");
template.send("topic", "onlyFailFirst");
template.send("topic", "good");
Map<String, Object> props = KafkaTestUtils.consumerProps("test1", "false", embedded.getEmbeddedKafka());
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
Consumer<Integer, String> consumer = new KafkaConsumer<>(props);
embedded.getEmbeddedKafka().consumeFromEmbeddedTopics(consumer, "successTopic");
List<ConsumerRecord<Integer, String>> received = new ArrayList<>();
int n = 0;
while (received.size() < 2 && n++ < 10) {
ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofSeconds(10));
records.forEach(rec -> received.add(rec));
}
assertThat(received.size() == 2);
consumer.close();
props.put(ConsumerConfig.GROUP_ID_CONFIG, "test2");
consumer = new KafkaConsumer<>(props);
embedded.getEmbeddedKafka().consumeFromEmbeddedTopics(consumer, "errorTopic");
n = 0;
while (received.size() < 3 && n++ < 10) {
ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofSeconds(10));
records.forEach(rec -> received.add(rec));
}
assertThat(received.size() == 3);
consumer.close();
verify(testService, times(3)).process(anyString());
verify(testService, times(4)).retry(anyString());
assertThat(received).extracting(rec -> rec.value())
.contains("good", "onlyFailFirst", "failAlways");
}
public static class TestService implements SomeService {
CountDownLatch latch = new CountDownLatch(1);
@Override
public void process(String in) {
System.out.println("topic: " + in);
if (in.toLowerCase().contains("fail")) {
throw new RuntimeException(in);
}
}
@Override
public void retry(String in) {
System.out.println("retryTopic: " + in);
if (in.startsWith("fail")) {
throw new RuntimeException(in);
}
}
}
}
EMMA показывает 97,2% покрытия
POM:
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.1.12.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.example</groupId>
<artifactId>so60172304-1</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>so60172304-1</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>