Datadog Logs Sink Connector - AutoMQ Documentation

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 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; 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 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 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.

connector.class=com.datadoghq.connect.logs.DatadogLogsSinkConnector
topics=<logs-topic>
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
datadog.api_key=<datadog-api-key>
datadog.site=<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:<logs-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:<topic-name> 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.<alias>.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 <alias> 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.<alias>.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 <alias> with an alias from predicates and use an instantiable Predicate implementation class.

`transforms.<alias>.predicate`

Selects the condition under which an SMT runs.

Type: string
Default: null
Importance: Medium
Valid values / notes: References a configured predicate alias; <alias> is the SMT alias.

`transforms.<alias>.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.

datadog.service=<service-name>
datadog.tags=env:<environment>

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.

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.

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, 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 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.

​Overview

​Prerequisites

​License

​Quick Start

​Configuration

​Connector and Subscription

​connector.class

​topics

​topics.regex

​tasks.max

​tasks.max.enforce

​Datadog Authentication and Target Site

​datadog.api_key

​datadog.site

​datadog.url

​Log Metadata

​datadog.tags

​datadog.service

​datadog.hostname

​datadog.add_published_date

​datadog.parse_record_headers

​Proxy

​datadog.proxy.url

​datadog.proxy.port

​Send Retries

​datadog.retry.max

​datadog.retry.backoff_ms

​Data Conversion and Transformations

​key.converter

​value.converter

​header.converter

​transforms

​transforms.<alias>.type

​predicates

​predicates.<alias>.type

​transforms.<alias>.predicate

​transforms.<alias>.negate

​External Credential Configuration Updates

​config.action.reload

​Framework Error Handling

​errors.retry.timeout

​errors.retry.delay.max.ms

​errors.tolerance

​errors.log.enable

​errors.log.include.messages

​errors.deadletterqueue.topic.name

​errors.deadletterqueue.topic.replication.factor

​errors.deadletterqueue.context.headers.enable

​Best Practices

​Add Consistent Search Tags During Onboarding

​Allow Recovery Time for Transient Send Failures During Operation

​Increase Consumption Parallelism by Partition When Scaling

​Monitoring

​What to Monitor

​Import the Grafana Dashboard

​Limitations

​FAQ

​The Task is RUNNING, but no logs can be found in Datadog

​HTTP sends keep failing, and the Connector still stops after a DLQ is configured

​Duplicate logs appear, or some logs are sent again after a restart

​The original log is JSON, but its fields are still inside message

​A URL error appears after configuring the intake address

​Business content appears in troubleshooting logs. How can exposure risk be controlled?