> ## 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. # Milvus Sink Connector > Configure and operate the Milvus Sink Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting. ## Overview The Milvus Sink Connector consumes records from Kafka topics and uses the Milvus REST API to batch-upsert them into a collection in a specified database. Fields in Kafka record values are matched by name against the collection schema. Matching fields are converted when the connector supports the target field type. Kafka record keys, topics, partitions, offsets, and headers are not used to populate target fields or route records to collections. Each connector instance writes to one fixed `database.name` and `collection.name`. It is suitable for writing entity data that has already been structured according to a Milvus schema from Kafka to a vector search collection, for example, writing an application primary key, text attributes, and a `FloatVector` field together. The target collection must be created and loaded in advance, auto-ID must be disabled, and the message value must explicitly provide the primary key. ## Prerequisites * Create and load the collection in the target Milvus or Zilliz Cloud deployment in advance, disable auto-ID, and prepare a token that can access the database and collection and perform schema queries and entity upserts. Fields to be written from each message must have the same names as the collection fields and compatible types. The Quick Start requires at least an explicit `Int64` primary key field named `id`, a `FloatVector` field named `vector`, and optional scalar fields. ## License Licensed under the Apache License 2.0. ## Quick Start Prepare a Connect cluster, Kafka, and a Milvus collection that has already been created and loaded, and verify network connectivity and access permissions. For preparation and management instructions, see [Manage Connectors](../manage-connectors). ```properties theme={null} connector.class=com.milvus.io.kafka.MilvusSinkConnector topics= public.endpoint= token= database.name= collection.name= value.converter=org.apache.kafka.connect.json.JsonConverter value.converter.schemas.enable=false ``` Replace ``, ``, ``, ``, and `` with values from your environment. For ``, use a base address that includes `http://` or `https://` and the required port. The token format depends on the target deployment and can be an API token or the `username:password` form required by that deployment. When writing schemaless JSON objects to the topic, each message value must contain fields whose names exactly match the collection fields, such as the explicit primary key `id` and the `FloatVector` field `vector`. The number of vector elements must match the dimension defined by the collection. ## Configuration ### Milvus Connection and Authentication #### `public.endpoint` The base address of the Milvus or Zilliz Cloud REST v2 API. * **Type**: `string` * **Default**: Empty string * **Importance**: Medium * **Valid values / Notes**: The connector does not validate the URL format. At runtime, you must provide a non-empty HTTP(S) address that the worker can access. When the task starts, it appends collection REST paths to this address and immediately sends requests. * **Required at runtime**: Yes #### `token` The Bearer credential used for Milvus REST requests. * **Type**: `password` * **Default**: `db_admin:****` * **Importance**: High * **Valid values / Notes**: The default is the literal value in ConfigDef, not a usable credential. The task reads the raw configuration directly before applying ConfigDef defaults, so at runtime you must explicitly provide a non-empty, valid API token or the `username:password` form required by the target deployment. Store this value using a secure configuration management method. * **Required at runtime**: Yes ### Target Database and Collection #### `database.name` The Milvus database that contains the target collection. * **Type**: `string` * **Default**: `default` * **Importance**: Medium * **Valid values / Notes**: Use the name of an existing database that the token can access. The connector does not validate the name locally or create the database. The server response determines whether the configuration is usable. #### `collection.name` The Milvus collection that receives Kafka records. * **Type**: `string` * **Default**: Empty string * **Importance**: Medium * **Valid values / Notes**: At runtime, you must specify an existing collection in the `LoadStateLoaded` state. The connector does not create or load the collection. Field names and types must be compatible with the message values, auto-ID must be disabled, and the message values must provide the primary key. * **Required at runtime**: Yes ### Kafka Input and Tasks #### `connector.class` The Milvus Sink Connector implementation class to load. * **Type**: `string` * **Default**: None * **Importance**: High * **Valid values / Notes**: Use `com.milvus.io.kafka.MilvusSinkConnector`. You can also use an alias that Kafka Connect can resolve to this class. * **Required**: Yes #### `topics` The list of Kafka topics to consume. * **Type**: `list` * **Default**: Empty list * **Importance**: High * **Valid values / Notes**: Use comma-separated topic names. Exactly one of `topics` and `topics.regex` must be non-empty. Message values from all selected topics must be compatible with the same target collection. #### `topics.regex` Selects Kafka topics to consume using a Java regular expression. * **Type**: `string` * **Default**: Empty string * **Importance**: High * **Valid values / Notes**: Use a non-empty, compilable Java regular expression. Exactly one of `topics.regex` and `topics` must be non-empty. All matched topics are written to the same target collection. #### `tasks.max` The maximum number of tasks that the connector may create. * **Type**: `int` * **Default**: `1` * **Importance**: High * **Valid values / Notes**: Must be greater than or equal to `1`. Effective parallelism is also limited by the number of partitions in the subscribed topics. Each active task independently checks the same collection and may issue concurrent upserts to it. ### Data Conversion #### `key.converter` Overrides the converter used for Kafka record keys at the connector level. * **Type**: `class` * **Default**: `null` * **Importance**: Low * **Valid values / Notes**: If omitted or set to `null`, the worker's key converter is inherited. If explicitly configured, the value must be an instantiable implementation of `org.apache.kafka.connect.storage.Converter`. The Milvus Sink Connector does not read the converted record key to populate fields or select a collection. #### `value.converter` Specifies the converter used for Kafka record values at the connector level. * **Type**: `class` * **Default**: `null` * **Importance**: Low * **Valid values / Notes**: If omitted or set to `null`, the worker's value converter is inherited. If explicitly configured, the value must be an instantiable implementation of `org.apache.kafka.connect.storage.Converter`. The top-level converted object must be a Connect `Struct` or `java.util.HashMap`, and its field names and convertible values must be compatible with the target collection schema. Sub-configurations for the converter itself are defined by the selected converter. ## Best Practices ### Scale Processing Tasks by Kafka Partition **Use case**: A single task has become the processing bottleneck for continuous writes, the input topic has multiple partitions, and messages in all partitions follow the same collection schema. In this situation, you can gradually increase the number of tasks so that the Kafka consumer group distributes partitions among them. **Configuration example**: Add or override the following property in the Quick Start configuration: ```properties theme={null} tasks.max=3 ``` **Key considerations**: `3` is only an example upper limit. Adjust it gradually based on the number of input partitions, Milvus write capacity, and observed latency. Setting more tasks than the number of assignable partitions does not increase record-processing parallelism. Multiple tasks synchronously upsert to the same collection independently. The connector does not provide global ordering across partitions or tasks, and it does not guarantee that throughput increases linearly with the number of tasks. ### Subscribe to Multiple Homogeneous Topics by Naming Pattern **Use case**: An active data pipeline needs to automatically subscribe to new topics whose names follow a common pattern, and the fields and types in messages from all these topics are compatible with the same target collection. In this situation, you can use a regular-expression subscription to reduce the work of maintaining topic lists individually. **Configuration example**: Keep the Milvus connection, collection, and converter configuration from the Quick Start, remove `topics`, and add: ```properties theme={null} topics.regex=milvus-events-.* ``` **Key considerations**: `topics.regex` and `topics` cannot both be non-empty. All topics matched by the regular expression are written to the same `database.name` and `collection.name`. The connector does not automatically select a collection based on the topic name. If different topics require different collections or schemas, use separate connector instances. ## Monitoring ### What to Monitor Monitor Kafka Connect health, connector and task status, throughput, latency, offset commits, errors, retries, and worker JVM signals. Monitor DLQ activity only when the corresponding error handling is enabled. ### Import the Grafana Dashboard Verify that Connect metrics are available through a Grafana data source and that the collected labels meet the dashboard's filtering requirements. Download the [Kafka Connect Dashboard](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json), import the JSON into Grafana, and select the corresponding data source. ## Limitations * Version 1.0.1 does not send insert or upsert requests for collections with auto-ID enabled and does not report the batch as failed. Therefore, you must disable auto-ID and explicitly provide the primary key in the message value. * The connector does not create or load collections, and it does not automatically migrate schemas. Each task reads the collection schema only once at startup. After modifying the schema while the task is running, restart the task so that it reads the schema again. * Top-level record values accept only a Connect `Struct` or a concrete `java.util.HashMap`. Top-level primitive values, JSON strings, lists, arrays, byte arrays, and other Map implementations fail conversion inside the connector and are skipped. * Fields are matched by exact, case-sensitive names. Fields in a message that do not exist in the collection schema are ignored, even when dynamic fields are enabled for the collection. The connector does not provide renaming, nested paths, flattening, or default value injection. * `FloatVector` depends on a parsable comma-separated list representation, and the connector does not validate vector dimensions. `Array`, `Float16Vector`, `BFloat16Vector`, and `None` do not have corresponding conversion branches. JSON values are serialized to strings before being sent instead of being sent as structured JSON trees. * Tombstones and records with any field-conversion exception are skipped by the connector itself. This path does not enter Kafka Connect's `ErrantRecordReporter`, DLQ, or the connector retry queue. The task may still return normally and commit offsets past these records. * The set of records delivered to a task by one worker poll forms one synchronous upsert request. The connector has no separate configuration for batch size, byte size, wait time, splitting, asynchronous queues, or rate limiting. One entity rejected by Milvus may cause the entire request to fail. * A non-200 HTTP status, a non-`0` Milvus response code, a network exception, or a response parsing exception terminates the current task as a non-recoverable runtime exception. The connector does not distinguish `429`, `5xx`, authentication, timeout, or validation errors, does not process `Retry-After`, and does not use `errors.tolerance` or the DLQ for this internal REST failure path. * Kafka offset commits and Milvus upserts are not transactional. If a failure occurs after Milvus accepts the request but before the offset is committed, records may be replayed after recovery. The connector does not guarantee exactly-once delivery, absence of duplicates, or global ordering across tasks. * The connector logs the collection name and complete data list for successfully converted batches at INFO level, which may include business fields and large vectors. Restrict access to task logs and configure an appropriate retention policy. ## FAQ ### Why does the task fail immediately at startup? When a task starts, it checks in sequence whether the collection exists and is loaded, and then reads the collection schema. Check whether `public.endpoint` is an accessible REST base address, whether `token` has the required permissions, whether `database.name` and `collection.name` are correct, and whether the collection is already in the `LoadStateLoaded` state. After correcting the target resource or permissions, restart the failed task. ### Why does the connector show as running while no new data appears in Milvus? First, verify that auto-ID is disabled for the collection. When version 1.0.1 encounters a collection with auto-ID enabled, it does not issue a write and does not report the batch as failed. Next, check whether `value.converter` converts the message value to a `Struct` or `HashMap`, whether the value contains an explicit primary key and `FloatVector`, whether field names exactly match the collection schema, and whether field-conversion exceptions appear in the logs. Records that fail conversion are skipped and do not automatically enter the DLQ. ### Why are some fields not written to the collection? The connector writes only fields that exist in the collection schema read at startup and whose names match exactly. Check case, field types, and whether the schema was modified after the task started. Extra fields are not written to dynamic fields. If the collection schema has changed, ensure that messages are compatible with the new schema, and then restart the task to read the description again. ### Why does one Milvus REST error cause the task to enter `FAILED`? The connector throws unsuccessful HTTP responses, nonzero Milvus response codes, network exceptions, and response parsing exceptions as ordinary runtime exceptions instead of Kafka Connect retriable exceptions. `errors.tolerance`, framework retry parameters, and the DLQ do not take over this internal upsert path. Check the task logs and Milvus service status. After correcting address, authentication, schema, request data, or target service issues, restart the task. Records whose offsets were not committed may be upserted again. ### Why do records from multiple topics all go to the same collection? `topics` and `topics.regex` determine only the Kafka input scope and do not participate in target routing. All records in one connector instance are written to the fixed `database.name` and `collection.name`. To write different topics to different collections, create separate connector instances for different sets of topics and configure their target databases and collections separately.