Skip to main content

​
Overview

The Vectara Sink Connector consumes records from Kafka topics and converts each record into a document in a Vectara Corpus. The Kafka record key becomes the Vectara document ID; top-level fields in the record value become document content, and selected fields can also be written to document metadata. Subsequent records with the same document ID update the target document, while tombstone records with a null value delete the corresponding document. The connector is suitable for continuously writing business content, knowledge entries, or other structured events from Kafka to an existing Vectara Corpus. Multiple Kafka topics can write to the same fixed Corpus, but document IDs do not include topic or partition information. Producers must therefore provide record keys that remain stable and do not conflict within the target Corpus.

​
Prerequisites

  • Create the target Vectara Corpus in advance and prepare an API Key with permission to create, update, and delete documents in that Corpus. If client certificate authentication is enabled, also prepare a keystore that the Connect Worker can read or its Base64-encoded content.

​
License

Licensed under the Apache License 2.0.

​
Quick Start

Prepare a Connect Cluster, Kafka, and a Vectara Corpus, and verify network connectivity and API Key permissions. For preparation and management instructions, see Manage Connectors.
Replace <vectara-api-key>, <vectara-customer-id>, and <vectara-corpus-id> with actual values. When writing records to vectara-documents, use a non-null string key and a top-level JSON object as the record value. For example, the key article-1001 becomes the target document ID. value.converter.schemas.enable=false is a JSON Converter setting that converts a regular JSON object into the top-level Map supported by the connector.

​
Configuration

​
Vectara Connection and Account

​
api.key

The credential used to call the Vectara API.
  • Type: password
  • Default: None
  • Importance: High
  • Valid values / Notes: Required. Configuration parsing permits an empty value, but an empty value cannot provide valid authentication. Manage this credential through a controlled Kafka ConfigProvider; do not write it to logs or public configuration repositories.
  • Required: Yes

​
api.url

The base URL of the Vectara API.
  • Type: string
  • Default: https://api.vectara.io/
  • Importance: High
  • Valid values / Notes: Must be an absolute URI. Configuration validation does not restrict the protocol, check the host, or verify that the address is reachable. Use a trusted HTTPS endpoint in production.

​
customer.id

The Vectara customer ID.
  • Type: long
  • Default: None
  • Importance: High
  • Valid values / Notes: Required and must be parseable as a long. Version 1.0.4 parses and stores this value, but it has not been confirmed to participate in API requests, headers, or Corpus routing. Do not rely on it for account routing.
  • Required: Yes

​
Corpus Routing

​
corpus.id.location

Specifies whether the target Corpus ID is obtained from the Connector configuration or a record field.
  • Type: string
  • Default: Config
  • Importance: High
  • Valid values / Notes: Accepts only the case-sensitive values Config or Field. Version 1.0.4 should use Config and specify a fixed Corpus through corpus.id. In this version, Field routing does not read the field named by corpus.field and cannot process tombstone deletions.

​
corpus.id

The fixed target Corpus ID.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid values / Notes: Required and must not be empty when corpus.id.location=Config. Records from all subscribed topics are written to this Corpus and share the same document ID namespace.
  • Required: Conditionally required

​
corpus.field

Declares the name of the top-level record-value field used to obtain the Corpus ID.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid values / Notes: Although version 1.0.4 exposes this configuration, the runtime does not read its value. Setting it does not change the field that Field mode actually looks up. Use corpus.id.location=Config with a non-empty corpus.id instead.

​
Documents and Metadata

​
document.metadata.fields

Selects top-level fields from the record value to write to Vectara document metadata.
  • Type: list
  • Default: Empty list
  • Importance: High
  • Valid values / Notes: Use a comma-separated list of case-sensitive top-level field names. Fields missing from a Map are ignored; for a Struct, only fields present in its Schema are checked. Duplicate names are deduplicated. If a record field and metadata.default.* produce metadata with the same name, the record field value takes precedence.

​
exclude.metadata.fields.from.document

Controls whether fields selected by document.metadata.fields are excluded from the document body.
  • Type: boolean
  • Default: false
  • Importance: Low
  • Valid values / Notes: false means selected fields appear in both metadata and document content; true means these fields are retained only as metadata. This setting does not exclude metadata keys added only through metadata.default.*.

​
metadata.default.*

Adds fixed metadata to every Vectara document. Replace * with the actual metadata key. For example, metadata.default.source=kafka produces the metadata entry source=kafka.
  • Type: Raw configuration value
  • Default: None
  • Importance: Not applicable
  • Valid values / Notes: This is a dynamic prefix configuration rather than a fixed Connector ConfigDef entry, so it has no uniform type, default, or validation rules. The prefix is removed; a record value selected through document.metadata.fields overrides a fixed value with the same name. Do not use this prefix to write sensitive information.

​
Batching and Concurrency

​
document.batch.timeout.seconds

The maximum time, in seconds, to wait for asynchronous document tasks while processing each batch of Kafka records.
  • Type: long
  • Default: 60
  • Importance: Low
  • Valid values / Notes: The configuration has no minimum-value validation, but use a positive value. This value is not an HTTP request timeout, a task cancellation deadline, or an offset commit barrier. Remote operations that remain incomplete after the timeout may continue to run.

​
callback.executor.pool.size

The number of local threads each Sink Task uses to execute Vectara requests.
  • Type: int
  • Default: 10
  • Importance: Low
  • Valid values / Notes: Should be greater than or equal to 1. Although 0 or a negative number passes configuration type parsing, it causes Task startup to fail. Total concurrency also grows with the number of active Tasks, and records in the same batch may complete concurrently.

​
max.requests

Declares the maximum number of requests allowed.
  • Type: int
  • Default: 10
  • Importance: Low
  • Valid values / Notes: The valid range is 1 to 20, inclusive. Version 1.0.4 validates and stores this value, but it has not been confirmed to limit local executor or client request concurrency. When tuning throughput, use tasks.max and callback.executor.pool.size as the primary controls.

​
TLS Client Keystore

​
ssl.keystore.location

Specifies the source of the client keystore.
  • Type: string
  • Default: None
  • Importance: High
  • Valid values / Notes: Accepts only the case-sensitive values File, Inline, or None. None does not load a client keystore; File uses ssl.keystore.path; Inline uses ssl.key.inline. The keystore type is the JVM default.

​
ssl.keystore.path

The path to the client keystore file.
  • Type: string
  • Default: Empty string
  • Importance: High
  • Valid values / Notes: Conditionally required when ssl.keystore.location=File. The file must be readable by the Connect Worker process and must match the JVM default KeyStore type and the configured password.
  • Required: Conditionally required

​
ssl.keystore.password

The password used to load a File or Inline client keystore.
  • Type: password
  • Default: Empty string
  • Importance: High
  • Valid values / Notes: Conditionally required when the keystore uses a non-empty password. Ignored when ssl.keystore.location=None. Manage this credential through a controlled Kafka ConfigProvider.
  • Required: Conditionally required

​
ssl.key.inline

The Base64-encoded client keystore content.
  • Type: string
  • Default: Empty string
  • Importance: High
  • Valid values / Notes: Conditionally required when ssl.keystore.location=Inline. The content must decode to a keystore of the JVM default KeyStore type. Although the configuration type is a string, it contains private key material and must be handled as sensitive information.
  • Required: Conditionally required

​
Input Subscription

​
topics

The list of Kafka topics to consume.
  • Type: list
  • Default: Empty list
  • Importance: High
  • Valid values / Notes: Mutually exclusive with topics.regex. Use a comma-separated list of non-empty topic names. When multiple topics write to the same Corpus, coordinate record keys to prevent different topics from producing the same document ID.
  • Required: Conditionally required

​
topics.regex

Subscribes to Kafka topics using a Java regular expression.
  • Type: string
  • Default: Empty string
  • Importance: High
  • Valid values / Notes: Mutually exclusive with topics and must be a valid, non-empty Java regular expression. When a DLQ is configured, the expression must not match the DLQ topic.
  • Required: Conditionally required

​
Connector Identity and Tasks

​
connector.class

The Vectara Sink Connector implementation class to load.
  • Type: string
  • Default: None
  • Importance: High
  • Valid values / Notes: Version 1.0.4 must use com.vectara.kafka.connect.VectaraDocumentSinkConnector.
  • Required: Yes

​
tasks.max

The maximum number of Tasks that this connector may create.
  • Type: int
  • Default: 1
  • Importance: High
  • Valid values / Notes: Must be greater than or equal to 1. The number of active Tasks cannot exceed the number of subscribed Kafka partitions; each Task also creates callback.executor.pool.size local execution threads.

​
tasks.max.enforce

Controls whether Kafka Connect enforces the framework constraint on tasks.max.
  • Type: boolean
  • Default: true
  • Importance: Low
  • Valid values / Notes: Still configurable in Kafka Connect 3.9.1, but deprecated and scheduled for removal in a later major version. Keep it enabled and use tasks.max to control the Task limit.
  • Deprecated: Yes

​
Data Conversion

​
key.converter

Specifies the Converter for Kafka record keys. The converted key is passed through toString() to produce the Vectara document ID.
  • Type: class
  • Default: null
  • Importance: Low
  • Valid values / Notes: If omitted, inherits the Worker’s key Converter. When configured, it must be an instantiable concrete Converter class. Every write and tombstone deletion record must produce a non-null, stable key string that is unique within the target Corpus.

​
value.converter

Specifies the Converter for Kafka record values.
  • Type: class
  • Default: null
  • Importance: Low
  • Valid values / Notes: If omitted, inherits the Worker’s value Converter. When configured, it must be an instantiable concrete Converter class. Non-tombstone records must convert to a top-level Connect Struct or Java Map; Map field keys must be strings. Top-level strings, numbers, arrays, and bytes are not supported.

​
Error Handling

​
errors.tolerance

Specifies how Kafka Connect handles tolerable record errors.
  • Type: string
  • Default: none
  • Importance: Medium
  • Valid values / Notes: Accepts only none or all. This setting does not guarantee that every failed asynchronous Vectara write or deletion can be retried, skipped, or sent to the DLQ. Confirm delivery status using Task logs and target-system results.

​
errors.deadletterqueue.topic.name

The name of the topic used to store error records that meet Kafka Connect DLQ conditions.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid values / Notes: An empty string disables the DLQ. A non-empty DLQ topic must not also appear in topics or be matched by topics.regex. Configuring a DLQ does not mean every asynchronous failure captured by the Connector is published to that topic.

​
errors.deadletterqueue.topic.replication.factor

The replication factor used when Kafka Connect creates a missing DLQ topic.
  • Type: short
  • Default: 3
  • Importance: Medium
  • Valid values / Notes: Used only when the configured DLQ topic does not exist and Kafka Connect creates it. The value must be appropriate for the number of available brokers and the replication policy of the Kafka cluster.

​
Best Practices

​
Use Business Classifications as Metadata and Exclude Them from the Document Body

Applicable scenario: The connector already writes content to a fixed Corpus. During continuous operation, you need to add a source identifier to every document and use classification fields such as language or department from each record for filtering, while keeping those classification values out of the document body. Configuration example:
Key points: Each JSON object should provide language and department at the top level; missing fields are ignored. The fixed source=kafka value is added to every document, while record fields are written to metadata using their original values. When exclusion is enabled, language and department no longer generate document content fragments but remain available as metadata; other non-null top-level fields continue to be converted into document content.

​
Scale Write Concurrency Gradually by Kafka Partition

Applicable scenario: A single Task is writing reliably, the Kafka topic has multiple partitions, and backlog or end-to-end latency indicates that more processing capacity is required. First increase the number of Tasks, then carefully adjust the number of local threads per Task while monitoring Vectara throttling, failures, and offset commits. Configuration example:
Key points: The number of active Tasks is limited by the number of Kafka partitions, and the upper bound on total local execution threads is approximately the number of active Tasks multiplied by callback.executor.pool.size. Start with smaller values and increase them gradually based on target-system throttling, error rates, and backlog. document.batch.timeout.seconds only controls how long put waits for asynchronous tasks; it does not cancel requests after the timeout or guarantee that offsets are committed only after remote writes complete. Consecutive updates to the same document ID may execute concurrently or be reordered, so do not use completion order to represent business state.

​
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

Confirm that Connect metrics are available in a Grafana data source and that collected labels satisfy the dashboard filters. Download the Kafka Connect Dashboard, import the JSON into Grafana, and select the corresponding data source.

​
Limitations

  • Non-tombstone records must have a non-null key, and their values must be a top-level Connect Struct or Java Map; top-level primitive values, arrays, strings, and bytes cannot be used as document input. Nested objects are not expanded recursively and are instead written to document content as strings.
  • In version 1.0.4, corpus.field does not participate in runtime field selection. corpus.id.location=Field actually looks for a top-level field named Field, and tombstone records have no value available for routing. Production configurations should use fixed Config routing.
  • Document IDs are derived only from the string representation of record keys and do not include the Kafka topic or partition. When multiple topics write to the same Corpus, identical key strings refer to the same document.
  • Document updates use a non-atomic delete-then-create process. Concurrent updates or deletions of the same document ID may be reordered, and the document may be temporarily absent. The Connector does not provide end-to-end transactions or exactly-once delivery guarantees.
  • After a put wait times out, remote operations may continue, and flush does not wait for incomplete requests. Kafka offsets may be committed before the corresponding Vectara operation completes or fails. A process failure may also cause successfully written records to be replayed after recovery.
  • max.requests and the required customer.id are parsed in version 1.0.4, but they have not been confirmed to limit request concurrency or participate in Vectara API requests, respectively. Do not rely on these two configurations for those runtime controls.

​
FAQ

​
Why Does the Task Report That the Record Value Type Is Unsupported?

Check the output of value.converter, not only the original serialization format in Kafka. Non-tombstone records must be converted into a top-level Map or Struct. For regular JSON objects, configure JsonConverter as shown in the Quick Start and set value.converter.schemas.enable=false. Also confirm that all top-level Map field keys are strings and that the record key is not null.

​
Why Are Records Not Written to the Expected Corpus After Configuring corpus.field?

This is a field-routing limitation in version 1.0.4: this version does not read the value of corpus.field and instead looks for a top-level field named Field when operating in Field mode. Change corpus.id.location to Config and write the target Corpus ID to a non-empty corpus.id. If multiple Corpora are required, create separate Connector instances for different input ranges, each using a fixed Corpus.

​
Why Do Metadata Fields Still Appear in the Document Body?

document.metadata.fields only selects metadata and does not remove those fields from the document body by default. Set exclude.metadata.fields.from.document=true, and confirm that each field name exactly matches the corresponding top-level record field, including case. Fixed metadata added by metadata.default.* does not automatically correspond to or exclude fields with the same name from the document body.

​
Why Is There No Significant Throughput Change After Increasing max.requests?

Version 1.0.4 validates the range of max.requests, but it has not been confirmed to control executor or client concurrency. First check the number of Kafka partitions, active Tasks, and backlog, then adjust concurrency gradually through tasks.max and callback.executor.pool.size. At the same time, monitor Vectara throttling, error rates, and Worker resource usage.

​
Why Have Kafka Offsets Advanced While Vectara Documents Are Still Missing or Appear Later?

The Connector executes remote operations asynchronously. A batch wait timeout does not cancel requests that have not completed, and flush does not wait for them. As a result, a window may exist between offset advancement and target-system completion. Check Connector and Task status, logs for the relevant time period, errors and DLQ activity, and the final document state in Vectara. When recovering data, use the same stable record keys and evaluate the business impact of replay and non-atomic updates.