> ## 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. # Timeplus Sink Connector > Configure and operate the Timeplus Sink Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting. ## Overview The Timeplus Sink Connector consumes records from Kafka topics and writes record values to a specified Timeplus stream through the Timeplus v1beta2 HTTP API. It sits between Kafka event streams and Timeplus real-time analytics storage, making it suitable for ingesting logs, business events, and other continuously generated text or JSON data into a single destination stream. The Connector converts the value of each Kafka record to a string and combines the values in record order into a newline-delimited request body. In `raw` mode, each string is written as one line of text. In JSON mode, the strings are sent to the streaming JSON ingestion endpoint. Kafka record keys, topics, partitions, offsets, timestamps, and headers are not automatically mapped to Timeplus columns. ## Prerequisites * Prepare an accessible Timeplus Workspace and an API Key with write permission. The default configuration requires a pre-created destination stream compatible with the input format. If automatic stream creation is enabled, the API Key must also have stream creation permission; JSON mode additionally requires schema inference permission, and the first record value must be a JSON object. ## License Licensed under the Apache License 2.0. ## Quick Start Prepare a Connect Cluster, Kafka, a Timeplus Workspace, and a destination stream, and verify network connectivity and access permissions. For preparation and management instructions, see [Manage Connectors](../manage-connectors). The following minimal practical configuration consumes strings from one Kafka topic and writes them in `raw` mode to a pre-created Timeplus stream. ```properties theme={null} connector.class=com.timeplus.kafkaconnect.TimeplusSinkConnector topics= value.converter=org.apache.kafka.connect.storage.StringConverter timeplus.sink.address= timeplus.sink.workspace= timeplus.sink.apikey= timeplus.sink.stream= ``` Replace ``, ``, ``, ``, and `` with actual resources. The destination stream must already exist and be able to accept the string representation of each record value. In `raw` mode, this is typically a String column named `raw`. Do not write the API Key to logs or commit it to version control. ## Configuration ### Timeplus Connection and Authentication #### `timeplus.sink.address` Sets the base address used to construct URLs for the Timeplus stream, ingestion, and schema inference APIs. * **Type**: `string` * **Default**: `https://dev.timeplus.cloud` * **Importance**: High * **Valid values / Notes**: Use a Timeplus base URL that every Connect Worker can access. The Connector does not validate the scheme, host, or connectivity, and it does not normalize trailing slashes. Explicitly set the correct address for other Timeplus Cloud regions or self-hosted deployments. #### `timeplus.sink.workspace` Sets the Workspace identifier used in Timeplus API URLs. * **Type**: `string` * **Default**: `default` * **Importance**: High * **Valid values / Notes**: Use the ID of a Workspace that exists at the destination address. The value is added directly to the URL without encoding. The Connector does not validate whether it is empty or whether the Workspace exists. The API Key must be authorized to access this Workspace. #### `timeplus.sink.apikey` Sets the API Key sent in the `X-API-KEY` header of Timeplus requests. * **Type**: `password` * **Default**: Empty string * **Importance**: High * **Valid values / Notes**: Use an API Key accepted by the destination Workspace. The Connector does not validate its length, format, or whether it is empty. Deployments that require authentication must explicitly provide a valid key. When automatic stream creation is enabled, the key must also have permission to create streams and perform schema inference. This configuration contains sensitive information and must be supplied through a secure configuration injection method. ### Destination Stream and Data Format #### `timeplus.sink.stream` Sets the name of the Timeplus stream that receives records. * **Type**: `string` * **Default**: None * **Importance**: High * **Required**: Yes * **Valid values / Notes**: Use a stream name accepted by Timeplus. The value is added directly to the ingestion URL without encoding. The Connector does not validate empty strings or naming rules. When `timeplus.sink.createStream=false`, the destination stream must already exist. #### `timeplus.sink.dataFormat` Sets the data format of ingestion requests and the schema generation path used during automatic stream creation. * **Type**: `string` * **Default**: `raw` * **Importance**: High * **Valid values / Notes**: Use the case-sensitive value `raw` or `json`. Only an exact value of `raw` uses the newline-delimited text endpoint. When automatic stream creation is enabled, this mode creates a stream with a single String column named `raw`. Any other value enters the streaming JSON branch, but the Connector does not validate whether such values are valid. Both modes send the `toString()` result of each converted value. In JSON mode, that string must conform to the Timeplus streaming JSON endpoint and the destination stream schema. #### `timeplus.sink.createStream` Controls whether each Task attempts to create the destination stream when it processes its first non-empty record batch. * **Type**: `boolean` * **Default**: `false` * **Importance**: High * **Valid values / Notes**: `false` uses a pre-created stream. When set to `true`, `raw` mode attempts to create a stream containing a String column named `raw`, while JSON mode uses the first record value for schema inference before creating the stream. The Connector does not first check whether the stream exists. Each Task has an independent first-creation state, and after a creation failure, that Task might continue writing without attempting creation again. ### Kafka Input Subscription and Tasks #### `connector.class` Selects the Timeplus Sink Connector implementation class. * **Type**: `string` * **Default**: None * **Importance**: High * **Required**: Yes * **Valid values / Notes**: Use `com.timeplus.kafkaconnect.TimeplusSinkConnector`. #### `topics` Specifies the list of Kafka topics consumed by the Connector. * **Type**: `list` * **Default**: Empty list `[]` * **Importance**: High * **Required**: Mutually exclusive with `topics.regex` * **Valid values / Notes**: Use comma-separated topic names. Exactly one of this setting and a non-empty `topics.regex` must be configured. Sink configuration validation fails if both are set or both are empty. Records from all selected topics are written to the same `timeplus.sink.stream`. #### `topics.regex` Selects the Kafka topics consumed by the Connector through a Java regular expression. * **Type**: `string` * **Default**: Empty string * **Importance**: High * **Required**: Mutually exclusive with `topics` * **Valid values / Notes**: Use a valid, non-empty Java regular expression. Exactly one of this setting and `topics` must be configured. Records from all matching topics are written to the same `timeplus.sink.stream`. #### `tasks.max` Sets the maximum number of Tasks that the Connector can create. * **Type**: `int` * **Default**: `1` * **Importance**: High * **Valid values / Notes**: Must be at least `1`. Actual parallelism depends on the number of input topic partitions, the number of Workers, and partition assignment. Each Task sends HTTP requests independently. When automatic stream creation is enabled, each Task might also independently attempt to create the same stream. ### Record Value Conversion #### `value.converter` Sets the Converter used by the Worker to deserialize Kafka record values. * **Type**: `class` * **Default**: `null` * **Importance**: Low * **Valid values / Notes**: When omitted, this setting inherits the Worker-level configuration. An explicit value must be an instantiable implementation of `org.apache.kafka.connect.storage.Converter`. The Connector calls `toString()` on the converted object, so select a Converter that produces the intended text lines or valid JSON text. `org.apache.kafka.connect.storage.StringConverter` is recommended to preserve the textual representation of Kafka values. The `toString()` results of Structs, Maps, byte arrays, and similar objects are not guaranteed to be valid JSON. ## Best Practices ### Write JSON Events to a Pre-created Stream **Applicable scenario**: Event fields and types are already defined, and you want to continuously write JSON objects from Kafka to a Timeplus stream with a corresponding schema so that Timeplus can query and analyze individual fields in real time. **Configuration example**: ```properties theme={null} connector.class=com.timeplus.kafkaconnect.TimeplusSinkConnector topics= value.converter=org.apache.kafka.connect.storage.StringConverter timeplus.sink.address= timeplus.sink.workspace= timeplus.sink.apikey= timeplus.sink.stream= timeplus.sink.dataFormat=json ``` Replace the placeholders with actual resources, and pre-create a destination stream compatible with the event fields. The value of each record in the input topic must be a string containing one valid JSON object. Do not use a Converter that converts values to a non-JSON `toString()` representation. **Key considerations**: Pre-creating the stream lets you define field names and types before ingestion instead of allowing the first event to determine the inferred result. The Connector does not normalize JSON or perform subsequent schema evolution. New fields, type changes, and nested structures must first be verified as compatible with the destination stream. ### Expand Input Topics by Naming Convention **Applicable scenario**: Multiple business topics follow a shared naming convention, and more topics of the same kind will be added later. You want the Connector to automatically subscribe to the matching set and ingest their records into one Timeplus stream. **Configuration example**: ```properties theme={null} connector.class=com.timeplus.kafkaconnect.TimeplusSinkConnector topics.regex=events\\..* value.converter=org.apache.kafka.connect.storage.StringConverter timeplus.sink.address= timeplus.sink.workspace= timeplus.sink.apikey= timeplus.sink.stream= ``` Remove `topics` when using `topics.regex`, and replace the other placeholders with actual resources. Ensure that the record values from all matching topics can be accepted by the same destination stream and data format. **Key considerations**: In a properties file, the regular expression escape backslash in `events\..*` must itself be escaped according to Java Properties rules, so the example uses `topics.regex=events\\..*`. After parsing, the actual regular expression passed to Kafka Connect is `events\..*`, which matches only topics whose names begin with `events.`. Regex subscription simplifies maintenance of a related topic set, but the Connector does not automatically route different topics to different streams or write topic names to destination columns. Check the matching scope before changing the expression to prevent unrelated topics from being written to the same stream. ### Increase Task Parallelism When Backlog Grows **Applicable scenario**: The Connector is running steadily, the input topic has multiple partitions, and consumer lag continues to grow. You want to use multiple Tasks to send HTTP ingestion requests in parallel. **Configuration example**: ```properties theme={null} connector.class=com.timeplus.kafkaconnect.TimeplusSinkConnector topics= tasks.max=3 value.converter=org.apache.kafka.connect.storage.StringConverter timeplus.sink.address= timeplus.sink.workspace= timeplus.sink.apikey= timeplus.sink.stream= ``` Replace the placeholders with actual resources, first verify that the input topic has multiple assignable partitions, and use a pre-created destination stream. Increase `tasks.max` gradually while monitoring consumer lag, HTTP latency, Timeplus rate limiting, and Worker resource usage. **Key considerations**: `tasks.max` is an upper limit, and the actual number of Tasks depends on partition assignment. HTTP requests from different Tasks can complete concurrently. The Connector does not provide global ordering or deduplication across Tasks. Increasing parallelism also does not split an oversized request batch. ## Monitoring ### What to Monitor Monitor the health and status changes of Kafka Connect Workers, the Connector, and its Tasks. Track input throughput, consumer lag, processing latency, offset commits, errors, retries, and Worker JVM CPU, memory, and garbage collection signals. Also compare the actual Timeplus ingestion volume with Worker logs, because some HTTP ingestion failures might only be logged without causing the Task to fail. Monitor DLQ activity only when the deployment has enabled the corresponding Kafka Connect error handling. ### Import the Grafana Dashboard Verify that Kafka Connect metrics are available in a Grafana data source and that the collected labels satisfy the dashboard filters. 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 * The Connector calls `toString()` on each record value and appends a newline. It does not read the Kafka key, topic, partition, offset, timestamp, headers, or Connect schema to generate destination columns. The string representations of Structs, Maps, byte arrays, and similar objects are not guaranteed to be valid JSON. * When an ingestion request ultimately returns a non-success status or an `IOException` occurs, the Connector might only log the failure without propagating it to Kafka Connect. A running Task or committed offsets alone do not prove that the corresponding data was written to Timeplus. Therefore, this version cannot be relied on to provide at-least-once delivery guarantees. * Timeplus ingestion and Kafka offset commits are not atomic. A failure after a successful ingestion but before the offset commit can replay the entire batch. If the Connector swallows an ingestion failure, `put` returns normally and the Worker can subsequently advance and commit the offsets, creating a data gap. The Connector does not provide transactions, idempotency keys, automatic deduplication, or exactly-once delivery. * Automatic stream creation does not check whether the stream already exists, and each Task might independently attempt creation after startup or restart. After a creation request fails, the Task might continue writing and mark its local state as having attempted creation. JSON automatic stream creation infers the schema only from the first record and does not handle subsequent schema evolution. * All input topics and partitions in one Connector configuration are written to the same destination stream. The Connector does not provide configuration for routing by topic or partition to multiple streams, and it does not guarantee global ordering across Tasks. * The Connector combines the entire batch received by each non-empty `put` call into one in-memory request body. It does not provide configuration for batch size, byte limits, request splitting, asynchronous queues, or client timeouts. If a single record or an entire batch exceeds Timeplus or proxy limits, the Connector does not automatically reduce the batch size. ## FAQ ### The Task is running. Why is there no new data in Timeplus? First check the Worker logs for Timeplus HTTP failure statuses or connection exception messages. Then verify `timeplus.sink.address`, `timeplus.sink.workspace`, `timeplus.sink.apikey`, `timeplus.sink.stream`, and the destination stream schema. This version might allow `put` to return normally after an ingestion failure, so directly compare Kafka input volume, committed offsets, and the actual Timeplus row count. After correcting the destination address, permissions, stream, or data format, use new test records to confirm that ingestion has recovered. ### JSON ingestion is rejected. How do I inspect record values? Confirm that `timeplus.sink.dataFormat=json`, and check whether the `toString()` result of the object produced by `value.converter` is one valid JSON object. With `StringConverter`, the Kafka value itself must be JSON text. With a Struct, Map, or another structured object, do not assume that its default string representation is valid JSON. Also verify that field names, types, and nested structures are compatible with the destination stream schema. ### Why was the stream not created correctly after automatic stream creation was enabled? Confirm that the API Key has stream creation and schema inference permissions. In JSON mode, the first record value must be parseable as a JSON object. Multiple Tasks might attempt to create the same stream concurrently, and an existing stream is not detected in advance. Because a creation failure might only be logged and that Task might not retry, prefer pre-creating and validating the stream in production, then keep `timeplus.sink.createStream=false`. ### Why can a restart cause duplicate data or data gaps? Timeplus HTTP ingestion and Kafka offset commits are separate steps. If Timeplus has accepted a batch but its offset has not yet been committed, the batch might be replayed after a restart. If an ingestion failure is logged and swallowed by the Connector, `put` still returns normally and the Worker can subsequently advance and commit the offset, so the batch might not be replayed automatically. Inspect Worker logs, offsets, and Timeplus row counts, and use stable event identifiers for application-level auditing or deduplication. Do not treat this version as an at-least-once or exactly-once delivery implementation. ### The Connector reports a `topics` and `topics.regex` configuration error at startup. What should I do? Check whether both settings are empty or both are configured. For a fixed set of topics, keep only a non-empty `topics` setting. To subscribe by naming convention, keep only a valid, non-empty `topics.regex`. Resubmit the configuration after making the change, and confirm that data from every selected topic is suitable for the same Timeplus stream.