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

Upgrade samples

Browse Source
This commit is contained in:
Gary Russell
2019-10-02 10:16:35 -04:00
committed by Artem Bilan
parent abdc1e136c
commit 0feba30e76
6 changed files with 22 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -5,7 +5,7 @@
<groupId>com.example</groupId>
<artifactId>kafka-sample-01</artifactId>
<version>0.0.1-SNAPSHOT</version>
<version>2.3.1.BUILD-SNAPSHOT</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.1.0.RELEASE</version>
<version>2.2.0.BUILD-SNAPSHOT</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>

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

@@ -32,6 +32,7 @@ import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.kafka.support.converter.StringJsonMessageConverter;
import org.springframework.util.backoff.FixedBackOff;
import com.common.Foo2;
@@ -58,7 +59,7 @@ public class Application {
ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
configurer.configure(factory, kafkaConsumerFactory);
factory.setErrorHandler(new SeekToCurrentErrorHandler(
new DeadLetterPublishingRecoverer(template), 3)); // dead-letter after 3 tries
new DeadLetterPublishingRecoverer(template), new FixedBackOff(0L, 2))); // dead-letter after 3 tries
return factory;
}

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

@@ -5,7 +5,7 @@
<groupId>com.example</groupId>
<artifactId>kafka-sample-02</artifactId>
<version>0.0.1-SNAPSHOT</version>
<version>2.3.1.BUILD-SNAPSHOT</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.1.0.RELEASE</version>
<version>2.2.0.BUILD-SNAPSHOT</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>

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

@@ -34,6 +34,7 @@ import org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper
import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper.TypePrecedence;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.kafka.support.converter.StringJsonMessageConverter;
import org.springframework.util.backoff.FixedBackOff;
import com.common.Bar2;
import com.common.Foo2;
@@ -53,7 +54,7 @@ public class Application {
ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
configurer.configure(factory, kafkaConsumerFactory);
factory.setErrorHandler(new SeekToCurrentErrorHandler(
new DeadLetterPublishingRecoverer(template), 3));
new DeadLetterPublishingRecoverer(template), new FixedBackOff(0L, 2)));
return factory;
}

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

@@ -5,7 +5,7 @@
<groupId>com.example</groupId>
<artifactId>kafka-sample-03</artifactId>
<version>0.0.1-SNAPSHOT</version>
<version>2.3.1.BUILD-SNAPSHOT</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.1.0.RELEASE</version>
<version>2.2.0.BUILD-SNAPSHOT</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>

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

@@ -30,6 +30,7 @@ import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerConta
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;
@@ -56,9 +57,10 @@ public class Application {
@Bean
public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
ConsumerFactory<Object, Object> kafkaConsumerFactory,
KafkaTemplate<Object, Object> template) {
ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
ConsumerFactory<Object, Object> kafkaConsumerFactory) {
ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
new ConcurrentKafkaListenerContainerFactory<>();
configurer.configure(factory, kafkaConsumerFactory);
factory.setBatchListener(true);
factory.setMessageConverter(batchConverter());
@@ -92,8 +94,13 @@ public class Application {
}
@Bean
public NewTopic topic() {
return new NewTopic("topic2", 1, (short) 1);
public NewTopic topic2() {
return TopicBuilder.name("topic2").partitions(1).replicas(1).build();
}
@Bean
public NewTopic topic3() {
return TopicBuilder.name("topic3").partitions(1).replicas(1).build();
}
}