Kafka Configuration

Preflight: Spring MCP

This skill is part of the Spring Agent Toolkit and is designed to work with the Spring MCP server (provided by the Amplicode IntelliJ plugin). Before doing anything else, check your tool list for any Spring MCP tool — they are exposed under the amplicode MCP server (e.g. get_project_summary, list_module_dependencies, list_application_properties_files); harnesses that flatten MCP tools into the tool list use the mcp__amplicode__ prefix on the same names.

• If at least one Amplicode tool is available — MCP is connected. Proceed with the skill below.
• If none are available — stop and invoke the `amplicode-install` skill (bundled with the Spring Agent Toolkit). It installs the Amplicode plugin and walks the user through the «Настроить Spring Agent» welcome-screen button + MCP-client restart. After it completes, the MCP tools become available — resume this skill.
• If amplicode-install is not registered in your skill list, tell the user (in their language): "This skill needs the Amplicode IntelliJ plugin and its MCP server. Install it from https://amplicode.ru/marketplace into IntelliJ IDEA Ultimate/Community or GigaIDE, open any project, click «Настроить Spring Agent» on the Amplicode welcome screen, then restart your MCP client."

---

Kafka Configuration

Wires Spring Boot's Kafka starter through application.properties / application.yml and, optionally, a generated KafkaConfiguration class. Ensures the starter dependency is on the module classpath.

CRITICAL: Code ONLY from `examples/` files. If no matching example — STOP and ask user.

CRITICAL: For questions with a fixed set of choices, prefer `AskUserQuestion` > its analogue > plain text list.

CRITICAL: Read the conversation context BEFORE running Step 1. Half the questions in Steps 2–3 may already be answered.

Two paths (Step 3 picks)

• Path A — `serializerSource = properties` (default). All six spring.kafka.* keys go into the property source: bootstrap-servers, consumer.group-id, producer key/value serializer, consumer key/value deserializer. KafkaAutoConfiguration creates ProducerFactory, ConsumerFactory, KafkaTemplate (bean kafkaTemplate), and kafkaListenerContainerFactory from them.
• Path B — `serializerSource = @Configuration`. Only bootstrap-servers + consumer.group-id go into the property source. A KafkaConfiguration class is generated with four beans: {prefix}ProducerFactory, {prefix}KafkaTemplate, {prefix}ConsumerFactory, kafkaListenerContainerFactory. {prefix} is the lowercased simple name of producerValueType. The listener factory bean name is fixed — autoconfig backs off via @ConditionalOnMissingBean(name = "kafkaListenerContainerFactory"), any other name leaves the autoconfig factory in place and silently routes @KafkaListener to the wrong one.

Defaults

Smart defaults. If the user says "use defaults" / "all defaults" / "minimal configuration" → skip every Always ask = NO question. Only ask the mandatory ones (consumerGroup, serializerSource, and any type the user already mentioned non-default for).

Smart answer recognition. When the user provides a value directly ("for OrderEvent messages", "group orders"), accept it without asking again. Multiple answers in one message → accept all.

Batch questions. Group related questions into one AskUserQuestion call (up to 4 questions). Recommended option first, (Recommended) in its label.

Decision-making — context first, then ask

For every input: try to derive from get_project_summary, list_module_dependencies, list_application_properties_files, prior turns, the user's prompt. Only ask when context yields no clear default.

1. Context unambiguous → decide silently. language, bootVersion, mainPackage, bootstrapServers = localhost:9092, single-file propsFile. 2. Strong signal → one-line confirmation. State decision + alternatives; user can accept silently. 3. No clear default → `AskUserQuestion` with recommended option first. 4. Empty for a critical input → ask plainly. consumerGroup, serializerSource, custom type FQNs.

Step 0 — Conversation context (mental, no tool calls)

Re-read the user's prompt and prior turns; tick off everything already stated:

Tick → skip the corresponding question. Do not announce Step 0.

Step 1 — Gather context (parallel MCP calls + file reads)

After the calls, Read each path in propsFiles and parse any existing spring.kafka.* keys into existingProps.

Multi-module selection. If modules has more than one entry: 1. Run list_module_dependencies + list_application_properties_files for each module in parallel. 2. Score every module: +1 if its deps contain org.springframework.boot:spring-boot-starter-kafka; +1 if its property files contain any spring.kafka.* key. 3. Exactly one module with score ≥ 1 → select it silently as moduleName. 4. Two or more, or all-zero → ask the user which module. Then re-run the module-scoped tools for the chosen one.

Derived:

• bootBranch — boot3 (Boot 3.0/3.1), boot3.2 (3.2/3.3/3.4/3.5), boot4 (≥4.0). Used in Step 5 + Step 6.
• kafkaAlreadyPresent — presentDeps contains org.springframework.boot:spring-boot-starter-kafka. Skip Step 5b if true.
• singlePropsFile — len(propsFiles) == 1. Skip the props-file question in Step 3 if true.
• existingBootstrapServers — value of spring.kafka.bootstrap-servers from existingProps if present, else null. Used silently in Step 2.
• existingConsumerGroup — value of spring.kafka.consumer.group-id from existingProps if present, else null.
• existingKafkaConfig — entries in existingBeans belonging to @Configuration classes that declare any of KafkaTemplate, ProducerFactory, ConsumerFactory, *ListenerContainerFactory beans. Each entry carries the class FQN, source file path, and the bean names already declared. Used in Step 4.

Step 2 — All questions in ONE batch

Ask everything in a single AskUserQuestion call (up to 5 questions). Pre-fill defaults from context, skip already-answered:

1. Producer key type? — options: java.lang.String (Recommended), java.lang.Integer, java.lang.Long, java.util.UUID, java.lang.Void, Custom (specify FQN) 2. Producer value type? — same options 3. Consumer group id? — plain text, mandatory. If existingConsumerGroup is non-null → pre-fill silently. 4. Bootstrap servers? — default localhost:9092. Если existingBootstrapServers is non-null → pre-fill silently. 5. Where to describe serializers? — application.properties (Recommended) (Path A) / @Configuration class (Path B)

Consumer types default to producer types (symmetric). Only ask separately if user indicated asymmetric.

If the user says "use defaults" / "all defaults" — skip type questions, default all four to java.lang.String. Only ask consumerGroup, bootstrapServers, serializerSource.

Step 3 — Class target (Path B only)

If existingKafkaConfig is non-empty, apply Decision principle 2 (one-line confirmation), naming the existing class:

Project already has Kafka beans in {class.fqn}. Add new beans there? (Yes/No)

• Yes → classTarget = existing. Reuse the existing class's packageName, className, and source file path. Step 5 then uses Edit to append beans, not Write to create a new file. Inserted bean names that collide with existing ones get a numeric suffix per Step 5 rules.
• No (or existingKafkaConfig was empty) → classTarget = new:
• className — default KafkaConfiguration; on file collision in {packageName}/{className}.{java|kt} append numeric suffix.
• packageName — default mainPackage; ask only if multiple @Configuration packages exist with no clear winner.

Usually answered silently from context.

Step 4 — Write properties + add dependency

4a. Properties

Always write:

spring.kafka.bootstrap-servers={bootstrapServers}
spring.kafka.consumer.group-id={consumerGroup}

If serializerSource = properties, also append the four (de)serializer keys (FQNs from examples/serializer-mapping.md):

spring.kafka.producer.key-serializer={producerKeySerializer}
spring.kafka.producer.value-serializer={producerValueSerializer}
spring.kafka.consumer.key-deserializer={consumerKeyDeserializer}
spring.kafka.consumer.value-deserializer={consumerValueDeserializer}

If consumerKeyType OR consumerValueType is a POJO, also append:

spring.kafka.consumer.properties[spring.json.trusted.packages]={packages}

{packages} = comma-separated packages of the POJO types on the consumer side.

If consumerKeyType is a POJO, also append:

spring.kafka.consumer.properties[spring.json.key.default.type]={consumerKeyTypeFqn}

If consumerValueType is a POJO, also append:

spring.kafka.consumer.properties[spring.json.value.default.type]={consumerValueTypeFqn}

YAML equivalents are in examples/serializer-mapping.md. Overwrite existing keys; never delete unrelated ones.

4b. Dependency

If kafkaAlreadyPresent = true, skip. Otherwise add org.springframework.boot:spring-boot-starter-kafka to buildFile (same artifact id for Boot 3.x and 4.x), then call refresh_build_system_model.

Step 5 — Generate @Configuration (Path B only)

1. Pick examples/configuration-class/{language}-{bootBranch}.md. 2. Substitute variables listed in the template. 3. {prefix} = decapitalize({producerValueType} simple name) — String → string, OrderEvent → orderEvent, UUID → uUID (only first char lowered). 4. Bean name collision rules (check against existingBeans from Step 1 and, for classTarget = existing, against the target class's declared beans):

• For {prefix}ProducerFactory / {prefix}KafkaTemplate / {prefix}ConsumerFactory — if a bean with that exact name exists, append numeric suffix (stringProducerFactory_1, ...).
• Listener factory: default name kafkaListenerContainerFactory. If it already exists, use {prefix}KafkaListenerContainerFactory and warn in Step 6 that @KafkaListener for {producerValueType} must set containerFactory = "{prefix}KafkaListenerContainerFactory" explicitly.

5. Detect indentation:

• classTarget = existing → sample indentation from the target file.
• classTarget = new → .editorconfig → sample any existing source file → fallback 4-space.

6. FQN handling: replace every FQN in the body with its short name; emit one import line per unique FQN after package; skip java.lang and same-package classes. Java grouping: third-party block (sorted alphabetically) + blank line + java.* block (sorted). Kotlin: kotlin.* joins the third-party block. 7. Write step:

• classTarget = new → Write the full file at {module sourceRoot}/{packageName as path}/{className}.{java|kt}.
• classTarget = existing → Read the target file, then Edit to append the four @Bean methods before the class's closing brace and merge required imports into the existing import block (no duplicates). Do not create a new file.

Step 6 — Report

Match the user's conversation language. Include:

• Path taken.
• Files written or edited (paths) and the keys/beans added. For Path B note whether a new class was created or beans were appended to an existing one.
• Dependency: added / already present.
• Pre-filled values from existing project state (when applicable): «Reused existing spring.kafka.bootstrap-servers={existingBootstrapServers} / consumer.group-id={existingConsumerGroup} from {propsFile} — overwrote with the new values only if you provided them.»
• How to use: inject KafkaTemplate<{producerKeyType}, {producerValueType}> (bean kafkaTemplate for Path A, {prefix}KafkaTemplate for Path B). If the listener factory bean is named exactly kafkaListenerContainerFactory, @KafkaListener uses it by default and no explicit containerFactory is needed.
• If a listener factory was name-suffixed: warn that @KafkaListener for {producerValueType} must set containerFactory = "{prefix}KafkaListenerContainerFactory" explicitly.

Anti-hallucination checklist

• [ ] (De)serializer FQNs come from examples/serializer-mapping.md rows matching each of the four type parameters — not from memory.
• [ ] bootstrap-servers + consumer.group-id are written regardless of serializerSource.
• [ ] Path A: the four serializer keys ARE in the property source; no @Configuration is generated.
• [ ] Path B: the four serializer keys are NOT in the property source; they live inside the generated class.
• [ ] Listener factory bean name is exactly kafkaListenerContainerFactory (or {prefix}KafkaListenerContainerFactory only when the default was already taken by a previous generation).
• [ ] Listener factory return type is ConcurrentKafkaListenerContainerFactory<{consumerKeyType}, {consumerValueType}> — never raw.
• [ ] KafkaProperties import matches bootBranch: boot3 / boot3.2 use org.springframework.boot.autoconfigure.kafka; boot4 uses org.springframework.boot.kafka.autoconfigure.
• [ ] boot3.2 calls buildProducerProperties(sslBundles.getIfAvailable()) / buildConsumerProperties(sslBundles.getIfAvailable()); boot3 and boot4 pass no argument.
• [ ] If a POJO consumer type: spring.json.trusted.packages (Path A) or JsonDeserializer.TRUSTED_PACKAGES / JacksonJsonDeserializer.TRUSTED_PACKAGES (Path B) is set to the relevant package(s) — never *, never the FQN.
• [ ] If consumerKeyType is a POJO: spring.json.key.default.type (Path A) or JsonDeserializer.KEY_DEFAULT_TYPE / JacksonJsonDeserializer.KEY_DEFAULT_TYPE (Path B) is set to the key type FQN.
• [ ] If consumerValueType is a POJO: spring.json.value.default.type (Path A) or JsonDeserializer.VALUE_DEFAULT_TYPE / JacksonJsonDeserializer.VALUE_DEFAULT_TYPE (Path B) is set to the value type FQN.
• [ ] Indentation matches detected project style.

Kafka @Configuration — Java, Spring Boot 3.2+

Generated when (language, bootBranch) = (java, boot3.2) and serializerSource = @Configuration.

Variables

• {packageName} — from Step 4.
• {className} — from Step 4.
• {prefix} — decapitalize({producerValueType} simple name).
• {producerKeyType} / {producerValueType} / {consumerKeyType} / {consumerValueType} — simple names.
• {producerKeySerializer} / {producerValueSerializer} / {consumerKeyDeserializer} / {consumerValueDeserializer} — short class names; resolve FQNs via examples/serializer-mapping.md.
• {jsonConsumerPropertiesLines} — present only when consumerKeyType OR consumerValueType is a POJO; otherwise the lines are omitted. Always include TRUSTED_PACKAGES when any consumer side is a POJO. Include KEY_DEFAULT_TYPE only when consumerKeyType is a POJO. Include VALUE_DEFAULT_TYPE only when consumerValueType is a POJO:

  consumerProperties.put(JsonDeserializer.TRUSTED_PACKAGES, "{packages}");
  consumerProperties.put(JsonDeserializer.KEY_DEFAULT_TYPE, "{consumerKeyTypeFqn}");
  consumerProperties.put(JsonDeserializer.VALUE_DEFAULT_TYPE, "{consumerValueTypeFqn}");

• {listenerFactoryName} — kafkaListenerContainerFactory by default; {prefix}KafkaListenerContainerFactory only on name collision.

Code

package {packageName};

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.ssl.SslBundles;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.Map;

@Configuration
public class {className} {

    @Bean
    DefaultKafkaProducerFactory<{producerKeyType}, {producerValueType}> {prefix}ProducerFactory(
            KafkaProperties properties,
            ObjectProvider<SslBundles> sslBundles) {
        Map<String, Object> producerProperties = properties.buildProducerProperties(sslBundles.getIfAvailable());
        producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, {producerKeySerializer}.class);
        producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, {producerValueSerializer}.class);
        return new DefaultKafkaProducerFactory<>(producerProperties);
    }

    @Bean
    KafkaTemplate<{producerKeyType}, {producerValueType}> {prefix}KafkaTemplate(
            DefaultKafkaProducerFactory<{producerKeyType}, {producerValueType}> {prefix}ProducerFactory) {
        return new KafkaTemplate<>({prefix}ProducerFactory);
    }

    @Bean
    ConsumerFactory<{consumerKeyType}, {consumerValueType}> {prefix}ConsumerFactory(
            KafkaProperties properties,
            ObjectProvider<SslBundles> sslBundles) {
        Map<String, Object> consumerProperties = properties.buildConsumerProperties(sslBundles.getIfAvailable());
        consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, {consumerKeyDeserializer}.class);
        consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, {consumerValueDeserializer}.class);
        {jsonConsumerPropertiesLines}
        return new DefaultKafkaConsumerFactory<>(consumerProperties);
    }

    @Bean
    ConcurrentKafkaListenerContainerFactory<{consumerKeyType}, {consumerValueType}> {listenerFactoryName}(
            ConsumerFactory<{consumerKeyType}, {consumerValueType}> {prefix}ConsumerFactory) {
        ConcurrentKafkaListenerContainerFactory<{consumerKeyType}, {consumerValueType}> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory({prefix}ConsumerFactory);
        return factory;
    }

}

Extra imports

Add an import line for each FQN that came in through a substitution: the four (de)serializer FQNs from examples/serializer-mapping.md, any POJO type FQNs, and org.springframework.kafka.support.serializer.JsonDeserializer when {jsonConsumerPropertiesLines} is present.

Kafka @Configuration — Java, Spring Boot 3.0 / 3.1

Generated when (language, bootBranch) = (java, boot3) and serializerSource = @Configuration.

Variables

• {packageName} — from Step 4.
• {className} — from Step 4.
• {prefix} — decapitalize({producerValueType} simple name).
• {producerKeyType} / {producerValueType} / {consumerKeyType} / {consumerValueType} — simple names.
• {producerKeySerializer} / {producerValueSerializer} / {consumerKeyDeserializer} / {consumerValueDeserializer} — short class names; resolve FQNs via examples/serializer-mapping.md.
• {jsonConsumerPropertiesLines} — present only when consumerKeyType OR consumerValueType is a POJO; otherwise the lines are omitted. Always include TRUSTED_PACKAGES when any consumer side is a POJO. Include KEY_DEFAULT_TYPE only when consumerKeyType is a POJO. Include VALUE_DEFAULT_TYPE only when consumerValueType is a POJO:

  consumerProperties.put(JsonDeserializer.TRUSTED_PACKAGES, "{packages}");
  consumerProperties.put(JsonDeserializer.KEY_DEFAULT_TYPE, "{consumerKeyTypeFqn}");
  consumerProperties.put(JsonDeserializer.VALUE_DEFAULT_TYPE, "{consumerValueTypeFqn}");

• {listenerFactoryName} — kafkaListenerContainerFactory by default; {prefix}KafkaListenerContainerFactory only on name collision.

Code

package {packageName};

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.Map;

@Configuration
public class {className} {

    @Bean
    DefaultKafkaProducerFactory<{producerKeyType}, {producerValueType}> {prefix}ProducerFactory(KafkaProperties properties) {
        Map<String, Object> producerProperties = properties.buildProducerProperties();
        producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, {producerKeySerializer}.class);
        producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, {producerValueSerializer}.class);
        return new DefaultKafkaProducerFactory<>(producerProperties);
    }

    @Bean
    KafkaTemplate<{producerKeyType}, {producerValueType}> {prefix}KafkaTemplate(
            DefaultKafkaProducerFactory<{producerKeyType}, {producerValueType}> {prefix}ProducerFactory) {
        return new KafkaTemplate<>({prefix}ProducerFactory);
    }

    @Bean
    ConsumerFactory<{consumerKeyType}, {consumerValueType}> {prefix}ConsumerFactory(KafkaProperties properties) {
        Map<String, Object> consumerProperties = properties.buildConsumerProperties();
        consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, {consumerKeyDeserializer}.class);
        consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, {consumerValueDeserializer}.class);
        {jsonConsumerPropertiesLines}
        return new DefaultKafkaConsumerFactory<>(consumerProperties);
    }

    @Bean
    ConcurrentKafkaListenerContainerFactory<{consumerKeyType}, {consumerValueType}> {listenerFactoryName}(
            ConsumerFactory<{consumerKeyType}, {consumerValueType}> {prefix}ConsumerFactory) {
        ConcurrentKafkaListenerContainerFactory<{consumerKeyType}, {consumerValueType}> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory({prefix}ConsumerFactory);
        return factory;
    }

}

Extra imports

Add an import line for each FQN that came in through a substitution: the four (de)serializer FQNs from examples/serializer-mapping.md, any POJO type FQNs, and org.springframework.kafka.support.serializer.JsonDeserializer when {jsonConsumerPropertiesLines} is present.

Kafka @Configuration — Java, Spring Boot 4

Generated when (language, bootBranch) = (java, boot4) and serializerSource = @Configuration.

Variables

• {packageName} — from Step 4.
• {className} — from Step 4.
• {prefix} — decapitalize({producerValueType} simple name).
• {producerKeyType} / {producerValueType} / {consumerKeyType} / {consumerValueType} — simple names.
• {producerKeySerializer} / {producerValueSerializer} / {consumerKeyDeserializer} / {consumerValueDeserializer} — short class names; resolve FQNs via examples/serializer-mapping.md.
• {jsonConsumerPropertiesLines} — present only when consumerKeyType OR consumerValueType is a POJO; otherwise the lines are omitted. Always include TRUSTED_PACKAGES when any consumer side is a POJO. Include KEY_DEFAULT_TYPE only when consumerKeyType is a POJO. Include VALUE_DEFAULT_TYPE only when consumerValueType is a POJO:

  consumerProperties.put(JacksonJsonDeserializer.TRUSTED_PACKAGES, "{packages}");
  consumerProperties.put(JacksonJsonDeserializer.KEY_DEFAULT_TYPE, "{consumerKeyTypeFqn}");
  consumerProperties.put(JacksonJsonDeserializer.VALUE_DEFAULT_TYPE, "{consumerValueTypeFqn}");

• {listenerFactoryName} — kafkaListenerContainerFactory by default; {prefix}KafkaListenerContainerFactory only when the default name already exists in the module.

Code

package {packageName};

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.boot.kafka.autoconfigure.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.Map;

@Configuration
public class {className} {

    @Bean
    DefaultKafkaProducerFactory<{producerKeyType}, {producerValueType}> {prefix}ProducerFactory(KafkaProperties properties) {
        Map<String, Object> producerProperties = properties.buildProducerProperties();
        producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, {producerKeySerializer}.class);
        producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, {producerValueSerializer}.class);
        return new DefaultKafkaProducerFactory<>(producerProperties);
    }

    @Bean
    KafkaTemplate<{producerKeyType}, {producerValueType}> {prefix}KafkaTemplate(
            DefaultKafkaProducerFactory<{producerKeyType}, {producerValueType}> {prefix}ProducerFactory) {
        return new KafkaTemplate<>({prefix}ProducerFactory);
    }

    @Bean
    ConsumerFactory<{consumerKeyType}, {consumerValueType}> {prefix}ConsumerFactory(KafkaProperties properties) {
        Map<String, Object> consumerProperties = properties.buildConsumerProperties();
        consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, {consumerKeyDeserializer}.class);
        consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, {consumerValueDeserializer}.class);
        {jsonConsumerPropertiesLines}
        return new DefaultKafkaConsumerFactory<>(consumerProperties);
    }

    @Bean
    ConcurrentKafkaListenerContainerFactory<{consumerKeyType}, {consumerValueType}> {listenerFactoryName}(
            ConsumerFactory<{consumerKeyType}, {consumerValueType}> {prefix}ConsumerFactory) {
        ConcurrentKafkaListenerContainerFactory<{consumerKeyType}, {consumerValueType}> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory({prefix}ConsumerFactory);
        return factory;
    }

}

Extra imports

Add an import line for each FQN that came in through a substitution: the four (de)serializer FQNs from examples/serializer-mapping.md, any POJO type FQNs, and org.springframework.kafka.support.serializer.JacksonJsonDeserializer when {jsonConsumerPropertiesLines} is present.

Kafka @Configuration — Kotlin, Spring Boot 3.2+

Generated when (language, bootBranch) = (kotlin, boot3.2) and serializerSource = @Configuration.

Variables

• {packageName} — from Step 4.
• {className} — from Step 4.
• {prefix} — decapitalize({producerValueType} simple name).
• {producerKeyType} / {producerValueType} / {consumerKeyType} / {consumerValueType} — simple names.
• {producerKeySerializer} / {producerValueSerializer} / {consumerKeyDeserializer} / {consumerValueDeserializer} — short class names; resolve FQNs via examples/serializer-mapping.md.
• {jsonConsumerPropertiesLines} — present only when consumerKeyType OR consumerValueType is a POJO; otherwise the lines are omitted. Always include TRUSTED_PACKAGES when any consumer side is a POJO. Include KEY_DEFAULT_TYPE only when consumerKeyType is a POJO. Include VALUE_DEFAULT_TYPE only when consumerValueType is a POJO:

  consumerProperties[JsonDeserializer.TRUSTED_PACKAGES] = "{packages}"
  consumerProperties[JsonDeserializer.KEY_DEFAULT_TYPE] = "{consumerKeyTypeFqn}"
  consumerProperties[JsonDeserializer.VALUE_DEFAULT_TYPE] = "{consumerValueTypeFqn}"

• {listenerFactoryName} — kafkaListenerContainerFactory by default; {prefix}KafkaListenerContainerFactory only on name collision.

Code

package {packageName}

import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.producer.ProducerConfig
import org.springframework.beans.factory.ObjectProvider
import org.springframework.boot.autoconfigure.kafka.KafkaProperties
import org.springframework.boot.ssl.SslBundles
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory
import org.springframework.kafka.core.ConsumerFactory
import org.springframework.kafka.core.DefaultKafkaConsumerFactory
import org.springframework.kafka.core.DefaultKafkaProducerFactory
import org.springframework.kafka.core.KafkaTemplate

@Configuration
class {className} {

    @Bean
    fun {prefix}ProducerFactory(
        properties: KafkaProperties,
        sslBundles: ObjectProvider<SslBundles>,
    ): DefaultKafkaProducerFactory<{producerKeyType}, {producerValueType}> {
        val producerProperties = properties.buildProducerProperties(sslBundles.getIfAvailable())
        producerProperties[ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG] = {producerKeySerializer}::class.java
        producerProperties[ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG] = {producerValueSerializer}::class.java
        return DefaultKafkaProducerFactory(producerProperties)
    }

    @Bean
    fun {prefix}KafkaTemplate(
        {prefix}ProducerFactory: DefaultKafkaProducerFactory<{producerKeyType}, {producerValueType}>,
    ): KafkaTemplate<{producerKeyType}, {producerValueType}> =
        KafkaTemplate({prefix}ProducerFactory)

    @Bean
    fun {prefix}ConsumerFactory(
        properties: KafkaProperties,
        sslBundles: ObjectProvider<SslBundles>,
    ): ConsumerFactory<{consumerKeyType}, {consumerValueType}> {
        val consumerProperties = properties.buildConsumerProperties(sslBundles.getIfAvailable())
        consumerProperties[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = {consumerKeyDeserializer}::class.java
        consumerProperties[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = {consumerValueDeserializer}::class.java
        {jsonConsumerPropertiesLines}
        return DefaultKafkaConsumerFactory(consumerProperties)
    }

    @Bean
    fun {listenerFactoryName}(
        {prefix}ConsumerFactory: ConsumerFactory<{consumerKeyType}, {consumerValueType}>,
    ): ConcurrentKafkaListenerContainerFactory<{consumerKeyType}, {consumerValueType}> {
        val factory = ConcurrentKafkaListenerContainerFactory<{consumerKeyType}, {consumerValueType}>()
        factory.consumerFactory = {prefix}ConsumerFactory
        return factory
    }
}

Extra imports

Add an import line for each FQN that came in through a substitution: the four (de)serializer FQNs from examples/serializer-mapping.md, any POJO type FQNs, and org.springframework.kafka.support.serializer.JsonDeserializer when {jsonConsumerPropertiesLines} is present.

Kafka @Configuration — Kotlin, Spring Boot 3.0 / 3.1

Generated when (language, bootBranch) = (kotlin, boot3) and serializerSource = @Configuration.

Variables

• {packageName} — from Step 4.
• {className} — from Step 4.
• {prefix} — decapitalize({producerValueType} simple name).
• {producerKeyType} / {producerValueType} / {consumerKeyType} / {consumerValueType} — simple names.
• {producerKeySerializer} / {producerValueSerializer} / {consumerKeyDeserializer} / {consumerValueDeserializer} — short class names; resolve FQNs via examples/serializer-mapping.md.
• {jsonConsumerPropertiesLines} — present only when consumerKeyType OR consumerValueType is a POJO; otherwise the lines are omitted. Always include TRUSTED_PACKAGES when any consumer side is a POJO. Include KEY_DEFAULT_TYPE only when consumerKeyType is a POJO. Include VALUE_DEFAULT_TYPE only when consumerValueType is a POJO:

  consumerProperties[JsonDeserializer.TRUSTED_PACKAGES] = "{packages}"
  consumerProperties[JsonDeserializer.KEY_DEFAULT_TYPE] = "{consumerKeyTypeFqn}"
  consumerProperties[JsonDeserializer.VALUE_DEFAULT_TYPE] = "{consumerValueTypeFqn}"

• {listenerFactoryName} — kafkaListenerContainerFactory by default; {prefix}KafkaListenerContainerFactory only on name collision.

Code

package {packageName}

import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.producer.ProducerConfig
import org.springframework.boot.autoconfigure.kafka.KafkaProperties
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory
import org.springframework.kafka.core.ConsumerFactory
import org.springframework.kafka.core.DefaultKafkaConsumerFactory
import org.springframework.kafka.core.DefaultKafkaProducerFactory
import org.springframework.kafka.core.KafkaTemplate

@Configuration
class {className} {

    @Bean
    fun {prefix}ProducerFactory(properties: KafkaProperties): DefaultKafkaProducerFactory<{producerKeyType}, {producerValueType}> {
        val producerProperties = properties.buildProducerProperties()
        producerProperties[ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG] = {producerKeySerializer}::class.java
        producerProperties[ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG] = {producerValueSerializer}::class.java
        return DefaultKafkaProducerFactory(producerProperties)
    }

    @Bean
    fun {prefix}KafkaTemplate(
        {prefix}ProducerFactory: DefaultKafkaProducerFactory<{producerKeyType}, {producerValueType}>,
    ): KafkaTemplate<{producerKeyType}, {producerValueType}> =
        KafkaTemplate({prefix}ProducerFactory)

    @Bean
    fun {prefix}ConsumerFactory(properties: KafkaProperties): ConsumerFactory<{consumerKeyType}, {consumerValueType}> {
        val consumerProperties = properties.buildConsumerProperties()
        consumerProperties[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = {consumerKeyDeserializer}::class.java
        consumerProperties[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = {consumerValueDeserializer}::class.java
        {jsonConsumerPropertiesLines}
        return DefaultKafkaConsumerFactory(consumerProperties)
    }

    @Bean
    fun {listenerFactoryName}(
        {prefix}ConsumerFactory: ConsumerFactory<{consumerKeyType}, {consumerValueType}>,
    ): ConcurrentKafkaListenerContainerFactory<{consumerKeyType}, {consumerValueType}> {
        val factory = ConcurrentKafkaListenerContainerFactory<{consumerKeyType}, {consumerValueType}>()
        factory.consumerFactory = {prefix}ConsumerFactory
        return factory
    }
}

Extra imports

Add an import line for each FQN that came in through a substitution: the four (de)serializer FQNs from examples/serializer-mapping.md, any POJO type FQNs, and org.springframework.kafka.support.serializer.JsonDeserializer when {jsonConsumerPropertiesLines} is present.

Kafka @Configuration — Kotlin, Spring Boot 4

Generated when (language, bootBranch) = (kotlin, boot4) and serializerSource = @Configuration.

Variables

• {packageName} — from Step 4.
• {className} — from Step 4.
• {prefix} — decapitalize({producerValueType} simple name).
• {producerKeyType} / {producerValueType} / {consumerKeyType} / {consumerValueType} — simple names.
• {producerKeySerializer} / {producerValueSerializer} / {consumerKeyDeserializer} / {consumerValueDeserializer} — short class names; resolve FQNs via examples/serializer-mapping.md.
• {jsonConsumerPropertiesLines} — present only when consumerKeyType OR consumerValueType is a POJO; otherwise the lines are omitted. Always include TRUSTED_PACKAGES when any consumer side is a POJO. Include KEY_DEFAULT_TYPE only when consumerKeyType is a POJO. Include VALUE_DEFAULT_TYPE only when consumerValueType is a POJO:

  consumerProperties[JacksonJsonDeserializer.TRUSTED_PACKAGES] = "{packages}"
  consumerProperties[JacksonJsonDeserializer.KEY_DEFAULT_TYPE] = "{consumerKeyTypeFqn}"
  consumerProperties[JacksonJsonDeserializer.VALUE_DEFAULT_TYPE] = "{consumerValueTypeFqn}"

• {listenerFactoryName} — kafkaListenerContainerFactory by default; {prefix}KafkaListenerContainerFactory only on name collision.

Code

package {packageName}

import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.producer.ProducerConfig
import org.springframework.boot.kafka.autoconfigure.KafkaProperties
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory
import org.springframework.kafka.core.ConsumerFactory
import org.springframework.kafka.core.DefaultKafkaConsumerFactory
import org.springframework.kafka.core.DefaultKafkaProducerFactory
import org.springframework.kafka.core.KafkaTemplate

@Configuration
class {className} {

    @Bean
    fun {prefix}ProducerFactory(properties: KafkaProperties): DefaultKafkaProducerFactory<{producerKeyType}, {producerValueType}> {
        val producerProperties = properties.buildProducerProperties()
        producerProperties[ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG] = {producerKeySerializer}::class.java
        producerProperties[ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG] = {producerValueSerializer}::class.java
        return DefaultKafkaProducerFactory(producerProperties)
    }

    @Bean
    fun {prefix}KafkaTemplate(
        {prefix}ProducerFactory: DefaultKafkaProducerFactory<{producerKeyType}, {producerValueType}>,
    ): KafkaTemplate<{producerKeyType}, {producerValueType}> =
        KafkaTemplate({prefix}ProducerFactory)

    @Bean
    fun {prefix}ConsumerFactory(properties: KafkaProperties): ConsumerFactory<{consumerKeyType}, {consumerValueType}> {
        val consumerProperties = properties.buildConsumerProperties()
        consumerProperties[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = {consumerKeyDeserializer}::class.java
        consumerProperties[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = {consumerValueDeserializer}::class.java
        {jsonConsumerPropertiesLines}
        return DefaultKafkaConsumerFactory(consumerProperties)
    }

    @Bean
    fun {listenerFactoryName}(
        {prefix}ConsumerFactory: ConsumerFactory<{consumerKeyType}, {consumerValueType}>,
    ): ConcurrentKafkaListenerContainerFactory<{consumerKeyType}, {consumerValueType}> {
        val factory = ConcurrentKafkaListenerContainerFactory<{consumerKeyType}, {consumerValueType}>()
        factory.consumerFactory = {prefix}ConsumerFactory
        return factory
    }
}

Extra imports

Add an import line for each FQN that came in through a substitution: the four (de)serializer FQNs from examples/serializer-mapping.md, any POJO type FQNs, and org.springframework.kafka.support.serializer.JacksonJsonDeserializer when {jsonConsumerPropertiesLines} is present.

Serializer / deserializer mapping

This table drives Step 5a (write properties) and Step 6 (generate @Configuration class). One lookup per type parameter; the skill has four independent type parameters (producerKeyType, producerValueType, consumerKeyType, consumerValueType), each resolved against the same table.

The serializer column is used on the producer side (writes producer.{key|value}-serializer or props.put(..., XxxSerializer.class) inside {prefix}ProducerFactory). The deserializer column is used on the consumer side (writes consumer.{key|value}-deserializer or props.put(..., XxxDeserializer.class) inside {prefix}ConsumerFactory).

Spring Boot's KafkaAutoConfiguration reads the property-source variant through KafkaProperties.buildProducerProperties() / buildConsumerProperties() and applies them to the auto-created ProducerFactory / ConsumerFactory.

Type → (serializer, deserializer) FQN

All FQNs are real classes shipped by org.apache.kafka:kafka-clients (basic types) or org.springframework.kafka:spring-kafka (JSON variants).

JacksonJson* classes exist only in spring-kafka 4.x (verified in spring-kafka repo at tag v4.0.4); on Boot 3.x they are not available and Json* must be used. The TRUSTED_PACKAGES constant resolves to the same property key spring.json.trusted.packages in both versions (verified in JsonDeserializer.java:88 and JacksonJsonDeserializer.java).

Extra JSON consumer properties

spring.json.trusted.packages is required when consumerKeyType OR consumerValueType is a Custom POJO. Without it, JsonDeserializer / JacksonJsonDeserializer rejects incoming records with IllegalArgumentException: The class is not in the trusted packages.

Value = comma-separated packages of the POJO types on the consumer side. Examples:

• consumerValueType = com.example.orders.OrderEvent, consumerKeyType = String → spring.json.trusted.packages=com.example.orders
• consumerKeyType = com.example.ids.OrderId, consumerValueType = com.example.orders.OrderEvent → spring.json.trusted.packages=com.example.ids,com.example.orders
• consumerKeyType = String, consumerValueType = String → no trusted.packages line at all

The producer-side types do not contribute to trusted.packages because JsonSerializer does not validate types on the way out.

spring.json.key.default.type is required when consumerKeyType is a Custom POJO. Value = the FQN of consumerKeyType.

spring.json.value.default.type is required when consumerValueType is a Custom POJO. Value = the FQN of consumerValueType.

These default type properties make POJO deserialization work even when incoming records do not include Spring JSON type headers. Do not add key.default.type for non-POJO keys, and do not add value.default.type for non-POJO values.

.properties syntax:

spring.kafka.consumer.properties[spring.json.trusted.packages]={packages}
spring.kafka.consumer.properties[spring.json.key.default.type]={consumerKeyTypeFqn}
spring.kafka.consumer.properties[spring.json.value.default.type]={consumerValueTypeFqn}

.yml syntax:

spring:
  kafka:
    consumer:
      properties:
        "[spring.json.trusted.packages]": {packages}
        "[spring.json.key.default.type]": {consumerKeyTypeFqn}
        "[spring.json.value.default.type]": {consumerValueTypeFqn}

Only emit the default type line for the POJO side it applies to. The bracketed key form is required because the property names contain dots. Both .properties and .yml flavours work in any Boot 3.x and 4.x version — verified against Spring Boot 4 reference docs (messaging/kafka.adoc).

Property block template — Path A (serializerSource = properties)

Final block to append (or overwrite per-key) into the chosen property source. Substitute {...} placeholders from the lookups above and the user's answers in Step 3.

.properties:

spring.kafka.bootstrap-servers={bootstrapServers}
spring.kafka.consumer.group-id={consumerGroup}
spring.kafka.producer.key-serializer={producerKeySerializer}
spring.kafka.producer.value-serializer={producerValueSerializer}
spring.kafka.consumer.key-deserializer={consumerKeyDeserializer}
spring.kafka.consumer.value-deserializer={consumerValueDeserializer}

If consumerKeyType OR consumerValueType is a Custom POJO append:

spring.kafka.consumer.properties[spring.json.trusted.packages]={packages}

If consumerKeyType is a Custom POJO append:

spring.kafka.consumer.properties[spring.json.key.default.type]={consumerKeyTypeFqn}

If consumerValueType is a Custom POJO append:

spring.kafka.consumer.properties[spring.json.value.default.type]={consumerValueTypeFqn}

.yml:

spring:
  kafka:
    bootstrap-servers: {bootstrapServers}
    producer:
      key-serializer: {producerKeySerializer}
      value-serializer: {producerValueSerializer}
    consumer:
      group-id: {consumerGroup}
      key-deserializer: {consumerKeyDeserializer}
      value-deserializer: {consumerValueDeserializer}

If consumerKeyType OR consumerValueType is a Custom POJO add under consumer::

      properties:
        "[spring.json.trusted.packages]": {packages}

If consumerKeyType is a Custom POJO add under consumer.properties:

        "[spring.json.key.default.type]": {consumerKeyTypeFqn}

If consumerValueType is a Custom POJO add under consumer.properties:

        "[spring.json.value.default.type]": {consumerValueTypeFqn}

Property block template — Path B (serializerSource = @Configuration)

Only two keys go into the property source. The four serializer FQNs are written inside the generated class (see configuration-class/{language}-{bootBranch}.md).

.properties:

spring.kafka.bootstrap-servers={bootstrapServers}
spring.kafka.consumer.group-id={consumerGroup}

.yml:

spring:
  kafka:
    bootstrap-servers: {bootstrapServers}
    consumer:
      group-id: {consumerGroup}

spring.json.trusted.packages for Path B is set in code via props.put(JsonDeserializer.TRUSTED_PACKAGES, "{packages}") (or JacksonJsonDeserializer.TRUSTED_PACKAGES on boot4) inside {prefix}ConsumerFactory — see the variant template.

spring.json.key.default.type and spring.json.value.default.type for Path B are set in code via JsonDeserializer.KEY_DEFAULT_TYPE / VALUE_DEFAULT_TYPE (or JacksonJsonDeserializer.KEY_DEFAULT_TYPE / VALUE_DEFAULT_TYPE on boot4) for the POJO consumer side(s).

Behaviour on existing keys

If a key is already present in the target file, overwrite its value. Do not append a duplicate key. Other unrelated keys must be left untouched.