Skip to main content

​
Overview

The Weaviate Sink Connector writes object data from a Kafka Topic to a Weaviate Collection, connecting a message pipeline that continuously produces business data with a vector search system. The value of each non-null record is mapped to an object’s properties. A Topic can map to a Collection with the same name or all records can be written to a specified Collection, supporting content indexing, knowledge base updates, and business object search. Object identifiers can be generated by the client or derived from the Kafka message key or a record field. Vectors can be generated by the target Collection’s vectorizer or extracted from a single vector field in the message; the Connector does not generate Embeddings. Kafka Connect Converter decodes the message serialization format, and the Connector receives the decoded object data.

​
Prerequisites

  • Use Weaviate 1.28.x or later and create the target Collection in advance; its property types and vector configuration must match the input objects. When vectors must be generated by the server, configure the corresponding vectorizer and its module dependencies.
  • The Weaviate HTTP interface and the configured gRPC interface must be available to the Worker; an explicit gRPC address typically uses port 50051, and its TLS settings must match the target service. When authentication is enabled, prepare an API Key or OIDC client credentials accepted by the target service, together with write authorization.
  • When using the original Connector 0.1.2 distribution ZIP, provide the plugin with complete standalone HTTP/2 runtime dependencies; prefer a vendor distribution with complete dependencies and security maintenance. If the original ZIP must be retained, install the complete dependencies in isolation in the same plugin location and keep the installation consistent across all Workers running this Connector. Production environments must still review the artifacts and deployment under their own security policies. This limitation applies to that distribution and is not a promise that every default deployment can use it directly.

​
License

Uses Apache License 2.0.

​
Quick Start

Prepare a Connect Cluster, Kafka, the input Topic, and a created Weaviate Collection, and confirm network connectivity and access permissions; see Manage Connectors for creation and management operations. The following configuration is for an environment without authentication and without TLS on either HTTP or gRPC, where input message values are JSON objects without a Schema envelope.
Replace the Topic, Weaviate address, ports, and Collection name, then apply the configuration. This example writes all input objects to one specified Collection, does not derive IDs from message keys, and does not extract vectors from properties. The client generates a UUID for each object. The Collection should have a server-side vectorizer configured, or allow objects without a supplied vector to be written. If the target service requires authentication or TLS, adjust the connection and authentication settings accordingly; do not reuse the unauthenticated connection settings above unchanged. The following JSON is an example of a message value sent to the Kafka Topic; it is business data, not Connector configuration. The target Collection must accept the title and content properties.
value.converter.schemas.enable=false makes JsonConverter convert an ordinary JSON object to a schemaless Map; do not send a schema / payload envelope or replace the object with a top-level string or array. Query the written object’s properties in Weaviate to confirm that the data reached the target Collection. The default ID strategy generates a new UUID when the same record is replayed, which may create duplicate objects.

​
Configuration

​
Connection

​
weaviate.connection.url

The address of the Weaviate HTTP interface.
  • Type: string
  • Default: http://localhost:8080
  • Importance: High
  • Valid values / notes: Use http://host:port or https://host:port including the scheme; do not provide only a hostname. HTTP operations such as deletion and reading objects after a timeout still require this address even when batch writes use gRPC.

​
weaviate.grpc.url

Explicitly specifies the address of the Weaviate gRPC interface.
  • Type: string
  • Default: localhost:50051
  • Importance: High
  • Valid values / notes: Use the host:port format without an HTTP scheme prefix. An empty string means that the client’s gRPC address is not explicitly overridden; it does not disable all gRPC operations.

​
weaviate.grpc.secured

Controls whether TLS is enabled for an explicit gRPC connection.
  • Type: boolean
  • Default: false
  • Importance: High
  • Valid values / notes: true or false; applies only when weaviate.grpc.url is non-empty. https in the HTTP address does not automatically set this option to true.

​
Authentication and Request Headers

​
weaviate.auth.scheme

Selects the authentication method used when connecting to Weaviate.
  • Type: string
  • Default: NONE
  • Importance: High
  • Valid values / notes: Use uppercase NONE, API_KEY, or OIDC_CLIENT_CREDENTIALS. They represent no authentication, API Key authentication, and OIDC client credentials authentication, respectively; do not use lowercase or mixed case.

​
weaviate.api.key

The key used for API Key authentication.
  • Type: string
  • Default: null
  • Importance: High
  • Valid values / notes: Provide a valid key when API_KEY is selected. This setting is not a password type, so do not assume that an administration interface or logs will redact it automatically; use the deployment environment’s secure credential management and do not put a real key in shared configuration or logs.

​
weaviate.oidc.client.secret

The Secret used for OIDC client credentials authentication.
  • Type: string
  • Default: null
  • Importance: High
  • Valid values / notes: Provide a Secret accepted by the target service when OIDC_CLIENT_CREDENTIALS is selected. This setting is not a password type, so do not assume automatic redaction; protect its read and export permissions.

​
weaviate.oidc.scopes

The Scope requested for OIDC client credentials authentication.
  • Type: list
  • Default: [openid]
  • Importance: High
  • Valid values / notes: Comma-separated Scope values, used only for OIDC_CLIENT_CREDENTIALS; fill them according to the identity provider’s authorization requirements.

​
weaviate.headers

Additional request headers added when building the Weaviate client, which can provide authentication information required by the target vectorizer module.
  • Type: list
  • Default: Empty list []
  • Importance: Medium
  • Valid values / notes: Comma-separated name=value entries. Both names and values should be non-empty. Do not include extra = characters in a value, or it may be truncated; a later entry overrides an earlier entry with the same name. Do not expose credential-bearing request headers in shared documents or logs.

​
Collection and Writes

​
collection.mapping

Specifies the target Collection name or a Topic-based name template.
  • Type: string
  • Default: ${topic}
  • Importance: High
  • Valid values / notes: Each literal ${topic} is replaced with the current record’s Topic name; a fixed string writes multiple Topics to the same Collection. This is not a comma-separated Topic-to-Collection mapping table. Names are not automatically sanitized or case-adjusted, must meet Weaviate naming requirements, and must refer to Collections created in advance.

​
consistency.level

The replica consistency level passed to object batch writes and delete operations.
  • Type: string
  • Default: QUORUM
  • Importance: Low
  • Valid values / notes: Use uppercase ALL, ONE, or QUORUM; the required replicas and their availability depend on the target Weaviate deployment. This is not a transaction setting between Kafka offsets and target writes.

​
Object Identity

​
document.id.strategy

Selects the strategy class used to derive an object ID from a record.
  • Type: class
  • Default: io.weaviate.connector.idstrategy.NoIdStrategy
  • Importance: Medium
  • Valid values / notes: Built-in classes are io.weaviate.connector.idstrategy.NoIdStrategy, io.weaviate.connector.idstrategy.KafkaIdStrategy, and io.weaviate.connector.idstrategy.FieldIdStrategy. NoIdStrategy supplies no ID, so the client generates a random UUID. KafkaIdStrategy converts the UTF-8 bytes of a string message key into a name-based UUID; even if the key itself looks like a UUID, it is derived again. Non-string keys are converted to strings, and the result must not be assumed to be a valid UUID. FieldIdStrategy converts a specified top-level field to a string and derives a name-based UUID, while removing that field from the object properties. Stable IDs should be unique within the target Collection; the same key from multiple Topics written to one Collection produces the same ID. Custom classes must implement IDStrategy and have an accessible no-argument constructor. When deletion is enabled, the built-in KafkaIdStrategy must be used; subclasses are not accepted as replacements.

​
document.id.field.name

Specifies the object ID field read by FieldIdStrategy.
  • Type: string
  • Default: id
  • Importance: Medium
  • Valid values / notes: Used only by FieldIdStrategy. It reads a top-level property of the converted object and does not support nested paths. The field should always be present and have a non-empty, stable, unique scalar value; a missing or empty value is derived as the string null, causing the same ID and conflicts. After extraction, the field is no longer written as a regular property.

​
Vectors

​
vector.strategy

Selects the strategy class used to extract a vector from a record.
  • Type: class
  • Default: io.weaviate.connector.vectorstrategy.NoVectorStrategy
  • Importance: Medium
  • Valid values / notes: Built-in classes are io.weaviate.connector.vectorstrategy.NoVectorStrategy and io.weaviate.connector.vectorstrategy.FieldVectorStrategy. The former does not submit an explicit vector; whether a vector is generated depends on the Collection. The latter extracts one top-level field as a single vector. Custom classes must implement VectorStrategy and have an accessible no-argument constructor.

​
vector.field.name

Specifies the vector field read by FieldVectorStrategy.
  • Type: string
  • Default: vector
  • Importance: Medium
  • Valid values / notes: Used only by FieldVectorStrategy. It supports Float[] or an iterable collection whose elements are Float / Double; Double is converted to Float. Integer elements and scalar values cannot be used as vector input. A missing or empty value supplies no vector; after successful extraction, the field is removed from the properties. The dimension must match the target Collection. ID extraction runs before vector extraction, so do not use the same field for document.id.field.name.

​
Batch Processing

​
batch.size

Controls the number of objects in each client batch for a Task.
  • Type: int
  • Default: 100
  • Importance: Low
  • Valid values / notes: No configuration-level numeric range validation is defined; set it according to individual object size and target processing capacity. Remaining objects are also submitted when each received record set ends, so an actual batch may be smaller than this value; the Connector does not wait until a batch is full before sending.

​
pool.size

Controls the size of the client batch-processing thread pool for a Task.
  • Type: int
  • Default: 1
  • Importance: Low
  • Valid values / notes: No configuration-level numeric range validation is defined, but the thread pool still requires a valid size. This setting is independent of tasks.max; assess target load before increasing concurrency, and do not infer object write ordering or linear throughput improvement from it.

​
await.termination.ms

Sets the wait duration, in milliseconds, when the batch-processing executor shuts down.
  • Type: int
  • Default: 10000
  • Importance: Low
  • Valid values / notes: No configuration-level numeric range validation is defined. This is a shutdown wait setting, not a general timeout for an individual request, put, or flush, and not a guarantee of write success.

​
Client Retry Parameters

​
max.connection.retries

The client batch-write setting for the number of connection-error retries.
  • Type: int
  • Default: 3
  • Importance: Low
  • Valid values / notes: No configuration-level numeric range validation is defined. It applies only to connection-error paths recognized by the client; it does not mean that all HTTP, gRPC, or individual-object errors are retried, does not cover deletes, and is not equivalent to Kafka Connect error retry settings.

​
max.timeout.retries

The client batch-write setting for the number of timeout retries.
  • Type: int
  • Default: 3
  • Importance: Low
  • Valid values / notes: No configuration-level numeric range validation is defined. It applies only to timeout paths recognized by the client; do not assume that every timeout triggers a retry or that the write will eventually succeed. It does not cover deletes.

​
retry.interval

The base interval for client batch retries, in milliseconds.
  • Type: int
  • Default: 2000
  • Importance: Low
  • Valid values / notes: No configuration-level numeric range validation is defined. The corresponding client retry path multiplies the retry count by this base interval to calculate the wait time; this is not exponential backoff. This setting does not expand the set of retryable errors.

​
Deletion

​
delete.enabled

Controls whether a null message value is treated as a request to delete the target object.
  • Type: boolean
  • Default: false
  • Importance: Low
  • Valid values / notes: false skips null-valued records; true requires document.id.strategy=io.weaviate.connector.idstrategy.KafkaIdStrategy. Deletion derives the ID from the message key using the same strategy, which must match the identifier used when the original object was written. The string "null" and an empty JSON object are not Tombstones. This setting does not parse CDC delete events and does not guarantee completion order between asynchronous writes and a subsequent delete.

​
Kafka Connect Runtime and Input

​
connector.class

Specifies the Connector class to run.
  • Type: string
  • Default: No fixed default, required
  • Importance: High
  • Valid values / notes: Use io.weaviate.connector.WeaviateSinkConnector.

​
tasks.max

Requests the maximum number of Tasks to run.
  • Type: int
  • Default: 1
  • Importance: High
  • Valid values / notes: At least 1. Effective consumption parallelism is constrained by the number of input partitions and assignment results. Each Task has its own client and batch-processing resources; Tasks are not automatically divided by Collection.

​
topics

Specifies the list of Topics to consume.
  • Type: list
  • Default: Empty list []
  • Importance: High
  • Valid values / notes: Comma-separated Topic names; exactly one of topics and topics.regex must be configured with a non-empty value. If the deployment has a DLQ configured, do not consume that DLQ Topic.

​
topics.regex

Uses a regular expression to select input Topics.
  • Type: string
  • Default: Empty string ""
  • Importance: High
  • Valid values / notes: Use Java regular-expression syntax. It is mutually exclusive with topics and must not match a configured DLQ Topic. When a new matching Topic is added, its object structure must still match the target Collection. When using ${topic} mapping, prepare each target Collection in advance.

​
key.converter

Decodes the Kafka message key into a Connect value.
  • Type: class
  • Default: null
  • Importance: Low
  • Valid values / notes: When unset, the Worker Converter configuration is used; when set explicitly, it must be an instantiable Converter class. When using KafkaIdStrategy, the decoded key type determines ID handling. Changing the Converter may change object identity; do not infer that IDs are the same merely because the original Kafka bytes are the same.

​
value.converter

Decodes the Kafka message value into a Connect value.
  • Type: class
  • Default: null
  • Importance: Low
  • Valid values / notes: When unset, the Worker Converter configuration is used; when set explicitly, it must be an instantiable Converter class. Non-null values must convert to a top-level Map or Struct. Avro, JSON, Protobuf, and other formats require the corresponding Converter; the Connector does not parse them directly. The Quick Start additionally disables the JsonConverter Schema envelope; that sub-setting is not the default for this configuration.

​
Best Practices

​
Increase Consumer Parallelism by Input Partitions When Scaling

Applicable business scenario: When the Connector is already running normally but multiple input partitions continue to accumulate lag and Weaviate still has processing capacity, let multiple Tasks share consumption. First confirm that the input has at least two partitions and observe target-service load, then increase parallelism gradually. Configuration example: Add the following setting to the Quick Start configuration while keeping the existing input Topic, Collection, and conversion method.
Key notes: This setting increases the Task limit from the default of one to two. Partitions are assigned by the consumer group, not split by Collection. If there are too few partitions, additional Tasks cannot provide more consumption parallelism; each Task adds client and batch-processing resources. After changing it, check partition assignment, lag changes, and target resource usage together. Do not increase pool.size at the same time, so the source of the change can be identified. There is no global object write-order guarantee across Tasks.

​
Monitoring

​
What to Monitor

Monitor Kafka Connect cluster health, Connector / Task status, input throughput, consumer lag and end-to-end latency, Offset commit progress and duration, error and retry signals, and Worker JVM heap, GC, and thread state. Monitor DLQ activity only when the corresponding error handling is enabled in the deployment. Do not treat a Task being RUNNING or an Offset being committed as proof that all target objects have been written successfully; also query the target objects to check arrival.

​
Import the Grafana Dashboard

Download the AutoMQ Connect Cluster Dashboard, confirm that monitoring data is collected in a Prometheus-compatible data source and that collection labels match the cluster, Connector, Task, and instance filter labels used by the dashboard, then import the JSON into Grafana and select the corresponding data source.

​
Limitations

  • The Connector does not create Collections or migrate Collection Schemas; the target Schema and automatic Schema policy determine whether new properties are accepted.
  • Non-null records must be objects. Top-level scalars and arrays are not supported. The Connector does not automatically unpack CDC envelopes and does not provide field filtering, renaming, or field-based Collection routing.
  • Built-in vector extraction handles only one vector. It does not support named vectors or multiple vectors, and it does not generate Embeddings inside the Connector.
  • delete.enabled=true can be used only with the built-in KafkaIdStrategy; it cannot be used with FieldIdStrategy or NoIdStrategy.
  • Batch write or delete failures returned as client errors do not automatically make the Task fail. The Connector does not actively send these failed records to a DLQ, so Task status or framework error-handling settings cannot be used to determine that these writes succeeded.

​
FAQ

​
The Task reports that the authentication scheme or consistency level is invalid at startup

Check the case of weaviate.auth.scheme and consistency.level. Use uppercase NONE, API_KEY, OIDC_CLIENT_CREDENTIALS, and ALL, ONE, or QUORUM. The authentication method must also match the target service, with the corresponding credentials supplied. If authentication succeeds but the connection still fails, check the HTTP scheme, gRPC address, and gRPC TLS setting separately.

​
A JSON message cannot be converted or written as an object

First check that value.converter matches the message format. When using the Quick Start JsonConverter, the message value should be an ordinary JSON object and value.converter.schemas.enable=false should be set; top-level strings and arrays cannot be mapped to objects. Then check property names, types, and the Collection’s vector configuration. Before changing the Converter, confirm the existing data format; do not use ignored errors as a substitute for correcting the format.

​
An object was written to an unexpected Collection

Check collection.mapping. The default ${topic} routes by the original Topic name, does not change case, and does not parse a topic:collection mapping form. To write to one Collection, enter a fixed Collection name. To write per Topic, create every Collection produced by the template replacement in advance.

​
Replaying records creates duplicate objects, or different records map to the same object ID

The default NoIdStrategy generates a new UUID each time a record is processed, so replaying records may add objects. When stable identity is required, first check whether the message already contains a stable, unique key or ID field, then select the appropriate ID strategy and plan cleanup or migration of existing objects. When using the key strategy, confirm the Converter output type and avoid sending the same key from multiple Topics into one Collection. When using the field strategy, check whether the field is missing or empty. Switching strategies does not automatically migrate existing object IDs, and a stable ID does not imply cross-system transactions or write ordering.

​
The Task is running normally, but the expected object cannot be found in the target

First confirm the input Topic, consumption progress, and actual target Collection. Then check Weaviate request errors, property types, vector requirements, and authentication authorization. A RUNNING state or advancing Offset does not prove that the target object was written; use the Task and Worker error information together with a query against the target to confirm the actual write result.

​
The target object still exists after sending a null value

Confirm that a Kafka null message value was sent rather than the JSON string "null", and check that delete.enabled is enabled and the ID strategy is the built-in KafkaIdStrategy. The message key after conversion must match the key used when the original object was written; an object written with a random ID cannot be located using that key. Also check the result of the target delete request and whether a write for the same ID is still in progress or followed by another write. Do not judge deletion completion only from Task status.