Chapi Kafka streaming API

chapi-streaming turns code-change records into raw and project-resolved Chapi AST events. Version 1 supports Java and keeps project source plus processed eventId values in fault-tolerant Kafka Streams state stores.

Topics

Topic	Direction	Value
chapi.code-change.v1	input	CodeChangeEvent JSON
chapi.ast.raw.v1	output	changed file after pass 1
chapi.ast.resolved.v1	output	every affected project file after pass 2
chapi.diagnostic.v1	output	validation or analysis failure

Input records must use projectId as the Kafka key. This co-partitions all files whose same-package types need to be resolved. eventId values are kept in a changelog-backed state store; replaying an event produces no duplicate AST or diagnostic output. Kafka Streams is configured with exactly_once_v2, so output records, input offsets, and state updates are committed atomically.

An upsert value looks like this:

{
  "eventId": "change-001",
  "projectId": "demo",
  "filePath": "src/main/java/com/example/Foo.java",
  "language": "java",
  "content": "package com.example; public class Foo {}",
  "operation": "UPSERT",
  "occurredAtEpochMs": 1785801600000
}

For DELETE, content may be omitted. Raw delete events have ast omitted, and the resolved topic receives new projections for all remaining files in the project.

Run locally

Kafka 4.x clients and Streams require Java 11 or newer. This module targets Java 11; the broker container bundles its own runtime.

cd chapi-streaming
docker compose up -d
cd ..

for topic in \
  chapi.code-change.v1 \
  chapi.ast.raw.v1 \
  chapi.ast.resolved.v1 \
  chapi.diagnostic.v1
do
  docker compose -f chapi-streaming/compose.yaml exec kafka \
    /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 \
    --create --if-not-exists --topic "$topic"
done

./gradlew :chapi-streaming:run

Produce records with parse.key=true so the text before the first colon is the project key:

docker compose -f chapi-streaming/compose.yaml exec kafka \
  /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic chapi.code-change.v1 \
  --property parse.key=true \
  --property key.separator=:

Then paste one JSON record per line, prefixed by the project ID:

demo:{"eventId":"foo-1","projectId":"demo","filePath":"Foo.java","language":"java","content":"package com.example; public class Foo {}"}
demo:{"eventId":"bar-1","projectId":"demo","filePath":"Bar.java","language":"java","content":"package com.example; public class Bar extends Foo {}"}

The resolved Bar event contains "Extend":"com.example.Foo" even if Bar arrived first: every accepted change reruns pass 2 for the current project snapshot.

Configuration

All settings can be environment variables; defaults are shown below.

Variable	Default
CHAPI_BOOTSTRAP_SERVERS	localhost:9092
CHAPI_APPLICATION_ID	chapi-streaming-v1
CHAPI_STATE_DIR	/tmp/chapi-streaming-state
CHAPI_NUM_STREAM_THREADS	1
CHAPI_CODE_CHANGE_TOPIC	chapi.code-change.v1
CHAPI_AST_RAW_TOPIC	chapi.ast.raw.v1
CHAPI_AST_RESOLVED_TOPIC	chapi.ast.resolved.v1
CHAPI_DIAGNOSTIC_TOPIC	chapi.diagnostic.v1

System properties use the lower-case dotted form, for example -Dchapi.bootstrap.servers=kafka:9092.