Building the Producer Microservice
The first real code on this site. By the end of this module, HTTP requests into your Spring Boot app are landing as durable records in Kafka.
Learning objectives
- Beginner: A single POST endpoint publishing unkeyed JSON events to one topic, letting the default partitioner spread them.
- Intermediate: POST + PUT with keyed updates, environment-specific producer config via Spring Profiles, and centralized validation error responses.
- Advanced: Combine with Module 13's reliability config (idempotence, acks=all) and Module 16's transactional send, so a single request atomically publishes multiple related events.
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-validation</artifactId> </dependency>
public record Book( Integer bookId, @NotBlank String bookName, @NotBlank String bookAuthor ) {} public enum LibraryEventType { NEW, UPDATE } public record LibraryEvent( Integer libraryEventId, LibraryEventType libraryEventType, @Valid @NotNull Book book ) {}
💻 Code example
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-validation</artifactId> </dependency>
POST publishes a NEW event without a key (order doesn't matter yet — the record doesn't exist). PUT publishes an UPDATE, keyed by libraryEventId so it's guaranteed to land on the same partition as (and therefore be ordered after) the original NEW event — directly applying the partitioning rule from Module 01.
@RestController @RequestMapping("/v1") public class LibraryEventController { private final LibraryEventProducer libraryEventProducer; public LibraryEventController(LibraryEventProducer libraryEventProducer) { this.libraryEventProducer = libraryEventProducer; } @PostMapping("/libraryevent") public ResponseEntity<LibraryEvent> createEvent(@Valid @RequestBody LibraryEvent libraryEvent) throws JsonProcessingException { LibraryEvent event = new LibraryEvent( null, LibraryEventType.NEW, libraryEvent.book()); libraryEventProducer.sendLibraryEvent(event); // no key — see §06.3 return ResponseEntity.status(HttpStatus.CREATED).body(event); } @PutMapping("/libraryevent") public ResponseEntity<LibraryEvent> updateEvent(@Valid @RequestBody LibraryEvent libraryEvent) throws JsonProcessingException { if (libraryEvent.libraryEventId() == null) { throw new IllegalArgumentException("libraryEventId is required for update"); } LibraryEvent event = new LibraryEvent( libraryEvent.libraryEventId(), LibraryEventType.UPDATE, libraryEvent.book()); libraryEventProducer.sendLibraryEventWithKey(event); // keyed — see §06.3 return ResponseEntity.ok(event); } }
💻 Code example
@RestController @RequestMapping("/v1") public class LibraryEventController { private final LibraryEventProducer libraryEventProducer; public LibraryEventController(LibraryEventProducer libraryEventProducer) { this.libraryEventProducer = libraryEventProducer; } @PostMapping("/libraryevent") public ResponseEntity<LibraryEvent> createEvent(@Valid @RequestBody LibraryEvent libraryEvent) throws JsonProcessingException { LibraryEvent event = new LibraryEvent( null, LibraryEventType.NEW, libraryEvent.book()); libraryEventProducer.sendLibraryEvent(event); // no key — see §06.3 return ResponseEntity.status(HttpStatus.CREATED).body(event); } @PutMapping("/libraryevent") public ResponseEntity<LibraryEvent> updateEvent(@Valid @RequestBody LibraryEvent libraryEvent) throws JsonProcessingException { if (libraryEvent.libraryEventId() == null) { throw new IllegalArgumentException("libraryEventId is required for update"); } LibraryEvent event = new LibraryEvent( libraryEvent.libraryEventId(), LibraryEventType.UPDATE, libraryEvent.book()); libraryEventProducer.sendLibraryEventWithKey(event); // keyed — see §06.3 return ResponseEntity.ok(event); } }
KafkaTemplate is Spring Kafka's wrapper over the raw KafkaProducer, giving you a Spring-idiomatic API — send() overloads, integration with Spring's transaction management (Module 16), and Spring Boot auto-configuration from application.yml instead of manual Properties construction.
@Component public class LibraryEventProducer { private static final String TOPIC = "library-events"; private final KafkaTemplate<String, String> kafkaTemplate; private final ObjectMapper objectMapper; // ASYNC (fire-and-forget with a callback) — does not block the calling thread public void sendLibraryEvent(LibraryEvent event) throws JsonProcessingException { String value = objectMapper.writeValueAsString(event); CompletableFuture<SendResult<String,String>> future = kafkaTemplate.send(TOPIC, value); future.whenComplete((result, ex) -> { if (ex != null) handleFailure(value, ex); else handleSuccess(value, result); }); } // Keyed, ordered publish for UPDATE events public void sendLibraryEventWithKey(LibraryEvent event) throws JsonProcessingException { String key = event.libraryEventId().toString(); String value = objectMapper.writeValueAsString(event); kafkaTemplate.send(TOPIC, key, value) .whenComplete((result, ex) -> { if (ex != null) handleFailure(value, ex); else handleSuccess(value, result); }); } // SYNC — blocks the calling thread until the broker acknowledges (or times out) public SendResult<String,String> sendLibraryEventSync(LibraryEvent event) throws JsonProcessingException, ExecutionException, InterruptedException, TimeoutException { String value = objectMapper.writeValueAsString(event); return kafkaTemplate.send(TOPIC, value).get(3, TimeUnit.SECONDS); } }
◆ KafkaTemplate under the hood — threading model
Calling kafkaTemplate.send() from a web request thread does not make that thread do Kafka I/O. Internally, KafkaTemplate delegates to a shared ProducerFactory -managed KafkaProducer, whose own background sender thread (§03.1) does the actual network work. Your web thread gets a CompletableFuture back and is free to return — Tomcat's request thread isn't tied up waiting on Kafka unless you explicitly call.get() (the sync path above), which blocks the caller until the future completes or the timeout elapses.
💻 Code example
@Component public class LibraryEventProducer { private static final String TOPIC = "library-events"; private final KafkaTemplate<String, String> kafkaTemplate; private final ObjectMapper objectMapper; // ASYNC (fire-and-forget with a callback) — does not block the calling thread public void sendLibraryEvent(LibraryEvent event) throws JsonProcessingException { String value = objectMapper.writeValueAsString(event); CompletableFuture<SendResult<String,String>> future = kafkaTemplate.send(TOPIC, value); future.whenComplete((result, ex) -> { if (ex != null) handleFailure(value, ex); else handleSuccess(value, result); }); } // Keyed, ordered publish for UPDATE events public void sendLibraryEventWithKey(LibraryEvent event) throws JsonProcessingException { String key = event.libraryEventId().toString(); String value = objectMapper.writeValueAsString(event); kafkaTemplate.send(TOPIC, key, value) .whenComplete((result, ex) -> { if (ex != null) handleFailure(value, ex); else handleSuccess(value, result); }); } // SYNC — blocks the calling thread until the broker acknowledges (or times out) public SendResult<String,String> sendLibraryEventSync(LibraryEvent event) throws JsonProcessingException, ExecutionException, InterruptedException, TimeoutException { String value = objectMapper.writeValueAsString(event); return kafkaTemplate.send(TOPIC, value).get(3, TimeUnit.SECONDS); } }
Spring Boot builds the ProducerFactory and KafkaTemplate beans for you from spring.kafka.* properties — no manual Properties map needed for the common case. KafkaAdmin similarly auto-creates topics declared as NewTopic beans, useful for local dev (never rely on it in a real production cluster, where topic creation should be deliberate and reviewed).
spring: kafka: producer: bootstrap-servers: localhost:9092 key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer
@Bean public NewTopic libraryEvents() { return TopicBuilder.name("library-events").partitions(3).replicas(3).build(); }
💻 Code example
spring: kafka: producer: bootstrap-servers: localhost:9092 key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer
| StringSerializer | JsonSerializer | |
|---|---|---|
| What it does | Sends raw bytes of a String you already serialized (e.g. via Jackson yourself) | Serializes a Java object to JSON automatically |
| Control over output | Full — you control exact JSON shape/formatting | Convenient, but couples the wire format to your Java class shape |
| Used in this app | Yes — the producer explicitly serializes with Jackson, giving full control and matching what the consumer expects to parse manually | Viable alternative; trade-off is tighter coupling between producer and consumer class definitions |
▲ Pitfall
If you use JsonSerializer, Spring Kafka by default adds type-hint headers (e.g. TypeId) to every record so the matching JsonDeserializer knows what Java class to deserialize into. Consumers written in a different language, or that don't trust those headers, need spring.json.use.type.headers=false plus an explicit target type — otherwise cross-team/cross-language consumption breaks in confusing ways.
Different environments need different producer configs (local single broker vs. staging/prod clusters, different acks settings) — Spring Profiles (application-local.yml, application-prod.yml) let you swap these without code changes.
@RestControllerAdvice public class GlobalExceptionHandler { @ExceptionHandler(MethodArgumentNotValidException.class) public ResponseEntity<Map<String,String>> handleValidation(MethodArgumentNotValidException ex) { Map<String,String> errors = new HashMap<>(); ex.getBindingResult().getFieldErrors() .forEach(fe -> errors.put(fe.getField(), fe.getDefaultMessage())); return ResponseEntity.badRequest().body(errors); } }
✓ Quick recap
Why does the PUT endpoint send with a key while POST doesn't? The UPDATE must be ordered relative to the original NEW event for the same libraryEventId — keying pins both to the same partition. Does calling kafkaTemplate.send() block the calling (e.g. web) thread by default? No — it's async, returning a CompletableFuture; you opt into blocking with.get(). What does KafkaAdmin's auto topic creation risk if relied on in production? Uncontrolled topic creation with default settings (wrong partition count/replication) instead of deliberate, reviewed provisioning.
💻 Code example
@RestControllerAdvice public class GlobalExceptionHandler { @ExceptionHandler(MethodArgumentNotValidException.class) public ResponseEntity<Map<String,String>> handleValidation(MethodArgumentNotValidException ex) { Map<String,String> errors = new HashMap<>(); ex.getBindingResult().getFieldErrors() .forEach(fe -> errors.put(fe.getField(), fe.getDefaultMessage())); return ResponseEntity.badRequest().body(errors); } }
Want a visual for this concept?
Generate a diagram tailored to “Building the Producer Microservice” — the AI picks whichever visual (flowchart, comparison, sequence, etc.) best fits.
Sign in to generate a visual →