dilipsundarraj1/kafka-for-developers-using-spring-boot-v2/kafka-consumer-skills

Name: kafka-consumer-skills
Author: dilipsundarraj1

Claude Code and agent skill by dilipsundarraj1/kafka-for-developers-using-spring-boot-v2

Trust tier 2

community

LLM review pending

By dilipsundarraj1
Testing
dilipsundarraj1/kafka-for-developers-using-spring-boot-v2
library-events-consumer-v2/.github/skills/kafka-consumer-skills/SKILL.md
Updated 4 Nov 2026
Version 1.0.01 version recorded

Source check pending

Checking that the SKILL.md file is still in the repository.

Kafka consumer patterns for this Spring Boot project, including JSON deserialization, persistence, and integration testing.

Install

npx vskill@latest install dilipsundarraj1/kafka-for-developers-using-spring-boot-v2/kafka-consumer-skills

SKILL.md

Kafka Consumer Skill

Scope: Creating a new Kafka consumer that listens to a topic, deserializes JSON into DTOs, processes via a service layer, and persists to PostgreSQL — plus integration tests using Embedded Kafka and Testcontainers (PostgreSQL).
Read AGENTS.md first for project-wide conventions.

Technology Stack

Component	Value
Spring Boot	4.0.x
Java	25
Kafka client	`spring-boot-starter-kafka`
Kafka test	`spring-boot-starter-kafka-test` (provides `@EmbeddedKafka`, `EmbeddedKafkaBroker`)
JSON deserialization	`spring.kafka.consumer.value-deserializer: JsonDeserializer` + type mapping in `application.yml`
JSON serialization (tests)	Jackson 2 `JsonSerializer` (Kafka's serializer still uses `com.fasterxml.jackson`)
JSON (app code)	Jackson 3 — `tools.jackson.databind.ObjectMapper` — not `com.fasterxml`
Persistence	Spring Data JPA + PostgreSQL
Schema management	Flyway (`ddl-auto: none`)
PostgreSQL (tests)	Testcontainers — `@ImportTestcontainers` + `@ServiceConnection`
Kafka (tests)	`@EmbeddedKafka` — in-process broker, no Docker needed
Ack mode	`MANUAL` — consumer calls `acknowledgment.acknowledge()` in `finally` block

Files to Create Per Kafka Consumer

When adding a new Kafka consumer for topic {topic-name} processing {Entity} events:

#	File	Package / Path	Purpose
1	`{Entity}Dto.java`	`com.learnkafka.dto`	Inbound DTO — Java `record` with bean-validation
2	Domain entity (if new)	`com.learnkafka.domain`	JPA entity with audit callbacks
3	Mapper method(s)	`com.learnkafka.dto.LibraryEventMapper`	Static `toEntity(dto)` conversion
4	`{Entity}Repository.java`	`com.learnkafka.repository`	`JpaRepository<{Entity}, {IdType}>`
5	`{Entity}Service.java`	`com.learnkafka.service`	`processEvent(ConsumerRecord<K, V>)` — `@Transactional`
6	`{Topic}Consumer.java`	`com.learnkafka.consumer`	`@KafkaListener` + MANUAL ack
7	Consumer config (if custom)	`com.learnkafka.config`	`ConcurrentKafkaListenerContainerFactory` bean
8	`application.yml` entries	`src/main/resources/application.yml`	Consumer group, deserializer, type mapping
9	Flyway migration (if new table)	`src/main/resources/db/migration/V{N}__...`	DDL
10	`{Topic}ConsumerIntegrationTest.java`	`com.learnkafka.consumer` (test)	End-to-end: Embedded Kafka → Consumer → DB assertions
11	`{Entity}ServiceIntegrationTest.java`	`com.learnkafka.service` (test)	Service-only: bypass Kafka, call `processEvent()` directly

1. Inbound DTO — `{Entity}Dto.java`

Java record in com.learnkafka.dto.
Annotate with @NotNull, @NotBlank, @Valid (for nested records).
This is what the JsonDeserializer produces on the consumer side.

Reference — `LibraryEventDto.java`

java

package com.learnkafka.dto;
 
import com.learnkafka.domain.LibraryEventType;
import jakarta.validation.Valid;
import jakarta.validation.constraints.NotNull;
 
public record LibraryEventDto(
        Integer libraryEventId,
 
        @NotNull
        LibraryEventType libraryEventType,
 
        @NotNull
        @Valid
        BookDto book
) {
}

2. Consumer — `{Topic}Consumer.java`

Conventions

`@Component` — not @Service (the consumer is a Kafka infrastructure concern, not business logic).
`@KafkaListener(topics = "{topic-name}")` on the handler method.
MANUAL acknowledgment — method signature includes Acknowledgment parameter.
`acknowledgment.acknowledge()` is called in a finally block — offsets are committed even if processing fails (error handling is delegated to the DefaultErrorHandler in the container factory).
Constructor injection of the service.
Logging — private static final Logger log = LoggerFactory.getLogger(...).

Template

java

package com.learnkafka.consumer;
 
import com.learnkafka.dto.{Entity}Dto;
import com.learnkafka.service.{Entity}Service;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
 
@Component
public class {Topic}Consumer {
 
    private static final Logger log = LoggerFactory.getLogger({Topic}Consumer.class);
 
    private final {Entity}Service {entity}Service;
 
    public {Topic}Consumer({Entity}Service {entity}Service) {
        this.{entity}Service = {entity}Service;
    }
 
    @KafkaListener(topics = "{topic-name}")
    public void onMessage(ConsumerRecord<Integer, {Entity}Dto> consumerRecord,
                          Acknowledgment acknowledgment) {
        log.info("ConsumerRecord : {}", consumerRecord);
        try {
            {entity}Service.processEvent(consumerRecord);
        } finally {
            acknowledgment.acknowledge();
        }
    }
}

Reference — `LibraryEventsConsumer.java`

java

@Component
public class LibraryEventsConsumer {
 
    private static final Logger log = LoggerFactory.getLogger(LibraryEventsConsumer.class);
    private final LibraryEventService libraryEventService;
 
    public LibraryEventsConsumer(LibraryEventService libraryEventService) {
        this.libraryEventService = libraryEventService;
    }
 
    @KafkaListener(topics = "library-events")
    public void onMessage(ConsumerRecord<Integer, LibraryEventDto> consumerRecord,
                          Acknowledgment acknowledgment) {
        log.info("ConsumerRecord : {}", consumerRecord);
        try {
            libraryEventService.processEvent(consumerRecord);
        } finally {
            acknowledgment.acknowledge();
        }
    }
}

3. Consumer Config — `{Topic}ConsumerConfig.java`

Only create a new config class if the new consumer requires a different container factory (e.g., different ack mode, concurrency, error handler). If the existing LibraryEventsConsumerConfig works, reuse it.

Conventions

`@Configuration` + `@EnableKafka`.
Bean returns KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<K, V>>.
Default ack mode is `MANUAL` — set via factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL).

Template

java

package com.learnkafka.config;
 
import com.learnkafka.dto.{Entity}Dto;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
 
@Configuration
@EnableKafka
public class {Topic}ConsumerConfig {
 
    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, {Entity}Dto>>
            kafkaListenerContainerFactory(ConsumerFactory<Integer, {Entity}Dto> consumerFactory) {
 
        var factory = new ConcurrentKafkaListenerContainerFactory<Integer, {Entity}Dto>();
        factory.setConsumerFactory(consumerFactory);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        return factory;
    }
}

Reference — `LibraryEventsConsumerConfig.java`

java

@Configuration
@EnableKafka
public class LibraryEventsConsumerConfig {
 
    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, LibraryEventDto>>
            kafkaListenerContainerFactory(ConsumerFactory<Integer, LibraryEventDto> consumerFactory) {
 
        var factory = new ConcurrentKafkaListenerContainerFactory<Integer, LibraryEventDto>();
        factory.setConsumerFactory(consumerFactory);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        return factory;
    }
}

4. Service — `{Entity}Service.java`

Conventions

`@Service` annotation.
`@Transactional` on processEvent().
Accepts ConsumerRecord<K, V> — extracts the DTO via consumerRecord.value().
Uses LibraryEventMapper for DTO → Entity conversion.
Save order matters for bidirectional OneToOne: save the parent entity first (to get the DB-generated ID), then save the child entity with the FK set. Temporarily null out the child reference to avoid cascade issues on the first save.
Constructor injection — no @Autowired.
Logging — LoggerFactory.getLogger(...).

Template

java

package com.learnkafka.service;
 
import com.learnkafka.domain.{ChildEntity};
import com.learnkafka.domain.{ParentEntity};
import com.learnkafka.dto.{Entity}Dto;
import com.learnkafka.dto.LibraryEventMapper;
import com.learnkafka.repository.{ChildEntity}Repository;
import com.learnkafka.repository.{ParentEntity}Repository;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
 
@Service
public class {Entity}Service {
 
    private static final Logger log = LoggerFactory.getLogger({Entity}Service.class);
 
    private final {ParentEntity}Repository {parentEntity}Repository;
    private final {ChildEntity}Repository {childEntity}Repository;
 
    public {Entity}Service({ParentEntity}Repository {parentEntity}Repository,
                           {ChildEntity}Repository {childEntity}Repository) {
        this.{parentEntity}Repository = {parentEntity}Repository;
        this.{childEntity}Repository = {childEntity}Repository;
    }
 
    @Transactional
    public void processEvent(ConsumerRecord<Integer, {Entity}Dto> consumerRecord) {
        {Entity}Dto dto = consumerRecord.value();
        log.info("{Entity}Dto : {}", dto);
 
        {ParentEntity} parentEntity = LibraryEventMapper.toEntity(dto);
 
        // Save parent first — it has @GeneratedValue(IDENTITY)
        parentEntity.set{ChildEntity}(null); // detach child to avoid cascade issues
        {ParentEntity} savedParent = {parentEntity}Repository.save(parentEntity);

This is the start of the file. Read the full SKILL.md on GitHub.

Security

Security scanPassed, 100 of 100Pattern checks on the skill's files.
Model reviewNot runA model review is scheduled for the next rescan.

See the full security report

Quality evaluation

Task evaluationNot runNo evaluation yet. Run one in Skill Studio.

See evaluation results

Source

SKILL.md in the repositoryCheckinglibrary-events-consumer-v2/.github/skills/kafka-consumer-skills/SKILL.md

Popularity

Repository stars: 114
Repository forks: 215
7-day trend score: 0

Stars and forks are for the whole GitHub repository, not this skill alone. Not refreshed yet.

Works with

Any agent vskill installs to: 56 targets in all. These 39 have a page on this site.

Claude Code
Cursor
GitHub Copilot
Windsurf
Codex
Gemini CLI
Cline
Amp
Roo Code
Goose
Continue
Kilo Code
Junie
Trae
OpenHands
Qwen Code
AI Agent
Tabnine Chat
Sourcegraph Cody
Amazon Q
Firebase Genkit
Sweep
Devin
Phind
Mentat
Claude Desktop
Aider
GPT Pilot
Plandex
bolt.new
v0
Replit Agent
Anse
TypingMind
BetterChatGPT
LibreChat
Warp AI
Pieces
Fig AI

Embed badge

[![vskill VERIFIED](https://verified-skill.com/api/v1/skills/dilipsundarraj1/kafka-for-developers-using-spring-boot-v2/kafka-consumer-skills/badge)](https://verified-skill.com/skills/dilipsundarraj1/kafka-for-developers-using-spring-boot-v2/kafka-consumer-skills)

Something wrong with this skill or its page? Tell us.

Install

SKILL.md

Kafka Consumer Skill

Technology Stack

Files to Create Per Kafka Consumer

1. Inbound DTO — {Entity}Dto.java

Reference — LibraryEventDto.java

2. Consumer — {Topic}Consumer.java

Conventions

Template

Reference — LibraryEventsConsumer.java

3. Consumer Config — {Topic}ConsumerConfig.java

Conventions

Template

Reference — LibraryEventsConsumerConfig.java

4. Service — {Entity}Service.java

Conventions

Template

Security

Quality evaluation

Source

Popularity

Works with

1. Inbound DTO — `{Entity}Dto.java`

Reference — `LibraryEventDto.java`

2. Consumer — `{Topic}Consumer.java`

Reference — `LibraryEventsConsumer.java`

3. Consumer Config — `{Topic}ConsumerConfig.java`

Reference — `LibraryEventsConsumerConfig.java`

4. Service — `{Entity}Service.java`