Feature/kafka client contract

README.md

@@ -42,12 +42,14 @@ Example usage in your `pom.xml`:
                     <packageName>my.package.path.model</packageName>
                 </models>
                 <clients>
-                    <springKafka>
+                    <kafka>
                         <packageName>my.package.path.client</packageName>
                         <!-- Optional: defaults to models.packageName when models are configured -->
                         <modelPackageName>my.package.path.model</modelPackageName>
-                        <mode>simple</mode> <!-- options: full, simple - default simple -->
-                    </springKafka>
+                        <springKafka>
+                            <enabled>true</enabled>
+                        </springKafka>
+                    </kafka>
                 </clients>
                 <schemas>
                     <avroProjection>
@@ -78,11 +80,13 @@ asyncapiGenerate {
         packageName.set("my.package.path.model")
     }
     clients {
-        springKafka {
+        kafka {
             packageName.set("my.package.path.client")
             // Optional: defaults to models.packageName when models are configured
             modelPackageName.set("my.package.path.model")
-            mode.set("simple") // options: full, simple - default simple
+            springKafka {
+                enabled.set(true)
+            }
         }
     }
     schemas {
@@ -111,11 +115,13 @@ asyncapiGenerate {
         packageName = 'my.package.path.model'
     }
     clients {
-        springKafka {
+        kafka {
             packageName = 'my.package.path.client'
             // Optional: defaults to models.packageName when models are configured
             modelPackageName = 'my.package.path.model'
-            mode = 'simple' // options: full, simple - default simple
+            springKafka {
+                enabled = true
+            }
         }
     }
     schemas {
@@ -301,24 +307,21 @@ Generated Java Protobuf message sources are produced by running `protoc` during
 
 ### Spring Kafka Clients
 
-Spring Kafka output is configured under `clients.springKafka`.
+Spring Kafka output is configured under `clients.kafka.springKafka`.
 
-Generated Spring Kafka clients use `models.packageName` for payload model types by default. If models are generated elsewhere, configure `clients.springKafka.modelPackageName` to point the client API at that package without generating model output in the same execution.
+Generated Spring Kafka clients use `models.packageName` for payload model types by default. If models are generated elsewhere, configure `clients.kafka.modelPackageName` to point the client API at that package without generating model output in the same execution.
 
-For native Avro message payloads, generated Spring Kafka clients use the Java type declared by the Avro schema namespace and name. For example, a native Avro schema with `namespace: com.example.avro` and `name: UserCreated` is used as `com.example.avro.UserCreated` in generated producer, consumer, listener, and handler APIs.
+Kafka client configuration can also be narrowed by capability. `clients.kafka.headers.enabled` controls typed header model generation, and `clients.kafka.springKafka.producer.enabled` / `clients.kafka.springKafka.consumer.enabled` control whether producer and consumer artifacts are generated.
+
+For native Avro message payloads, generated Spring Kafka clients use the Java type declared by the Avro schema namespace and name. For example, a native Avro schema with `namespace: com.example.avro` and `name: UserCreated` is used as `com.example.avro.UserCreated` in generated producer and consumer APIs.
 
 For native Protobuf message payloads, generated Spring Kafka clients use the Java type declared by `option java_package`, or by the Protobuf `package` when `java_package` is omitted. Protobuf client generation requires `option java_multiple_files = true;` so the generated message can be referenced as a top-level Java type. The `.proto` schema must contain a top-level message matching the payload name.
 
 The generator does not configure Kafka Avro or Protobuf serializers and deserializers yet; applications still own that runtime wiring.
 
-The current generator has two modes:
-
-- `mode = "full"` generates Spring Boot-oriented client artifacts. This includes producer classes, listener classes, handler interfaces, an auto-configuration class, and the Spring Boot auto-configuration import resource. Generated producers and listeners use topic property keys, for example `kafka.topics.customerUpdated`, instead of hard-coding topic names directly in the generated source.
-- `mode = "simple"` generates lightweight producer and consumer source artifacts without Spring Boot auto-configuration. The application owns how those generated types are instantiated and connected to Spring Kafka infrastructure.
-
-When `mode` is omitted, the generator uses `simple`.
+Generated Spring Kafka clients are contract-only source artifacts. Producer-oriented channels generate producer wrappers around application-provided `KafkaTemplate` instances and return Spring Kafka `CompletableFuture<SendResult<...>>` values from send methods. Consumer-oriented channels generate consumer interfaces with abstract methods that receive typed `ConsumerRecord` values. The generator does not create Spring Boot auto-configuration, `@KafkaListener` classes, listener containers, serializer configuration, deserializer configuration, or schema registry configuration.
 
-The generated output depends on the channel direction from the AsyncAPI operations. Producer-oriented channels generate producer artifacts. Consumer-oriented channels generate consumer, listener, or handler artifacts depending on the selected mode. When the channel direction is not declared, the generator treats the channel as both producer and consumer.
+The generated output depends on the channel direction from the AsyncAPI operations. Producer-oriented channels generate producer artifacts. Consumer-oriented channels generate consumer artifacts. When the channel direction is not declared, the generator treats the channel as both producer and consumer.
 
 The Spring Kafka client surface is still being redesigned for the next major version. The generated artifacts should currently be treated as a source-generation contract, not as final application architecture guidance.
 

asyncapi-generator-cli/src/main/kotlin/dev/banking/asyncapi/generator/cli/Main.kt

@@ -16,7 +16,6 @@ import dev.banking.asyncapi.generator.core.generator.configuration.GeneratorConf
 import dev.banking.asyncapi.generator.core.generator.configuration.GeneratorConfigurationRequest
 import dev.banking.asyncapi.generator.core.generator.configuration.JavaModelType
 import dev.banking.asyncapi.generator.core.generator.model.GeneratorName
-import dev.banking.asyncapi.generator.core.generator.plan.SpringKafkaClientType
 import dev.banking.asyncapi.generator.core.parser.AsyncApiParser
 import dev.banking.asyncapi.generator.core.registry.AsyncApiRegistry
 import dev.banking.asyncapi.generator.core.validator.AsyncApiValidator
@@ -94,32 +93,48 @@ class AsyncApiGeneratorCli : CliktCommand(name = "asyncapi-generator") {
         "false" to false,
     )
 
-    private val clientsSpringKafka by option(
-        "--clients-spring-kafka",
-        help = "Enable Spring Kafka client generation",
+    private val clientsKafka by option(
+        "--clients-kafka",
+        help = "Enable Kafka client generation",
     ).flag(default = false)
 
-    private val clientsSpringKafkaPackage by option(
-        "--clients-spring-kafka-package",
-        help = "Package for generated Spring Kafka clients",
+    private val clientsKafkaPackage by option(
+        "--clients-kafka-package",
+        help = "Package for generated Kafka clients",
     )
 
-    private val clientsSpringKafkaModelPackage by option(
-        "--clients-spring-kafka-model-package",
-        help = "Package containing model types used by generated Spring Kafka clients",
+    private val clientsKafkaModelPackage by option(
+        "--clients-kafka-model-package",
+        help = "Package containing model types used by generated Kafka clients",
     )
 
-    private val clientsSpringKafkaMode by option(
-        "--clients-spring-kafka-mode",
-        help = "Spring Kafka generation mode (default: simple)",
+    private val clientsKafkaHeaders by option(
+        "--clients-kafka-headers",
+        help = "Generate typed Kafka header models when headers are defined (default: true)",
     ).choice(
-        SpringKafkaClientType.FULL.configurationValue to SpringKafkaClientType.FULL,
-        SpringKafkaClientType.SIMPLE.configurationValue to SpringKafkaClientType.SIMPLE,
+        "true" to true,
+        "false" to false,
+    )
+
+    private val clientsKafkaSpringKafka by option(
+        "--clients-kafka-spring-kafka",
+        help = "Enable Spring Kafka client generation under Kafka clients",
+    ).flag(default = false)
+
+    private val clientsKafkaSpringKafkaProducer by option(
+        "--clients-kafka-spring-kafka-producer",
+        help = "Generate Spring Kafka producer APIs (default: true)",
+    ).choice(
+        "true" to true,
+        "false" to false,
     )
 
-    private val clientsSpringKafkaTopicPropertyPrefix by option(
-        "--clients-spring-kafka-topic-property-prefix",
-        help = "Spring Kafka topic property prefix (default: kafka.topics)",
+    private val clientsKafkaSpringKafkaConsumer by option(
+        "--clients-kafka-spring-kafka-consumer",
+        help = "Generate Spring Kafka consumer APIs (default: true)",
+    ).choice(
+        "true" to true,
+        "false" to false,
     )
 
     private val clientsQuarkusKafka by option(
@@ -220,13 +235,24 @@ class AsyncApiGeneratorCli : CliktCommand(name = "asyncapi-generator") {
 
     private fun clientRequest(): GeneratorConfigurationRequest.Clients =
         GeneratorConfigurationRequest.Clients(
-            springKafka =
-                GeneratorConfigurationRequest.springKafka(
-                    enabled = true.takeIf { clientsSpringKafka },
-                    packageName = clientsSpringKafkaPackage,
-                    modelPackageName = clientsSpringKafkaModelPackage,
-                    mode = clientsSpringKafkaMode?.configurationValue,
-                    topicPropertyPrefix = clientsSpringKafkaTopicPropertyPrefix,
+            kafka =
+                GeneratorConfigurationRequest.kafka(
+                    enabled = true.takeIf { clientsKafka },
+                    packageName = clientsKafkaPackage,
+                    modelPackageName = clientsKafkaModelPackage,
+                    headers = GeneratorConfigurationRequest.kafkaHeaders(enabled = clientsKafkaHeaders),
+                    springKafka =
+                        GeneratorConfigurationRequest.kafkaSpringKafka(
+                            enabled = true.takeIf { clientsKafkaSpringKafka },
+                            producer =
+                                GeneratorConfigurationRequest.kafkaProducer(
+                                    enabled = clientsKafkaSpringKafkaProducer,
+                                ),
+                            consumer =
+                                GeneratorConfigurationRequest.kafkaConsumer(
+                                    enabled = clientsKafkaSpringKafkaConsumer,
+                                ),
+                        ),
                 ),
             quarkusKafka =
                 GeneratorConfigurationRequest.quarkusKafka(

asyncapi-generator-cli/src/test/kotlin/dev/banking/asyncapi/generator/cli/AsyncApiGeneratorCliTest.kt

@@ -26,9 +26,9 @@ class AsyncApiGeneratorCliTest {
                 "--codegen-output", codegenDir.absolutePath,
                 "--resource-output", resourceDir.absolutePath,
                 "--models-package", "com.example.cli.model",
-                "--clients-spring-kafka-package", "com.example.cli.client",
+                "--clients-kafka-package", "com.example.cli.client",
+                "--clients-kafka-spring-kafka",
                 "--generator", "kotlin",
-                "--clients-spring-kafka-mode", "full",
             )
         )
         val packageDir = codegenDir.resolve("src/main/kotlin/com/example/cli/client")
@@ -46,8 +46,9 @@ class AsyncApiGeneratorCliTest {
                 "--input", inputFile.absolutePath,
                 "--codegen-output", codegenDir.absolutePath,
                 "--resource-output", resourceDir.absolutePath,
-                "--clients-spring-kafka-package", "com.example.cli.client",
-                "--clients-spring-kafka-model-package", "com.example.cli.model",
+                "--clients-kafka-package", "com.example.cli.client",
+                "--clients-kafka-model-package", "com.example.cli.model",
+                "--clients-kafka-spring-kafka",
                 "--generator", "kotlin",
             )
         )
@@ -196,7 +197,7 @@ class AsyncApiGeneratorCliTest {
     }
 
     @Test
-    fun `should fail if spring kafka client is enabled without client package`(@TempDir tempDir: Path) {
+    fun `should fail if kafka spring kafka client is enabled without client package`(@TempDir tempDir: Path) {
         val inputFile = File("src/test/resources/asyncapi_kafka_complex.yaml")
         val codegenDir = tempDir.resolve("codegen").toFile()
         val exception =
@@ -206,14 +207,14 @@ class AsyncApiGeneratorCliTest {
                         "-i", inputFile.absolutePath,
                         "--codegen-output", codegenDir.absolutePath,
                         "--models-package", "com.example.cli.model",
-                        "--clients-spring-kafka",
+                        "--clients-kafka-spring-kafka",
                     )
                 )
             }
 
         assertTrue(
             exception.message.orEmpty().contains(
-                "clients.springKafka.packageName is required when clients.springKafka is configured",
+                "clients.kafka.packageName is required when clients.kafka is configured",
             ),
         )
     }
@@ -263,24 +264,6 @@ class AsyncApiGeneratorCliTest {
         )
     }
 
-    @Test
-    fun `should fail if spring kafka mode is invalid`(@TempDir tempDir: Path) {
-        val inputFile = File("src/test/resources/asyncapi_kafka_complex.yaml")
-        val codegenDir = tempDir.resolve("codegen").toFile()
-        assertFailsWith<BadParameterValue> {
-            cli.parse(
-                arrayOf(
-                    "-i", inputFile.absolutePath,
-                    "--codegen-output", codegenDir.absolutePath,
-                    "--models-package", "com.example.cli.model",
-                    "-g", "kotlin",
-                    "--clients-spring-kafka-package", "com.example.cli.client",
-                    "--clients-spring-kafka-mode", "basic",
-                )
-            )
-        }
-    }
-
     @Test
     fun `should fail if java model type is invalid`(@TempDir tempDir: Path) {
         val inputFile = File("src/test/resources/asyncapi_kafka_complex.yaml")

asyncapi-generator-core/src/main/kotlin/dev/banking/asyncapi/generator/core/generator/analyzer/AnalyzedMessage.kt

@@ -7,4 +7,5 @@ data class AnalyzedMessage(
     val payloadTypeName: String, // The payload type name (e.g. "UserSignedUpPayload")
     val schema: Schema, // The payload schema
     val keySchema: Schema? = null, // Optional Kafka Key schema
+    val headers: AnalyzedMessageHeaders? = null,
 )

asyncapi-generator-core/src/main/kotlin/dev/banking/asyncapi/generator/core/generator/analyzer/AnalyzedMessageHeaders.kt

@@ -0,0 +1,8 @@
+package dev.banking.asyncapi.generator.core.generator.analyzer
+
+import dev.banking.asyncapi.generator.core.model.schemas.SchemaInterface
+
+data class AnalyzedMessageHeaders(
+    val typeName: String,
+    val properties: Map<String, SchemaInterface>,
+)

asyncapi-generator-core/src/main/kotlin/dev/banking/asyncapi/generator/core/generator/analyzer/AnalyzedMultiFormatMessage.kt

@@ -15,4 +15,5 @@ data class AnalyzedMultiFormatMessage(
     val messageName: String,
     val payloadName: String,
     val schema: MultiFormatSchema,
+    val headers: AnalyzedMessageHeaders? = null,
 )

asyncapi-generator-core/src/main/kotlin/dev/banking/asyncapi/generator/core/generator/analyzer/ChannelAnalyzer.kt

@@ -63,7 +63,7 @@ class ChannelAnalyzer {
                 val usage = channelUsage[name]!!
                 val finalProducer = if (!usage.isProducer && !usage.isConsumer) true else usage.isProducer
                 val finalConsumer = if (!usage.isProducer && !usage.isConsumer) true else usage.isConsumer
-                val resolvedMessages = resolveMessages(channel.messages)
+                val resolvedMessages = resolveMessages(channelName = name, messages = channel.messages)
 
                 AnalyzedChannel(
                     channelName = name,
@@ -78,7 +78,10 @@ class ChannelAnalyzer {
         return ChannelAnalysisResult(analyzedChannels)
     }
 
-    private fun resolveMessages(messages: Map<String, MessageInterface>?): ResolvedMessages {
+    private fun resolveMessages(
+        channelName: String,
+        messages: Map<String, MessageInterface>?,
+    ): ResolvedMessages {
         if (messages.isNullOrEmpty()) return ResolvedMessages()
         val analyzedMessages = mutableListOf<AnalyzedMessage>()
         val analyzedMultiFormatMessages = mutableListOf<AnalyzedMultiFormatMessage>()
@@ -95,6 +98,12 @@ class ChannelAnalyzer {
             var typeName: String? = null
             val baseName = MapperUtil.toPascalCase(message.name ?: message.title ?: name)
             val inlinePayloadTypeName = if (baseName.endsWith("Payload")) baseName else "${baseName}Payload"
+            val headers =
+                MessageHeaderAnalyzer.analyze(
+                    channelName = channelName,
+                    messageKey = name,
+                    message = message,
+                )
 
             when (val p = message.payload) {
                 is SchemaInterface.SchemaInline -> {
@@ -123,6 +132,7 @@ class ChannelAnalyzer {
                         messageName = baseName,
                         payloadTypeName = typeName,
                         schema = payloadSchema,
+                        headers = headers,
                     ),
                 )
             } else if (multiFormatSchema != null) {
@@ -131,6 +141,7 @@ class ChannelAnalyzer {
                         messageName = baseName,
                         payloadName = typeName,
                         schema = multiFormatSchema,
+                        headers = headers,
                     ),
                 )
             }