> ## Documentation Index > Fetch the complete documentation index at: https://docs.automq.com/llms.txt > Use this file to discover all available pages before exploring further. # Weaviate Sink Connector > Configure and operate the Weaviate Sink Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting. ## 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](../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. ```properties theme={null} connector.class=io.weaviate.connector.WeaviateSinkConnector topics= weaviate.connection.url=http://: weaviate.grpc.url=: collection.mapping= value.converter=org.apache.kafka.connect.json.JsonConverter value.converter.schemas.enable=false ``` 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. ```json theme={null} {"title":"Kafka Connect","content":"Stream object data into Weaviate."} ``` `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. ```properties theme={null} tasks.max=2 ``` **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](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json), 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.