Kafka Streams Programming

Step 6: Run the App Before Handing Off

You must actually start the app against a real broker and observe it reach RUNNING before declaring the task done. Generated code that compiles and passes TopologyTestDriver tests can still fail at startup — version-mismatch NoClassDefFoundErrors, silent logger fallbacks, missing runtime deps, and import-path errors all slip past compile + test and only surface against a real broker / Schema Registry. A green build is not a working app.

Branch on the target environment chosen in Step 1:

Local (Apache Kafka or Confluent Platform via the generated docker-compose.yml):

1. docker compose up -d and wait for Kafka + SR to be healthy (docker compose ps, or curl SR /subjects)
2. ./create-topics.sh
3. Start the app in the background (./gradlew run or mvn exec:java) so you can read its logs while it runs
4. Tail the log and confirm State transition from REBALANCING to RUNNING within ~30s. If you don't see it, read the actual stack trace, diagnose via references/debugging.md § Startup Failures, fix, restart, re-verify
5. If the user wanted sample data: produce a few records and confirm output appears on the destination topic
6. Stop the app and docker compose down (or leave running if the user wants to keep iterating — ask)

Confluent Cloud: You usually cannot run end-to-end yourself because the cluster + SR API keys are the user's. Do the most you can without them, then hand off the rest:

1. If .env has real CC creds (the user has set up a real .env): run the app locally pointed at CC (./gradlew run auto-loads .env) and follow steps 3–5 above. Don't skip just because it's CC — if you have creds, run it.
2. If creds are placeholders or not provided: do not fabricate a successful run. Instead:
   • Run ./gradlew build (compile + unit tests) and report the result
   • List the exact commands the user must run to verify (./create-topics.sh --cloud, ./gradlew run, the consume command from references/verification.md § Confluent Cloud) and what success looks like (State transition from REBALANCING to RUNNING, records on the output topic)
   • Tell the user explicitly: "I couldn't run this against your CC cluster because I don't have your API keys — please run the steps above and paste any errors back."

WarpStream: You usually cannot run end-to-end yourself because the WarpStream cluster and credentials are the user's. Follow the same approach as Confluent Cloud:

1. If .env has real WarpStream creds: run the app locally (./gradlew run auto-loads .env) and follow the Local steps 3–5 above. Note that State transition from REBALANCING to RUNNING may take longer due to WarpStream's higher metadata latency.
2. If creds are placeholders or not provided: do not fabricate a successful run. Instead:
   • Run ./gradlew build (compile + unit tests) and report the result
   • List the exact commands the user must run to verify (./create-topics.sh, ./gradlew run) and what success looks like (State transition from REBALANCING to RUNNING, records on the output topic)
   • Remind the user to set client.id with ws_az=<az> in their .env for zone-aware routing
   • Tell the user explicitly: "I couldn't run this against your WarpStream cluster because I don't have your credentials — please run the steps above and paste any errors back."

In the handoff, state plainly which of the above you did. If you ran it and saw RUNNING, say so. If you only compiled, say only that. Don't imply a runtime verification you didn't perform.

For CC consume commands, schema-aware producers, and reset procedures, read references/verification.md.

Debug Mode

Step 1: Classify the Problem

Step 2: Gather Context

Confirm target environment first (see preamble) — most debug paths branch on it. Then ask for: error message, config, KS/Java versions, new app or regression?

Step 3: Diagnose and Fix

Read the relevant section in references/debugging.md for the identified category. Provide fix with explanation.

Invariant Checklist

Non-negotiable defaults. Apply all. Read reference files only if you need implementation details.

1. Schematized data: SR serdes (SpecificAvroSerde, KafkaProtobufSerde, KafkaJsonSchemaSerde). Set schema.registry.url, default.key.serde, default.value.serde. JSON Schema: set json.value.type. Protobuf: set specific.protobuf.value.type (references/config-baseline.md)
2. Versions: KS 4.x / CP 8.x, Java 17+
3. KIP-1071: group.protocol=streams (default). Remove if UnsupportedVersionException. Unsupported: static membership, regex topics, standby replicas, warm-up replicas (references/topology-patterns.md § Assignment Strategy)
4. Four-tier error handling: DeserializationExceptionHandler, ProcessingExceptionHandler (KIP-1034), ProductionExceptionHandler, StreamsUncaughtExceptionHandler. Use MaxFailures pattern for uncaught (references/production-hardening.md § Error Handling)
5. Explicit naming: ensure.explicit.internal.resource.naming=true
6. Graceful shutdown: Hook with streams.close(30s) on SIGTERM/SIGINT
7. Monitoring: metrics.recording.level=INFO (references/config-baseline.md)
8. Log verbosity: Generate simplelogger.properties (references/build-templates.md)
9. Defensive topology: Guard lambdas in prod (null checks, try/catch). Dev: simpler is fine.
10. Schema parity: Avro/Protobuf/JSON Schema all supported (references/build-templates.md, references/schema-patterns.md)
11. Test caching: TopologyTestDriver tests need statestore.cache.max.bytes=0 to avoid non-deterministic assertions.
12. Avro logical type Java mappings: Avro 1.12+ generates java.time.Instant for timestamp-millis/timestamp-micros, LocalDate for date, BigDecimal for decimal, etc. Never use raw long/int literals with generated setter methods — use Instant.EPOCH, Instant.now(), Instant.ofEpochMilli(...). Use Instant.isAfter()/isBefore() instead of Math.max()/Math.min() for timestamp comparisons. Applies to topology code, aggregation initializers, producers, AND test helpers (references/schema-patterns.md § Java type mapping).

Bundled Scripts

scripts/: create-topics.sh (pre-create topics, --cloud), teardown.sh (delete topics/state, --cloud), produce-test-data.sh (generate if requested).

Reference Files (read on-demand only)

references/topology-patterns.md — design, joins, windows, aggregations | references/architecture.md — internals, sizing | references/debugging.md — troubleshooting | references/config-baseline.md — config | references/build-templates.md — project structure | references/schema-patterns.md — Avro/Protobuf/JSON | references/production-hardening.md — prod setup | references/cli-commands.md — CLI | references/docker-compose.md — local dev | references/verification.md — checklists | references/warpstream-optimization.md — WarpStream client config overrides