Spring for Apache Kafka integrates Spring’s dependency-injection container with Kafka applications. Components are discovered and wired into an application context, letting application code focus on handling records while Spring supplies configured dependencies such as KafkaTemplate instances. section 12.0 section 12.2.1
Spring Boot reduces setup through convention over configuration. A configuration class can load property values, declare NewTopic beans, scan application components, and start without a web server when the application only processes Kafka records. section 12.2 section 12.4
The main application model combines Spring-managed components with Kafka-facing annotations and templates:
- Mark processing classes with @Component so the container discovers them and can inject dependencies. section 12.2.1
- Use @KafkaListener on a method to consume a topic as a consumer group; a listener container wraps the Kafka consumer. section 12.2.1
- Use KafkaTemplate to produce processed records to selected output topics. section 12.2.1
- Configure listener concurrency with ConcurrentKafkaListenerContainerFactory when partitions permit parallel consumers; listener methods must then be thread-safe. section 12.2.2 section 12.4
- Receive record metadata through @Header parameters and attach callbacks to sends to log produced-record topic, offset, and timestamp. section 12.2.2
Loan application routing
A listener receives each loan application, calculates approval from debt ratio and credit rating, then sends the updated record to an accepted or rejected topic. A random subset is also sent to a quality-assurance topic, where separate listeners perform post-processing. section 12.2.1
For Kafka Streams, Spring Boot can activate lifecycle support with @EnableKafkaStreams and inject a StreamsBuilder for defining a topology. This trades some construction control for less infrastructure code; alternatively, Spring can provide dependency injection while the application explicitly builds, starts, and closes KafkaStreams. section 12.3 section 12.4
Listing 12.7 Loan processing handling in the method
@KafkaListener(topics = "${loan.app.input.topic}", #1
groupId = "${application.group}") #2
public void doProcessLoanApplication(LoanApplication loanApplication) {
boolean loanApproved = debtRatio <= 0.33 &&
loanApplication.getCreditRating() > 650;
String topicToSend = loanApproved ? acceptedLoansTopic :
rejectedLoansTopic; #3
LoanApplication processedLoan = LoanApplication.Builder.newBuilder(
loanApplication).withApproved(loanApproved).build();
kafkaTemplate.send(topicToSend, processedLoan.getCustomerId(),
processedLoan); #4
if (random.nextInt(100) > 75) {
kafkaTemplate.send(qaLoansTopic, processedLoan.getCustomerId(),
processedLoan); #5
}
}
The processor declaration shows property injection and constructor injection of KafkaTemplate in a Spring-managed component. section 12.2.1
