developing-kafka-java-client

Files (21)

• SKILL.md
• evals.json · evals/
• AppTest.java · references/
• AvroConsumer.java · references/
• AvroProducer.java · references/
• AvroProducerSync.java · references/
• AvroShareConsumer.java · references/
• JsonSchemaConsumer.java · references/
• JsonSchemaProducer.java · references/
• KafkaConfig.java · references/
• build.gradle · references/
• consumer.md · references/
• docker-compose.yml · references/
• multi-event-guide.md · references/
• pom.xml · references/
• readme-template.md · references/
• schema-generation-rules.md · references/
• share-consumer.md · references/
• value.avsc · references/
• value.schema.json · references/
• warpstream-optimization.md · references/

SKILL.md (first lines)

<HARD-GATE>
Do NOT generate any code, scaffold any project, or modify any file until you have
explicitly asked and received answers for questions #1 (existing app or greenfield),
#2 (target environment), and #3 (producer, consumer, or both). If the user's prompt
partially answers some questions, still confirm your understanding before generating.
This applies to EVERY prompt regardless of how specific it appears.
</HARD-GATE>

Begin by announcing: "Using the Confluent Kafka Java Client skill to guide this project."

# Confluent Kafka Java Client Creation

Generate a production-ready Java project for producing to and/or consuming from Kafka using the Apache Kafka Java clients (`KafkaProducer` / `KafkaConsumer`) with Confluent Schema Registry serializers. Supports three target environments: **Confluent Cloud** (managed), **Local Docker**, and **WarpStream** (Kafka-compatible, object-storage-backed); two build tools: **Maven** (default) and **Gradle**; and three schema formats: **Avro** (default), **JSON Schema**, and **Protobuf**. The generated code follows Confluent's best practices.

## Step 1: Gather Requirements

Before generating any code, work through the questions below. **Skip any question the user has already answered explicitly in their prompt** — do not re-ask just for form's sake. For example, "build a producer and consumer on Confluent Cloud with synchronous sends using Gradle" already answers #2, #3, #4, and #5; only #1, #6, #7, #8, #9, and #10 remain.

**Mandatory confirmation gate — do not skip, even if the user answered every question.** Before writing any file, you MUST send one message that:
1. Recaps the answers you extracted as a short bulleted list (e.g., "Target: Confluent Cloud · Components: producer + consumer · Send pattern: async callback · Build: Maven · Schema: Avro · From scratch: yes").
2. Asks any remaining open questions inline.
3. Explicitly asks the user to confirm or correct before you proceed.

Then STOP and wait for the user's reply. Do not generate files in the same turn as the recap, and do not proceed on the assumption that a fully-specified prompt implies consent to generate immediately — the recap catches misinterpretations of the prompt and is required even when questions #1–#10 are all pre-answered. The only way to skip the gate is if the user has already confirmed the recap earlier in this conversation.

Do not assume defaults for #1, #2, or #3 — if any of these are not answered by the prompt, you must ask.

1. **Are you adding Kafka to an existing application, or starting from scratch?**
   - If the user has existing Java code (mentions an existing project, has a `pom.xml`/`build.gradle`, uses Spring Boot/Quarkus/Micronaut, etc.), do **not** scaffold a new project. Instead: (a) identify their existing producer or data-sending code, (b) ask whether they already have schemas registered in Schema Registry, (c) add Schema Registry integration to their existing code following the patterns in the reference files. Generate only the files they are missing (e.g., `KafkaConfig.java`, `src/main/avro/value.avsc`) and modify their existing code inline. Add the Confluent Maven repository and serializer dependencies to their existing build file.
   - If the user already produces to Kafka without Schema Registry (e.g., `StringSerializer` with hand-rolled JSON), help them migrate: (1) generate an Avro/JSON schema from their existing message structure, (2) register it, and (3) replace their raw `String`/`byte[]` serializer with a Confluent Schema Registry serializer. Do not discard their existing code.
   - If starting from scratch, proceed with the full scaffold below.
2. **Target environment?** — Confluent Cloud, local Kafka (Docker), or WarpStream. **Always prompt for this, even if the user didn't mention it.** If they mention "local", "docker", "self-hosted", or just want to try Kafka without a cloud account, choose **local Docker**. If they mention "Confluent Cloud", "CC", or have existing cloud credentials, choose **Confluent Cloud**. If they mention "WarpStream", choose **WarpStream**. Default to Confluent Cloud if they confirm they don't have a preference, but always ask first.
   - **If WarpStream:** Read `references/warpstream-optimization.md` and apply the Java client overrides from that reference. Key changes: disable idempotence, dramatically increase batch sizes and in-flight requests, set large fetch sizes, add `ws_az=<az>` to `client.id` for zone-aware routing. Prefer null message keys for sticky partitioning unless entity-based ordering is required.
3. **Producer, consumer, or both?**
4. **Send pattern?** (Only if producer is requested.) The Java `KafkaProducer.send()` always returns a `Future<RecordMetadata>` — the producer is inherently asynchronous. The choice is how you handle the result. Help the user choose:
   - **Async with callback** (recommended default): `producer.send(record, callback)`. Non-blocking; the `Callback` handles per-record delivery reports and errors. Best throughput for most applications. Use `references/AvroProducer.java`.
   - **Synchronous**: `producer.send(record).get()`. Blocks until the broker acks each record. Lowest throughput but strongest per-message confirmation and simplest error handling. Best for batch/ETL scripts and steps that must confirm one write before the next. Use `references/AvroProducerSync.java`.
   - (Fire-and-forget — `producer.send(record)` with no callback and no `.get()` — is mentioned for completeness but **not** recommended: delivery failures are silently dropped.)
   If the user mentions batch, ETL, or "confirm each write", default to **synchronous**. Otherwise default to **async callback**.
5. **Build tool?** — **Maven** (default, `pom.xml`) or **Gradle** (`build.gradle`). If the user mentions Gradle, an existing `build.gradle`, or Kotlin DSL, use Gradle (`references/build.gradle`). Otherwise default to Maven (`references/pom.xml`).
6. **Do you have an existing schema you'd like to use?** If yes, ask the user to paste it or provide the file path, then use it instead of generating one. If no, proceed to ask about their data fields.
7. **What kind of data are you producing?** (Only if the user doesn't have an existing schema. Get field names and types so you can generate a matching schema and sample data.)
8. **Topic name?** (Default: `demo-topic`)
9. **Consumer group ID?** (Only if consumer; default: `java-consumer-group`)
10. **Consumer API or Share Consumer API?** (Only if consumer is requested.) Help the user choose, explaining the key distinction: with the regular **Consumer API** (`KafkaConsumer`, consumer groups) each partition is assigned to exactly one consumer, so parallelism is capped at the partition count and per-key ordering is preserved. With the **Share Consumer API** (`KafkaShareConsumer`, share groups — "Queues for Kafka") **multiple share consumers can consume from the same partition**, giving **queue-like messaging semantics** where multiple consumers cooperatively process messages concurrently and you can scale consumers beyond the partition count (the trade-off: no per-key ordering, and each record must be acknowledged). **Default to the regular Consumer API.** Choose the Share Consumer API when the user asks for queue/work-queue semantics, competing consumers, cooperative concurrent processing, or scaling consumers independently of partitions. If the user picks the Share Consumer API, read `references/share-consumer.md` before generating — it requires Kafka 4.x clients (the reference build files already pin `kafka.version=4.2.1` / `confluent.version=8.2.1`) and cluster-side feature enablement.

Don't ask whether to use Schema Registry — always include it. **Default to Avro** (idiomatic for the JVM: generated `SpecificRecord` classes give compile-time type safety and a compact wire format). Offer JSON Schema or Protobuf if the user prefers — ask only if they hint at a preference; otherwise use Avro. If the target is WarpStream and the user is using WarpStream's built-in schema registry, note that it only supports Avro and Protobuf (`GET /schemas/types` returns `["AVRO","PROTOBUF"]`) — Avro is already the default, so no change is needed unless they want Protobuf.

### Common Agent Mistakes

| Thought | Reality |
|---------|---------|
| "The user said Spring Boot, so I'll scaffold a fresh Maven project" | If they have an existing app, modify it in place — add deps to their build file and integrate the serializer. Do not scaffold a competing project. |
| "I'll default to JSON Schema like the Python skill" | This skill defaults to **Avro** on the JVM. Generated `SpecificRecord` classes are idiomatic and type-safe. Only use JSON Schema/Protobuf if the user prefers. |
| "I'll leave `auto.register.schemas` at its default" | The serializer default is `true`. Always set it to **`false`** and register explicitly via `CachedSchemaRegistryClient.register(...)`. Silent auto-registration is a production hazard. |
| "I'll create a `KafkaProducer` inside the send method" | One `KafkaProducer` instance, created in `main()`, passed as a parameter. Producers are thread-safe and expensive to create. |
| "I'll write the schema ID into a record header like the Python sync producer" | Java Confluent serializers embed the schema ID in the **wire format** (magic byte + 4-byte ID) automatically. Do not hand-roll headers. |
| "Avro just needs the `.avsc` on the classpath" | Avro `SpecificRecord` classes are **generated at build time** by `avro-maven-plugin` (Maven) or the `com.github.davidmc24.gradle.plugin.avro` plugin (Gradle) from `src/main/avro/*.avsc`. The producer imports the generated class (e.g., `com.example.kafka.Transaction`). Run `mvn generate-sources` / `./gradlew generateAvroJava` before referencing it. |
| "I'll catch and swallow `WakeupException` errors generally" | `WakeupException` is the **expected** signal from `consumer.wakeup()` in a shutdown hook. Catch it narrowly to break the poll loop, then `close()` in `finally`. |
| "Consumer means `KafkaConsumer` — I'll skip asking about the Share Consumer API" | When a consumer is requested, always ask Consumer API vs Share Consumer API (question #10). Default to the regular Consumer API, but offer the Share Consumer API (`KafkaShareConsumer`) for queue-like/cooperative consumption. |
| "I'll forget the Confluent Maven repo — the serializers are on Maven Central" | `io.confluent:kafka-avro-serializer` and friends resolve from `https://packages.confluent.io/maven/`. The build file MUST declare that repository. |
| "`schema.registry.url` goes on the `KafkaProducer` config only" | The serializer reads `schema.registry.url` (and `basic.auth.*`) from the **same** producer/consumer `Properties`. Put SR config in the client properties so the serializer picks it up. |

## Step 1b: Confirm Understanding

After gathering all answers, present a confirmation summary before generating any code:

```
Before I generate the project, let me confirm:
- Project type: [Greenfield scaffold / Migration of existing code]
- Environment: [Confluent Cloud (SASL_SSL) / Local Docker (PLAINTEXT) / WarpStream]
- Schema format: [Avro (default) / JSON Schema / Protobuf]
- Build tool: [Maven / Gradle]
- Components: [Producer only / Consumer only / Both]
- Send pattern: [Async callback / Synchronous] (if producer)
- Consumer API: [Consumer API / Share Consumer API] (if consumer)
- Schema: [brief description of user's data fields]
- Topic: [topic name]
- Consumer group: [group ID] (if consumer)

Does this look right?
```

Wait for user confirmation before proceeding to Step 2. If the user corrects anything, update your understanding and re-confirm.

## Step 2: Generate the Project

### Decision Flowchart

```dot
digraph decisions {
  "Q1: Existing app?" -> "Migration path:\nmodify existing code +\nadd deps to build file" [label="yes"];
  "Q1: Existing app?" -> "Q2: Environment?" [label="no / greenfield"];
  "Q2: Environment?" -> "Cloud config\n(SASL_SSL)" [label="Confluent Cloud"];
  "Q2: Environment?" -> "Local Docker config\n(PLAINTEXT) + docker-compose.yml" [label="local / docker / OSS"];
  "Q2: Environment?" -> "WarpStream config\n(apply overrides from\nreferences/warpstream-optimization.md)" [label="WarpStream"];
  "Cloud config\n(SASL_SSL)" -> "Q4: Send pattern?";
  "Local Docker config\n(PLAINTEXT) + docker-compose.yml" -> "Q4: Send pattern?";
  "WarpStream config\n(apply overrides from\nreferences/warpstream-optimization.md)" -> "Q4: Send pattern?";
  "Q4: Send pattern?" -> "AvroProducer.java\n(send + Callback)" [label="async callback"];
  "Q4: Send pattern?" -> "AvroProducerSync.java\n(send().get())" [label="synchronous"];
}
```

Create this file structure in the user's chosen directory (Maven layout shown; for Gradle the source tree is identical, only the build file differs):

```
<project-dir>/
├── pom.xml                       # Maven (or build.gradle for Gradle)
├── docker-compose.yml            # (local Docker path only)
├── .properties.example           # template for credentials
├── README.md
└── src/
    ├── main/
    │   ├── java/com/example/kafka/
    │   │   ├── KafkaConfig.java   # shared config loading + connectivity verification
    │   │   ├── AvroProducer.java  # (if producer requested; or AvroProducerSync.java)
    │   │   └── AvroConsumer.java  # (if consumer requested; or AvroShareConsumer.java for the Share Consumer API)
    │   └── avro/

From confluentinc/agent-skills, Apache-2.0.

More from confluentinc/agent-skills