> ## 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. # Adobe Experience Platform Sink Connector > Configure and operate the Adobe Experience Platform Sink Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting. ## Overview Adobe Experience Platform Sink Connector consumes JSON messages from Kafka Topics and sends them in batches to Adobe Experience Platform (AEP) through an HTTP streaming ingestion inlet. It sits between event producers and AEP dataflows, bringing data such as customer activities and application events into configured XDM Schemas and Datasets for subsequent analytics or customer profile processing. Each Kafka message value corresponds to one AEP message. The Connector groups these messages into a `messages` array for delivery. Target resources and field mappings depend on the AEP dataflow configuration and message content. The Connector does not automatically create Datasets, generate mappings, or add AEP message headers, nor does it automatically include Kafka message keys, partitions, Offsets, or Headers in business data. ## Prerequisites * Prepare the target XDM Schema and Dataset, along with the HTTP API Source base connection, source connection, target connection, and Dataflow, in an AEP project, organization, and Sandbox for which you have permission. Obtain the `inletUrl` returned by the base connection and the corresponding `flowId`. If RAW data requires transformation, configure a Data Prep Mapping in AEP and associate it with the Dataflow; XDM data that already conforms to the target Schema can use a path without transformation. * Check the `authenticationRequired` policy of the AEP base connection. An inlet that requires authentication must have a valid, authorized Adobe identity integration. Use an authenticated inlet for data containing personally identifiable information. The Bearer Token, API Key, IMS organization, and Sandbox request headers for resource management APIs are used to manage AEP resources; do not copy them directly as a set of Connector settings. * Upstream messages must have the AEP message structure and field types required by the selected inlet and dataflow. Wait approximately five minutes after creating a Dataflow before sending data, and account for brief ingestion pauses when creating or updating a dataflow. ## License Licensed under Apache License 2.0. ## Quick Start Prepare a Connect Cluster, Kafka, and the AEP resources described above, and confirm network connectivity and access permissions. For creation and management operations, see [Manage Connectors](../manage-connectors). The following minimal configuration applies to a pre-created inlet with `authenticationRequired=false` and a Topic whose message keys are UTF-8 strings or null and whose message values are plain JSON objects. It does not apply to inlets that require authentication or to data containing personally identifiable information. ```properties theme={null} name=adobe-experience-platform-sink connector.class=com.adobe.platform.streaming.sink.impl.AEPSinkConnector topics= key.converter=org.apache.kafka.connect.storage.StringConverter value.converter=org.apache.kafka.connect.json.JsonConverter value.converter.schemas.enable=false aep.endpoint= ``` Replace the Topic and inlet address with your actual resources. Set `aep.endpoint` to the HTTPS `inletUrl` returned by AEP, with a path of `/collection/`. Do not use an address that already contains `/collection/batch/`; the Connector converts it to the batch inlet itself. There is no need to explicitly set the default single Task, batch thresholds, or disabled authentication settings. `value.converter.schemas.enable=false` removes the need for the Connect `schema` / `payload` envelope, but does not remove the AEP message structure requirements. The following is a single message value written to Kafka by an upstream producer, illustrating the XDM message structure for HTTP API Source. It is not Connector configuration or the outer `messages` array of the final HTTP request. Replace the Schema, Dataflow, Dataset, and business fields before use; `xdmEntity` must conform to the actual target Schema. ```json theme={null} { "header": { "schemaRef": { "id": "", "contentType": "application/vnd.adobe.xed-full-notext+json; version=1.0" }, "flowId": "", "datasetId": "" }, "body": { "xdmMeta": { "schemaRef": { "id": "", "contentType": "application/vnd.adobe.xed-full-notext+json; version=1.0" } }, "xdmEntity": { "_id": "customer-event-001", "person": { "name": { "firstName": "Example", "lastName": "Customer" } } } } } ``` Here, `header` is a JSON message header, not HTTP Headers; `flowId`, `datasetId`, `schemaRef`, and `body` are not Connector parameters. The Connector does not populate these fields. Do not wrap each Kafka message in an additional `messages` layer beforehand. Other AEP ingestion paths may require different message metadata, such as `imsOrgId` in batch messages. Follow the requirements of the actual inlet rather than mixing examples from different paths. After applying the configuration, send a small amount of Schema-compliant data first. Check Connect runtime status, ingestion responses, AEP validation results, and Dataset ingestion results separately. A successful HTTP response indicates only acceptance at the ingestion layer; it does not prove successful ingestion into the final Dataset. **Security warning**: The Connector and Tasks output raw configuration in INFO logs at startup, including any Client Secret, authorization code, proxy password, or custom Authorization request header that may be present. Disabling DEBUG alone or keeping `errors.log.include.messages=false` does not resolve this risk. Do not assume that Config Provider substitution automatically redacts values. Before supplying production credentials, implement startup log redaction or suppression and log access controls. Do not enable DEBUG for sensitive data or include credentials and message content in support tickets or shared logs. ## Configuration ### Instance and Input Subscription #### `name` The Connector instance name. * **Type**: `string` * **Default**: None * **Importance**: High * **Valid Values / Notes**: Required; must be non-empty and contain no ISO control characters. Use a unique name. #### `connector.class` Selects the AEP Sink Connector. * **Type**: `string` * **Default**: None * **Importance**: High * **Valid Values / Notes**: Required; use `com.adobe.platform.streaming.sink.impl.AEPSinkConnector`. #### `topics` Specifies the Kafka Topics to consume. * **Type**: `list` * **Default**: Empty list * **Importance**: High * **Valid Values / Notes**: Comma-separated Topic names. Exactly one of `topics` and `topics.regex` must be non-empty. Must not include the DLQ Topic. #### `topics.regex` Subscribes to Kafka Topics using a regular expression. * **Type**: `string` * **Default**: Empty string * **Importance**: High * **Valid Values / Notes**: A valid Java regular expression; mutually exclusive with non-empty `topics`. Must not match the DLQ Topic. #### `tasks.max` The maximum number of Tasks that may be created. * **Type**: `int` * **Default**: `1` * **Importance**: High * **Valid Values / Notes**: At least `1`. Actual parallelism is limited by input partition assignment. All Tasks use the same inlet, with no global ordering across Tasks. ### Message Conversion #### `key.converter` Deserializes Kafka message keys. * **Type**: `class` * **Default**: `null` * **Importance**: Low * **Valid Values / Notes**: Uses the Worker Converter when omitted. Must match the upstream key encoding. The Quick Start uses `org.apache.kafka.connect.storage.StringConverter` for UTF-8 string keys; message keys are not sent to AEP. #### `value.converter` Deserializes Kafka message values. * **Type**: `class` * **Default**: `null` * **Importance**: Low * **Valid Values / Notes**: Uses the Worker Converter when omitted. The Quick Start explicitly uses `org.apache.kafka.connect.json.JsonConverter`; this is not the default class. Input must meet AEP message requirements, not merely be syntactically valid JSON. #### `value.converter.schemas.enable` Controls whether JsonConverter uses the Connect Schema envelope. * **Type**: `boolean` * **Default**: `true` * **Importance**: High * **Valid Values / Notes**: Applies only to JsonConverter. Use `false` for plain JSON objects; `true` requires the Connect `schema` / `payload` format. This setting does not control AEP XDM Schema validation. ### Target Inlet #### `aep.endpoint` Specifies the AEP streaming ingestion inlet. * **Type**: `string` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: Required, with no runtime fallback; must not be empty. Specify the full HTTPS inlet URL using the original `/collection/` path. This path is replaced with `/collection/batch/`; an already converted address will be rewritten again. #### `aep.connection.endpoint.headers` Adds HTTP Headers to ingestion requests. * **Type**: `string` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: JSON object text mapping header names to string values. Uses an empty map at runtime when missing or an empty string; these headers are not added if JSON parsing fails. This differs from `header` in message values. Authorization information in this setting is at risk of disclosure through raw configuration logs. ### Batch Delivery #### `aep.flush.interval.seconds` The elapsed-time threshold for triggering delivery within one consumption batch, in seconds. * **Type**: `int` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: Uses `1` at runtime when missing or blank, multiplied by `1000` using Java `int` arithmetic. If the multiplication result is less than `1`, it falls back to `1000` ms. Malformed integer text or text outside the `int` range throws `NumberFormatException` rather than falling back. There is no ConfigDef range validation; a positive integer is recommended, with multiplication overflow avoided. This is not a background timer. Remaining messages are still sent at the end of a consumption batch; they are not held across consumption batches to fill a batch. #### `aep.flush.bytes.kb` The accumulated-message-size threshold for triggering delivery within one consumption batch, in KB. * **Type**: `int` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: Uses `4` at runtime when missing or blank, multiplied by `1024` using Java `int` arithmetic. If the multiplication result is less than `1`, it falls back to `4096`. Malformed integer text or text outside the `int` range throws `NumberFormatException` rather than falling back. There is no ConfigDef range validation; a positive integer is recommended, with multiplication overflow avoided. The accumulated size is actually Java string length, not UTF-8 byte count, and excludes the final request envelope overhead. The threshold is checked only after a record is added; it cannot ensure compliance with Adobe's 1 MB request size limit. ### HTTP Connections and Retries #### `aep.connection.timeout` The connection timeout for ingestion HTTP requests, in milliseconds. * **Type**: `int` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: Uses `5000` at runtime when missing or blank; use a positive integer. Controls only ingestion connections, not Token request timeouts. #### `aep.connection.readTimeout` The read timeout for ingestion HTTP requests, in milliseconds. * **Type**: `int` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: Uses `60000` at runtime when missing or blank; use a positive integer and preserve the parameter name's capitalization. Does not apply to Token requests. #### `aep.connection.maxRetries` Limits the total number of attempts for one ingestion HTTP call. * **Type**: `int` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: Uses `3` at runtime when missing or blank, meaning at most three total attempts, not an initial request plus three retries. There is no parameter range validation. Use a positive value to send requests; do not set zero or a negative value. Resends the same request body at fixed intervals for 5xx responses and I/O exceptions. Other non-2xx responses, such as 429, do not take this retry path, nor do response content parsing failures. Resending may cause duplicates. #### `aep.connection.retryBackoff` The fixed wait after a failed ingestion HTTP attempt, in milliseconds. * **Type**: `int` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: Uses `300` at runtime when missing or blank; use a non-negative integer. A wait may also occur after the final 5xx or I/O failure. Waiting blocks the Task; this is not an authentication retry or adaptive rate limiting setting. ### Authentication #### `aep.connection.auth.enabled` Enables the Connector's Bearer Token acquisition and attachment functionality. * **Type**: `string` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: Uses the string `false` at runtime when missing. Only the exact lowercase value `true` enables authentication; `TRUE` or values containing whitespace do not. This setting cannot change the AEP inlet's authentication policy. #### `aep.connection.auth.token.type` Selects the Adobe Token acquisition method. * **Type**: `string` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: Uses `jwt_token` at runtime when authentication is enabled and this setting is absent. Supports the exact values `access_token`, `jwt_token`, and `oauth2_access_token`; empty or unknown values fail. The JWT authentication path is deprecated. When migrating, explicitly select OAuth2 according to Adobe identity integration requirements and confirm authorization and scope compatibility. The presence of an option does not mean that the server still supports legacy JWT. #### `aep.connection.auth.client.id` Specifies the Client ID of the Adobe identity integration. * **Type**: `string` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: No runtime fallback; required for all three authentication methods when enabled. It is not an Access Token. #### `aep.connection.auth.client.secret` Specifies the Client Secret of the Adobe identity integration. * **Type**: `string` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: No runtime fallback; required for all three authentication methods when enabled. This setting has no automatic PASSWORD-type redaction protection. Resolve the risk of disclosure through raw configuration logs before supplying it. #### `aep.connection.auth.endpoint` Specifies the Adobe IMS base URL. * **Type**: `string` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: No runtime fallback; required when authentication is enabled. Do not omit it in reliance on environment variables. The three methods append `/ims/token/v1`, `/ims/exchange/jwt/`, and `/ims/token/v3`, respectively. Do not append these paths beforehand. #### `aep.connection.auth.client.code` Specifies the authorization code required for IMS authorization code exchange. * **Type**: `string` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: No runtime fallback; required only for the `access_token` method and unused by the other methods. It is not a pre-issued Bearer Token. It is a sensitive credential that may appear in raw configuration logs. ### Legacy JWT Authentication #### `aep.connection.auth.imsOrg` Specifies the IMS organization ID for JWT authentication. * **Type**: `string` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: No runtime fallback; required only for the legacy `jwt_token` method and unused by other authentication methods. This authentication path is deprecated; the replacement is `oauth2_access_token` configured according to Adobe identity integration requirements. #### `aep.connection.auth.accountKey` Specifies the technical account identifier for JWT authentication. * **Type**: `string` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: No runtime fallback; required only for the legacy `jwt_token` method. It is not an RSA private key. This authentication path is deprecated. Other methods do not use this setting; migrate to the corresponding OAuth2 Client settings. #### `aep.connection.auth.filePath` Specifies the path to the RSA private key file used for JWT authentication. * **Type**: `string` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: No runtime fallback; required only for the legacy `jwt_token` method. Every Worker running a Task must be able to read this local file. Use a PEM-encoded PKCS#8 private key, with a file size no greater than 1 MiB. Do not put the private key content in configuration. This authentication path is deprecated; OAuth2 does not use this setting. ### HTTP Proxy #### `aep.connection.proxy.host` Specifies the proxy host for ingestion requests and authentication providers. * **Type**: `string` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: Uses `null` at runtime when missing or blank, leaving the proxy host unspecified. Proxy behavior is handled by the HTTP connection. #### `aep.connection.proxy.port` Specifies the proxy port. * **Type**: `int` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: Uses `443` at runtime when missing or blank. Specify a valid TCP port when configuring a proxy host. There is no Connector-level range validation. #### `aep.connection.proxy.user` Specifies the username for a proxy that requires authentication. * **Type**: `string` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: Uses `null` at runtime when missing or blank. Use it with the password as required by the proxy. Proxy authentication uses the JVM-global default Authenticator; consider the impact on other connections in the same Worker. #### `aep.connection.proxy.password` Specifies the proxy authentication password. * **Type**: `string` (type read at runtime) * **Default**: No ConfigDef default * **Importance**: Not declared * **Valid Values / Notes**: Uses `null` at runtime when missing or blank. There is no automatic PASSWORD redaction protection. Both raw configuration logs at startup and proxy DEBUG logs pose a risk of credential exposure. ### Error Handling #### `errors.tolerance` Controls the tolerance policy for record errors that the framework can handle. * **Type**: `string` * **Default**: `none` * **Importance**: Medium * **Valid Values / Notes**: `none` or `all`. `all` does not mean all remote HTTP errors are recoverable, nor can it prevent unreported record drops inside the Connector. 401/403 may still cause Task failure. #### `errors.deadletterqueue.topic.name` Specifies the DLQ Topic for Sink error records. * **Type**: `string` * **Default**: Empty string * **Importance**: Medium * **Valid Values / Notes**: An empty value disables the DLQ. A non-empty DLQ Topic must be excluded from the input subscription. Only records entering the framework's error reporting path may be written to it; do not treat it as a backup of all failed AEP records. #### `errors.deadletterqueue.topic.replication.factor` Sets the replication factor when automatically creating a DLQ Topic. * **Type**: `short` * **Default**: `3` * **Importance**: Medium * **Valid Values / Notes**: For a DLQ Topic that does not yet exist, match the number of replicas supported by the Kafka deployment. Does not change the replication factor of an existing Topic. #### `errors.deadletterqueue.context.headers.enable` Adds error context headers to DLQ records. * **Type**: `boolean` * **Default**: `false` * **Importance**: Medium * **Valid Values / Notes**: Adds Headers with the `__connect.errors.` prefix when enabled. Assess the risk of exposing context information. These Headers are not part of the message headers sent to AEP. #### `errors.log.enable` Controls framework error logging. * **Type**: `boolean` * **Default**: `false` * **Importance**: Medium * **Valid Values / Notes**: Independent of the Connector log level. Disabling it does not prevent Connector startup logs from outputting raw configuration. #### `errors.log.include.messages` Controls whether framework error logs include message details. * **Type**: `boolean` * **Default**: `false` * **Importance**: Medium * **Valid Values / Notes**: Keep disabled for sensitive input. This is not a redaction mechanism for Adobe configuration or proxy credentials. ## Best Practices ### Increase Consumption Parallelism When Backlog Grows **Applicable Scenario**: Integration is complete and data has been confirmed to reach the Dataset, but a single Task's consumption backlog keeps growing during ongoing operation. You want to send in parallel using multiple input Topic partitions while keeping the message format and target inlet unchanged. **Configuration Example**: Add the following setting to the Quick Start configuration. Here, `2` is an example starting point for scaling, not a universally optimal value. The input Topic should have enough assignable partitions. ```properties theme={null} tasks.max=2 ``` **Key Points**: First check whether AEP rate limiting, request failures, or ingestion validation errors are causing the backlog; blindly adding Tasks is not a solution to these conditions. Each Task sends independently to the same inlet. Adding Tasks increases concurrent HTTP requests but does not provide global ordering across partitions. After changing the setting, compare consumption backlog, request latency, and AEP ingestion and Dataset results. Reduce parallelism if target-side pressure or errors increase. Raising the limit will not provide corresponding parallelism if there are too few input partitions. ## Monitoring ### What to Monitor Monitor Kafka Connect cluster health, Connector / Task status, consumption throughput and backlog, end-to-end latency, Offset commits, errors and retries, and Worker JVM memory, GC, and thread signals. Monitor DLQ activity only when the relevant error handling is configured. RUNNING, HTTP 2xx, and committed Offsets do not indicate successful Dataset ingestion. Independently observe AEP Dataflows, Dataset batch status, and streaming validation results, checking ingestion-layer acceptance counts separately from final Dataset ingestion results. ### Import the Grafana Dashboard Download the shared [Kafka Connect Grafana Dashboard](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json). Confirm that Connect / Worker metrics are collected and that a compatible data source and the cluster, Connector, Task, and other labels required by the dashboard are configured. Then import the JSON into Grafana and select the corresponding data source. ## Limitations * The Connector does not automatically create AEP resources, perform Data Prep mappings, or populate the XDM message structure. It also does not provide routing to different target inlets by Topic. * It does not provide CDC deletes, update merging, or Schema evolution management. Kafka null values and tombstones do not represent AEP delete operations and may trigger data conversion exceptions. * Raw JSON strings that fail to parse may be excluded directly without entering the error reporting path. Some target-side errors may also be logged while Offsets continue to advance. Do not interpret Offset advancement as complete delivery. * It does not provide transactional writes, idempotency keys, or automatic deduplication. HTTP retries, restarts before Offset commits, and rebalances may produce duplicates. It guarantees neither unconditional at-least-once delivery to AEP nor exactly-once delivery. * Batch thresholds do not equal the final HTTP request size in bytes and cannot automatically ensure compliance with Adobe's 1 MB request limit. A single large message may also exceed the threshold. * It does not guarantee global event ordering or final AEP ingestion ordering. Tasks and partitions may be processed in parallel. ## FAQ ### Why Is There Still No Data in the Dataset When the Task Shows RUNNING? Startup status is not an inlet connectivity check, and HTTP acceptance does not mean Dataset ingestion. Verify that the inlet address, Dataflow status, and the message's `flowId` and `datasetId` correspond to each other. Then check the target Schema, XDM field types, AEP streaming validation, and Dataset batch errors. Wait approximately five minutes after creating a Dataflow before sending data. Do not judge success solely by Offset advancement; use identifiable business events to verify final Dataset ingestion results. ### Why Does the Task Not Recover Automatically After a 401 or 403? Check whether the AEP inlet requires authentication, whether the Token is valid, and whether the identity integration has the required permissions. To enable Connector authentication, use the exact lowercase value `true` and supply the parameters required by the selected method. Do not rely on implicit selection of legacy JWT. OAuth2 caches and refreshes Tokens according to their expiration time, but a 401 does not automatically clear the cache, obtain a new Token, and resend the request. Resolve authorization issues before resuming the Task. First ensure that logs will not expose credentials during troubleshooting. ### Why Can Some Messages Still Fail When HTTP Returns Success? The 207 response to a batch request can contain per-message results; do not check only the HTTP status code. The Connector identifies an entry as failed when the `status` field exists and its value is not `null`, without requiring a non-empty string, and obtains the record index from the content after the last hyphen in `xactionId`. Adobe's public examples instead use `statusCode`, may separate success identifiers with colons, and may omit `xactionId` for failed items. This is a static contract difference between the examples and the plugin's parsing rules, not confirmation of an actual cloud failure. Before going live, verify the responses from the actual inlet. In particular, do not rely on the plugin to identify failed items containing only `statusCode`; it also does not automatically retry individual failed messages. Check results against AEP validation and Dataset status. If necessary, contact the maintainers to confirm response compatibility, and avoid exposing business data when obtaining responses. ### Why Does the DLQ Not Contain All Failed Messages? The DLQ covers only records passed to the framework's error reporting path, not all internal parsing drops or all remote response problems. Enabling error tolerance also does not resolve validation failures after AEP ingestion. Check the error tolerance policy and whether the input subscription excludes the DLQ. Confirm that the DLQ Topic exists, or that the permissions and replication factor requirements for automatic creation are met. Correlate Task logs with AEP results. Do not treat the absence of DLQ records as proof that all messages succeeded, or equate record reporting with completed synchronous persistence. ### Why Do Duplicates Appear After Recovery or Restart? AEP may already have accepted a request when the connection is interrupted or before the Kafka Offset is committed. Subsequent HTTP retries or resumed consumption can resend it. The Connector does not deduplicate automatically. Identify duplicates using business event identifiers, and separately design and confirm deduplication rules in the target data processing pipeline. Do not assume that the XDM `_id` field alone gives this Connector exactly-once semantics.