feast-kafka-connect-sink

What it does

1. Consumes JSON, schemaful Connect records, or Schema Registry-framed Avro/JSON Schema/Protobuf feature events from one or more Kafka topics
2. Converts each event into a Feast push payload, preserving the event timestamp
3. Pushes features to Feast over HTTP — online, offline, or both
4. Batches efficiently and retries transient failures
5. Integrates with Kafka Connect operational semantics: offsets, dead-letter queue, JMX metrics, structured logging

Architecture

Components

Requirements

• JDK 21
• Apache Kafka Connect 4.x+ (compiled against connect-api 4.3)
• Feast Feature Server reachable over HTTP

Build & test

./gradlew test               # unit tests
./gradlew integrationTest    # integration tests (WireMock; Kafka via Testcontainers if Docker present)
./gradlew connectPluginZip   # build the deployable plugin ZIP

The wrapper is pinned to Gradle 8.14.4 (Kotlin 2.4.0, JVM target 21) for reproducibility.

Packaging & install

./gradlew connectPluginZip produces:

build/distributions/feast-kafka-connect-sink-<version>.zip
└── feast-kafka-connect-sink-<version>/
    ├── lib/feast-kafka-connect-sink-<version>.jar   # the connector
    ├── lib/jackson-*.jar, kotlin-stdlib-*.jar, ...  # bundled runtime deps
    ├── README.md
    └── feast-sink-connector.json

connect-api and slf4j-api are intentionally not bundled — the Connect worker provides them. Unzip the folder into a directory on your worker's plugin.path and restart the worker.

Configuration

Register via the Connect REST API (see feast-sink-connector.json):

curl -s -X POST -H 'Content-Type: application/json' \
  --data @config/feast-sink-connector.json http://localhost:8083/connectors

Connector settings

Recommended standard Connect settings

value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
key.converter=org.apache.kafka.connect.storage.StringConverter
errors.tolerance=all
errors.deadletterqueue.topic.name=dlq
errors.deadletterqueue.context.headers.enable=true

Schema Registry formats

There are two supported deployment models for Schema Registry formats:

# Preferred when the worker already has Confluent converters installed.
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://schema-registry:8081

For JSON Schema, use the matching converter:

value.converter=io.confluent.connect.json.JsonSchemaConverter
value.converter.schema.registry.url=http://schema-registry:8081

For Protobuf, use the matching converter:

value.converter=io.confluent.connect.protobuf.ProtobufConverter
value.converter.schema.registry.url=http://schema-registry:8081

With Confluent converters, Kafka Connect passes schemaful Struct values to the connector; no connector Schema Registry settings are needed.

# In-connector Schema Registry decoding path.
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
schema.registry.url=http://schema-registry:8081
schema.registry.request.timeout.ms=10000
schema.registry.cache.size=1000

With ByteArrayConverter, the connector expects Confluent Schema Registry wire-format values (magic byte + schema id + payload). AVRO payloads are decoded with the writer schema. JSON payloads are parsed as UTF-8 JSON objects. PROTOBUF payloads are decoded with Confluent message indexes and the registered .proto descriptor. Unknown schema IDs and malformed payloads are treated as record-level conversion failures and routed to the DLQ. Registry timeouts, I/O failures, and HTTP 5xx responses are treated as operational failures and pause consumption with RetriableException.

Feast push payload

An event:

{
  "trip_id": "yellow_2024_01_000001",
  "event_timestamp": "2024-01-01T00:13:42Z",
  "pickup_location_id": 161,
  "fare_amount": 18.4
}

becomes a column-oriented push request (one POST carries a whole batch):

POST {feast.url}/push
{
  "push_source_name": "taxi_trips_push_source",
  "df": {
    "trip_id": ["yellow_2024_01_000001", "..."],
    "event_timestamp": ["2024-01-01T00:13:42Z", "..."],
    "pickup_location_id": [161, "..."],
    "fare_amount": [18.4, "..."]
  },
  "to": "online_and_offline"
}

Operational semantics

• Batching & flush: records are buffered per push source and flushed when a source reaches batch.size, when its oldest record has waited linger.ms, or at the preCommit durability barrier. batch.size is a hard per-request maximum (large polls are chunked).
• Ordering: buffers are FIFO and flushed in order; a task never reorders a partition's records.
• Delivery: at-least-once. preCommit only lets offsets advance once buffered records have been pushed (or routed to the DLQ), so no record is lost.
• Backpressure: while Feast is unavailable, the task retries with backoff; if a batch still cannot be delivered, it is returned to the buffer and the task throws RetriableException to pause consumption until Feast recovers — bounding in-memory buffering.
• Idempotency: online pushes are upserts (retry-safe). Offline pushes may duplicate on retry; dedup downstream by entity + event timestamp.

Error handling & DLQ

When dlq.context.headers.enable=true, DLQ records carry: __feast_error_stage (CONVERT/VALIDATE/PUSH), __feast_error_class, __feast_error_message, __feast_push_source.

Metrics

A JMX MBean is registered per task under com.arthenica.kafka.feastconnect.sink:type=task,connector=<name>,task=<id> with attributes: RecordsReceived, RecordsPushed, RecordsMalformed, BatchesPushed, PushRetries, PushFailures, PushLatencyAvgMs, PushLatencyMaxMs, BufferedRecords, LastPushTimestamp.

License

feast-kafka-connect-sink is licensed under the terms of the Apache License version 2.0. See LICENSE for more details.