> ## 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. # Datadog Logs Sink Connector > Configure and operate the Datadog Logs Sink Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting. ## Overview The Datadog Logs Sink Connector consumes log records from a Kafka topic and sends them to Datadog over HTTPS. It is suitable for collecting application logs, service runtime events, and similar data into a unified log search pipeline. The [official project](https://github.com/DataDog/datadog-kafka-connect-logs) provides this Connector. For each non-null record, the value is converted to JSON and placed in the log's `message` field. The outer payload includes `ddsource=kafka-connect` and a Topic tag, and can consistently add service, host, and environment tags. The Kafka key is not written into the log body; fields in the value are not automatically promoted to Datadog's outer metadata. The Connector sends logs through the [Datadog Logs API](https://docs.datadoghq.com/api/latest/logs/); successful HTTP acceptance does not mean that a log has already been indexed or is queryable. ## Prerequisites * Prepare a valid API key for the target Datadog organization and confirm the organization's [Datadog Site](https://docs.datadoghq.com/getting_started/site/) to avoid sending logs to another site. ## License Licensed under Apache License 2.0. ## Quick Start Prepare a Connect Cluster, Kafka, the log Topic, and the target Datadog organization in advance. Confirm network connectivity and access permissions; see [Manage Connectors](../manage-connectors) for Connect Cluster and Connector preparation and management. The following configuration applies to a Topic whose key is a string or null and whose value is UTF-8 log text. ```properties theme={null} connector.class=com.datadoghq.connect.logs.DatadogLogsSinkConnector topics= key.converter=org.apache.kafka.connect.storage.StringConverter value.converter=org.apache.kafka.connect.storage.StringConverter datadog.api_key= datadog.site= ``` Replace the log Topic, API key, and Site domain, such as `datadoghq.com` or `datadoghq.eu`, then apply the configuration. Provide real secrets through controlled credential management and do not commit them to the repository. A string value becomes a JSON string as `message`; even if the text content is JSON, it is not automatically parsed as an object. After connecting, send a small number of logs, search Datadog with `source:kafka-connect` and `topic:`, and check the consumption progress to confirm the pipeline. The Connector does not need to wait for a full batch before sending. ## Configuration ### Connector and Subscription #### `connector.class` Specifies the Connector implementation class. * **Type**: `string` * **Default**: None * **Importance**: High * **Required**: Yes * **Valid values / notes**: Use `com.datadoghq.connect.logs.DatadogLogsSinkConnector`. #### `topics` Specifies the log Topics to consume. * **Type**: `list` * **Default**: Empty list * **Importance**: High * **Valid values / notes**: Comma-separated; exactly one of `topics` and `topics.regex` must be non-empty, and the subscription must not include a DLQ Topic. #### `topics.regex` Subscribes to log Topics by name using a regular expression. * **Type**: `string` * **Default**: Empty string * **Importance**: High * **Valid values / notes**: Use a valid Java regular expression; mutually exclusive with `topics`, and must not match a DLQ Topic. #### `tasks.max` Sets the maximum number of Tasks. * **Type**: `int` * **Default**: `1` * **Importance**: High * **Valid values / notes**: At least `1`; effective consumption parallelism is constrained by Topic partitions and task assignment and does not imply proportional throughput growth. #### `tasks.max.enforce` Controls whether the generated Task count limit is enforced. * **Type**: `boolean` * **Default**: `true` * **Importance**: Low * **Deprecated**: Yes * **Replacement**: No separate replacement parameter; keep the generated Task count within `tasks.max`. * **Valid values / notes**: `true` or `false`; disabling the count limit check is not recommended. ### Datadog Authentication and Target Site #### `datadog.api_key` Authenticates requests to the Datadog log intake endpoint. * **Type**: `password` * **Default**: None * **Importance**: High * **Required**: Yes * **Valid values / notes**: A valid API key for the target organization; an empty string is not a valid credential. The `password` type does not imply end-to-end credential encryption. #### `datadog.site` Specifies the Datadog Site and therefore the log intake host. * **Type**: `string` * **Default**: `null` * **Importance**: Medium * **Valid values / notes**: Enter a Site domain, such as `datadoghq.eu`, rather than the full console URL. A non-empty `datadog.url` takes precedence; when both settings are absent or empty, runtime falls back to `http-intake.logs.datadoghq.com`. This fallback is not the default value of this property. #### `datadog.url` Overrides the log intake host. * **Type**: `string` * **Default**: `null` * **Importance**: Medium * **Valid values / notes**: Enter a host and optional port, such as `http-intake.logs.datadoghq.com:443`; do not include `https://` or a path. The Connector automatically appends HTTPS and `/api/v2/logs`. When non-empty, this setting overrides `datadog.site`; confirm that the target is trusted and supports the API. ### Log Metadata #### `datadog.tags` Adds common tags to all sent logs. * **Type**: `list` * **Default**: `null` * **Importance**: Medium * **Valid values / notes**: A comma-separated list of tags; the Connector also adds the `topic:` tag. #### `datadog.service` Sets a common outer service name for logs. * **Type**: `string` * **Default**: `null` * **Importance**: Medium * **Valid values / notes**: `service` is omitted when unset; the service name is not automatically extracted from each record's value. #### `datadog.hostname` Sets a common outer host name for logs. * **Type**: `string` * **Default**: `null` * **Importance**: Medium * **Valid values / notes**: `hostname` is omitted when unset; the Kafka Broker or Worker host name is not used automatically. #### `datadog.add_published_date` Adds the Kafka record timestamp to the log. * **Type**: `boolean` * **Default**: `false` * **Importance**: Medium * **Valid values / notes**: When `true`, adds a millisecond `published_date` value to logs whose record timestamp is non-null; this does not mean that Datadog automatically treats it as the standard event time. #### `datadog.parse_record_headers` Adds Kafka Headers as a `kafkaheaders` object. * **Type**: `boolean` * **Default**: `false` * **Importance**: Medium * **Valid values / notes**: `true` or `false`; before enabling, confirm that the Header converter is compatible with the actual data and check whether Headers contain sensitive information. Arbitrary Header structures are not guaranteed to convert successfully. ### Proxy #### `datadog.proxy.url` Sets the HTTP proxy host. * **Type**: `string` * **Default**: `null` * **Importance**: Low * **Valid values / notes**: Enter the proxy host name, not a complete URL; when enabled, `datadog.proxy.port` must also be set. No public proxy authentication parameter is available. #### `datadog.proxy.port` Sets the HTTP proxy port. * **Type**: `int` * **Default**: `null` * **Importance**: Low * **Valid values / notes**: Use together with a non-empty proxy host and enter an actually valid port; there is no default proxy port. ### Send Retries #### `datadog.retry.max` Sets the Connector retry budget for consecutive send failures. * **Type**: `int` * **Default**: `5` * **Importance**: Low * **Valid values / notes**: Use a non-negative integer; `0` means no Connector retries. The budget applies to a failed `put` and resets after success; it is not a strict upper bound on the number of underlying network requests. The Task fails after retries are exhausted. #### `datadog.retry.backoff_ms` Sets the base wait time for send retries in milliseconds. * **Type**: `int` * **Default**: `3000` * **Importance**: Low * **Valid values / notes**: Use a positive integer; later waits use backoff and randomization, so this is not an exact resend timer or total retry deadline. It is different from the framework's `errors.retry.*` settings. ### Data Conversion and Transformations #### `key.converter` Specifies the converter for Kafka keys. * **Type**: `class` * **Default**: `null` * **Importance**: Low * **Valid values / notes**: When unset, inherits the Worker converter; although the key is not written to Datadog, the framework still converts it, so this setting must match the upstream encoding. #### `value.converter` Specifies the converter for Kafka values. * **Type**: `class` * **Default**: `null` * **Importance**: Low * **Valid values / notes**: When unset, inherits the Worker converter; it must match the upstream serialization format, and additional parameters belong to the selected converter. The converted value is then converted to JSON and placed in `message`; raw bytes are not sent directly to Datadog. #### `header.converter` Specifies the converter for Kafka Headers. * **Type**: `class` * **Default**: `null` * **Importance**: Low * **Valid values / notes**: When unset, inherits the Worker Header converter; check type compatibility when exposing Headers in logs. #### `transforms` Declares aliases for single message transformations (SMTs) to run in order. * **Type**: `list` * **Default**: Empty list * **Importance**: Low * **Valid values / notes**: Aliases must be unique, and each alias needs a corresponding transformation type; detailed transformation parameters are defined by the selected plugin. #### `transforms..type` Specifies the implementation class for an SMT alias. * **Type**: `class` * **Default**: None * **Importance**: High * **Required**: `required` when this SMT alias is configured * **Valid values / notes**: Replace `` with an alias from `transforms` and use an instantiable Transformation implementation class. #### `predicates` Declares aliases for predicates used to conditionally run SMTs. * **Type**: `list` * **Default**: Empty list * **Importance**: Low * **Valid values / notes**: Aliases must be unique; predicate-specific parameters are defined by each plugin. #### `predicates..type` Specifies the implementation class for a predicate alias. * **Type**: `class` * **Default**: None * **Importance**: High * **Required**: `required` when this predicate alias is configured * **Valid values / notes**: Replace `` with an alias from `predicates` and use an instantiable Predicate implementation class. #### `transforms..predicate` Selects the condition under which an SMT runs. * **Type**: `string` * **Default**: `null` * **Importance**: Medium * **Valid values / notes**: References a configured predicate alias; `` is the SMT alias. #### `transforms..negate` Controls whether the SMT predicate result is inverted. * **Type**: `boolean` * **Default**: `false` * **Importance**: Medium * **Valid values / notes**: When this property is explicitly set, the corresponding `predicate` must also be explicitly set, even when this property is `false`. ### External Credential Configuration Updates #### `config.action.reload` Controls how changes to a Config Provider value are handled. * **Type**: `string` * **Default**: `restart` * **Importance**: Low * **Valid values / notes**: `none` or `restart`; registering a Provider is a Worker configuration, and this property alone does not enable credential management or guarantee that arbitrary secret changes are detected automatically. ### Framework Error Handling #### `errors.retry.timeout` Sets the total retry duration for framework error handling in milliseconds. * **Type**: `long` * **Default**: `0` * **Importance**: Medium * **Valid values / notes**: `0` means no retries, and `-1` means unlimited retries; this applies to framework-supported stages such as conversion and SMT processing and cannot replace Datadog HTTP send retries. #### `errors.retry.delay.max.ms` Sets the maximum wait interval for framework retries. * **Type**: `long` * **Default**: `60000` * **Importance**: Medium * **Valid values / notes**: This is not `datadog.retry.backoff_ms` and does not control the plugin's HTTP send backoff. #### `errors.tolerance` Controls whether record-processing errors supported by the framework may be skipped. * **Type**: `string` * **Default**: `none` * **Importance**: Medium * **Valid values / notes**: `none` or `all`; `all` can tolerate supported conversion or SMT errors, but does not automatically skip HTTP send or log serialization failures inside `put`. A skipped record is not sent to Datadog. #### `errors.log.enable` Enables logging of framework-failed records. * **Type**: `boolean` * **Default**: `false` * **Importance**: Medium * **Valid values / notes**: `true` or `false`; independent of the Connector's own exception logging and does not change failure handling. #### `errors.log.include.messages` Controls whether framework error logs include detailed record context. * **Type**: `boolean` * **Default**: `false` * **Importance**: Medium * **Valid values / notes**: Takes effect when framework error logging is enabled; Sink context includes the Topic, partition, Offset, and timestamp, so restrict log access. #### `errors.deadletterqueue.topic.name` Sets the framework dead-letter queue (DLQ) Topic. * **Type**: `string` * **Default**: Empty string * **Importance**: Medium * **Valid values / notes**: An empty value disables it; together with `errors.tolerance=all`, it can retain failed records within the framework's supported scope. The Topic must not be subscribed to by this Connector; HTTP send failures inside the plugin's `put` are not captured automatically. #### `errors.deadletterqueue.topic.replication.factor` Sets the replication factor when a DLQ Topic is created automatically. * **Type**: `short` * **Default**: `3` * **Importance**: Medium * **Valid values / notes**: Used when the DLQ Topic does not exist and must be created; it should be compatible with the number of available Brokers in the Kafka cluster. #### `errors.deadletterqueue.context.headers.enable` Adds error context Headers to DLQ records. * **Type**: `boolean` * **Default**: `false` * **Importance**: Medium * **Valid values / notes**: When `true`, adds `__connect.errors.*` context; DLQ must already be enabled, and this does not expand the range of errors it can capture. ## Best Practices ### Add Consistent Search Tags During Onboarding **Applicable business scenario**: After logs can be delivered, you need to distinguish services and environments within the same Datadog organization for later search and operations. Configure log streams that share the same service identity first, then connect more sources to avoid labeling mixed services as one service. **Configuration example**: Add the following settings to the Quick Start configuration; the placeholders represent the service name and deployment environment for this log stream. ```properties theme={null} datadog.service= datadog.tags=env: ``` **Key notes**: These values are applied uniformly to the outer log metadata for this Connector, while the Topic tag is still added automatically. Use separate Connectors for different services; do not expect a `service` field inside the value to override the configured value. Common tags should not contain secrets or personal information. ### Allow Recovery Time for Transient Send Failures During Operation **Applicable business scenario**: A transient intake-side failure occurs during operation, and you need to allow retries and observe backlog instead of stopping the Task immediately. Before increasing the retry budget, confirm that the failure is not caused by an invalid key, an incorrect Site, or a data problem that reproduces consistently. **Configuration example**: Add the following settings to the Quick Start configuration to override the Connector's default retry budget and base wait time. ```properties theme={null} datadog.retry.max=8 datadog.retry.backoff_ms=5000 ``` **Key notes**: The example values are not universally optimal; adjust them according to acceptable backlog and recovery time. Synchronous sending and waiting create backpressure. The budget resets after success, and after it is exhausted you must fix the cause and recover the Task. Retries may resend logs that were already received, and the framework DLQ cannot replace this failure recovery; do not treat duplicate logs as exactly-once writes. ### Increase Consumption Parallelism by Partition When Scaling **Applicable business scenario**: After continuous operation, consumption backlog appears, the input Topic has multiple partitions, and the target still has intake capacity. You need to add consumption Tasks to share the processing load. Observe latency and errors first, then scale gradually to avoid turning a downstream bottleneck into more failed requests. **Configuration example**: For an input with at least two assignable partitions, set the Task limit from the Quick Start configuration to the following value. ```properties theme={null} tasks.max=2 ``` **Key notes**: Connect manages task assignment, and effective parallelism is limited by the number of partitions. This value is only a starting point for scaling and does not promise linear speedup or global ordering. After increasing the task count, also monitor Datadog intake errors, throughput, and consumption backlog. ## Monitoring ### What to Monitor Monitor Kafka Connect cluster health, Connector and Task status, input and processing throughput, consumption backlog and end-to-end latency, Offset commit progress and failures, error and retry frequency, and Worker JVM memory, GC, and thread signals. Monitor DLQ activity only when the corresponding error handling is enabled. When backlog persists or retries are frequent, inspect task exceptions together with authentication, Site, and downstream responses; do not determine that logs have been indexed based only on `RUNNING`, input record counts, or HTTP success. ### 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 Connect metrics are collected by a Prometheus-compatible data source and that labels such as cluster and instance match the dashboard queries, then import the JSON into Grafana and select the corresponding data source. ## Limitations * Exactly-once writes and idempotent deduplication are not provided; retries after partial request success, lost responses, and restarts before the send Offset is committed may produce duplicate logs. * Records whose value is `null` are skipped and do not result in a delete request to Datadog, so the number of consumed Kafka records is not equal to the number of logs actually sent. * Global ordering across Topics, partitions, or Tasks is not guaranteed, and Datadog display order is not guaranteed. * HTTP success is the send-acceptance boundary and does not guarantee that each log has been persisted, indexed, or made queryable in Datadog; `flush` also does not check Datadog downstream status. * Logs remain subject to Datadog API and organization intake policies. The Connector splits batches at an internal uncompressed JSON threshold of about 4.5 MB; a serialized log that exceeds this threshold and cannot fit into an empty batch is skipped. Automatic batching does not mean that a single log of any size will be accepted by Datadog. See the [Logs API](https://docs.datadoghq.com/api/latest/logs/) for the applicable constraints. ## FAQ ### The Task is RUNNING, but no logs can be found in Datadog First check whether the input Topic contains non-null values, whether consumption progress is advancing, and whether the Task log contains send or conversion errors. Confirm that the API key and Site belong to the target organization, use an appropriate time range, and search with `source:kafka-connect` and the corresponding Topic tag. If HTTP has been accepted but the log is still not searchable, continue checking Datadog log processing, indexing, and exclusion policies; do not treat HTTP success as proof of query visibility. ### HTTP sends keep failing, and the Connector still stops after a DLQ is configured The framework DLQ mainly handles conversion and SMT errors within its supported scope; it does not automatically receive HTTP failures inside the plugin's `put`. Check the complete exception and safe response context. Fix the API key and Site for authentication errors, and allow Connector retries for transient intake failures. After persistent failure or exhausted retries, fix the cause and recover the Task. This plugin does not distinguish permanent and transient failures by HTTP status and does not schedule retries according to `Retry-After`. ### Duplicate logs appear, or some logs are sent again after a restart Log intake and Kafka consumer Offset commits are not one transaction. A process exit after a successful send but before the commit, or a later request failure while processing a batch, may cause a resend. Check retries, Task restarts, and Offset commit errors, and include duplicates in downstream processing design; do not interpret a larger retry budget as a no-duplicate guarantee. ### The original log is JSON, but its fields are still inside `message` The Quick Start uses a string converter, so JSON text remains a string rather than an automatically parsed object. When a structured Connect value is required, choose a converter and its parameters that match the actual upstream serialization format, then confirm the output structure. Even when the value is already an object, its fields remain inside `message` and are not automatically converted into outer `service` or `hostname` fields. ### A URL error appears after configuring the intake address Check whether `datadog.url` was mistakenly set to a complete HTTPS address or to the `/api/v2/logs` path. This property accepts only a host and optional port; the Connector adds the protocol and path. When non-empty, it also overrides `datadog.site`. For normal Site-based intake, use the Site setting and remove an unnecessary URL override. ### Business content appears in troubleshooting logs. How can exposure risk be controlled? Do not leave TRACE enabled for long periods in production. Request bodies, responses, and previews of some oversized logs may enter runtime logs. Restrict log level, access permissions, and retention time; avoid exposing sensitive Headers to Datadog and confirm that a custom intake host is trusted. Redact shared diagnostic material, while retaining exception stacks and safe Topic, partition, and Offset context for troubleshooting.