Spring boot kafka job example
application-kafka.yml file:
spring:
kafka:
listener:
ack-mode: manual
consumer:
key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
properties:
spring.json.use.type.headers: false
spring.json.trusted.packages: com.example.*
spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
auto-offset-reset: latest
enable-auto-commit: false
operation-reversing:
group-id: com.example.operation.reversing
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
retries: 1
properties:
spring.json.trusted.packages: com.example.*
topic:
operation-reversing:
name: ${OPERATION_REVERSING_TOPIC:com.example.operation.reversing}
enable: true
concurrency: 1
partition-count: 1
replication-factor: 1
retry:
partition-count: 1
replication-factor: 1
attempts: 1
delay: 2000
max-delay: 17000
multiplier: 2.0
retry:
retry-topic-suffix: -retry
dlt-topic-suffix: -dlt
Consumer class:
@Slf4j
@Component
@RequiredArgsConstructor
@ConditionalOnProperty(name = "spring.kafka.topic.operation-reversing.enable")
public class OperationReversingConsumer {
private final ExampleService exampleService;
@KafkaListener(
topics = "${spring.kafka.topic.operation-reversing.name}",
groupId = "${spring.kafka.consumer.operation-reversing.group-id}",
concurrency = "${spring.kafka.topic.operation-reversing.concurrency}",
properties = {
"spring.json.value.default.type:"
+ "com.example.exam.operation.job.model.event.JobTriggerEvent"
}
)
@RetryableTopic(
attempts = "${spring.kafka.topic.operation-reversing.retry.attempts}",
retryTopicSuffix = "${spring.kafka.topic.retry.retry-topic-suffix}",
dltTopicSuffix = "${spring.kafka.topic.retry.dlt-topic-suffix}",
numPartitions = "${spring.kafka.topic.operation-reversing.retry.partition-count}",
replicationFactor = "${spring.kafka.topic.operation-reversing.retry.replication-factor}",
backoff = @Backoff(
multiplierExpression = "${spring.kafka.topic.operation-reversing.retry.multiplier}",
delayExpression = "${spring.kafka.topic.operation-reversing.retry.delay}",
maxDelayExpression = "${spring.kafka.topic.operation-reversing.retry.max-delay}"
),
topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_DELAY_VALUE
)
public void handleOperationReversing(JobTriggerEvent event,
Acknowledgment acknowledgment) {
try {
log.info("Reverse unsuccessful operation {}", event);
exampleService.reverseUnSuccessPayments();
acknowledgment.acknowledge();
log.info("Reverse unsuccessful operation is successful");
} catch (Exception e) {
throw new ExampleJobException(EVENT_CONSUMING_FAILED_ERROR, e);
}
}
}
Properties mapping class:
@Data
@ConfigurationProperties(prefix = "spring.kafka.topic")
public class KafkaTopicProperties {
private OperationReversing operationReversing;
private GlobalRetry retry;
@Data
public static class OperationReversing {
private String name;
private boolean enable;
private int concurrency;
private int partitionCount;
private int replicationFactor;
private Retry retry;
}
@Data
public static class Retry {
private int partitionCount;
private int replicationFactor;
private int attempts;
private long delay;
private long maxDelay;
private double multiplier;
}
@Data
public static class GlobalRetry {
private String retryTopicSuffix;
private String dltTopicSuffix;
}
}
Comments
Post a Comment