# HTTP Sink Connector for Confluent Cloud The fully managed HTTP Sink connector for Confluent Cloud sends Apache Kafka® records to an HTTP or HTTPS API endpoint. The connector consumes records from Kafka topics and converts each record value to STRING or JSON format before sending it, in the request body, to the configured `http.api.url`. The API URL can reference a record key or topic name using substitution variables `${topic}` and `${key}` in the URL property. You can also use [fields from the Kafka record](#cc-http-sink-template-parameters). The targeted API must support either a `POST`, `PATCH`, or `PUT` request. The connector batches records up to the set **Batch max size** (`batch.max.size`) before sending the batched request to the API. Each record is converted to its String representation or its JSON representation with **Request Body Format** (`request.body.format=json`) and then separated with the **Batch separator** (`batch.separator`). See [Configuration Properties](#cc-http-sink-config-properties) for configuration property descriptions. The HTTP Sink connector supports connecting to APIs using SSL along with Basic Authentication, OAuth2, or a Proxy Authentication Server. #### NOTE * This Quick Start is for the fully managed Confluent Cloud connector. If you are installing the connector locally for Confluent Platform, see [HTTP Sink Connector for Confluent Platform](https://docs.confluent.io/kafka-connectors/http/current/). * If you require private networking for fully managed connectors, make sure to set up the proper networking beforehand. For more information, see [Manage Networking for Confluent Cloud Connectors](networking/internet-resource.md#clusters-connect-cloud). ## Features The HTTP Sink connector supports the following features: * **At least once delivery**: This connector guarantees that records from the Kafka topic are delivered at least once. * **Supports multiple tasks**: The connector supports running one or more tasks. More tasks may improve performance (that is, consumer lag is reduced with multiple tasks running). * **Automatically creates topics**: The following three topics are automatically created when the connector starts: - Success topic - Error topic - [Dead letter queue (DLQ) topic](dead-letter-queue.md#ccloud-dlq-topics) The suffix for each topic name is the connector’s logical ID. In the example below, there are the three connector topics and one pre-existing Kafka topic named pageviews. ![HTTP Sink Connector Topics](images/ccloud-datadog-metrics-sink-topics.png) If the records sent to the topic are not in the correct format, or if important fields are missing in the record, the errors are recorded in the error topic, and the connector continues to run. * **Supported data formats**: The connector supports Avro, JSON Schema (JSON-SR), Protobuf, JSON (schemaless), and Bytes formats. [Schema Registry](../get-started/schema-registry.md#cloud-sr-config) must be enabled to use a Schema Registry-based format (for example, Avro, JSON Schema, or Protobuf). * **Template parameters**: The connector allows you to specify [fields from the Kafka record](#cc-http-sink-template-parameters), other than `{$topic}` and `{$key}` and constructs a unique URL using these parameters. * **Regex Replacements**: The connector can take a number of regex patterns and replacement strings that are applied to a record before it is submitted to the destination API. To do this, the connector uses the configuration options `regex.patterns`, `regex.replacements`, and `regex.separator`. * **Supports Batching**: The connector batches requests submitted to HTTP APIs for efficiency. Batches can be built with the configuration options `batch.prefix`, `batch.suffix` and `batch.separator`. All regex options apply when batching and are applied to individual records before being submitted to the batch. For more information and examples to use with the Confluent Cloud API for Connect, see the [Confluent Cloud API for Connect Usage Examples](connect-api-section.md#ccloud-connect-api) section. ## Limitations Be sure to review the following information. * For connector limitations, see [HTTP Sink Connector](limits.md#http-sink-limits) limitations. * If you plan to use one or more Single Message Transformations (SMTs), see [SMT Limitations](single-message-transforms.md#cc-single-message-transforms-limitations). * The connector does not batch requests for messages containing Kafka header values that are different. ## Template Parameters The connector forwards the message (record) value to the HTTP API. You can add parameters to have the connector construct a unique HTTP API URL containing the record key and topic name. For example, you enter `http://eshost1:9200/api/messages/${topic}/${key}` to have the HTTP API URL contain the topic name and record key. In addition to the `${topic}` and `${key}` parameters, you can also refer to fields from the Kafka record. As shown in the following example, you may want the connector to construct a URL that uses the Order ID and Customer ID. The Avro format that the producer uses to generate records in the Apache Kafka® topic `order` is shown below: ```json { "name": "MyClass", "type": "record", "namespace": "com.acme.avro", "fields": [ { "name": "customerId", "type": "int" }, { "name": "order", "type": { "name": "order", "type": "record", "fields": [ { "name": "id", "type": "int" }, { "name": "amount", "type": "int" } ] } } ] } ``` To send the Order ID and Customer ID, you would use the following URL in the HTTP API URL (`http.api.url`) configuration property: ```properties "http.api.url" : "http://eshost1:9200/api/messages/order/${order.id}/customer/${customerId}/" ``` Assuming the data in the Kafka topic contains the following values: ```json { "customerId": 123, "order": { "id": 1, "amount": 12345 } } ``` The connector constructs the following URL: ```bash http://eshost1:9200/api/messages/order/1/customer/123/ ``` #### NOTE * The maximum depth for added parameters is 10. For example, connector validation fails if you were to use the URL `https://eshost1:9200/api/messages/order/${a.b.c.d.e.f.g.h.i.j.k}`. * When you add parameters to the HTTP API URL, each record can result in a unique URL. For this reason, batching is disabled when using additional URL parameters. * The connector throws a runtime exception if fields referred to in the HTTP API URL do not exist in the Kafka record. ## Quick Start Use this quick start to get up and running with the Confluent Cloud HTTP Sink connector. The quick start provides the basics of selecting the connector and configuring it to stream events to an HTTP endpoint. Prerequisites : - Authorized access to a [Confluent Cloud](https://www.confluent.io/confluent-cloud/) cluster on Amazon Web Services (AWS), Microsoft Azure (Azure), or Google Cloud. - The Confluent CLI installed and configured for the cluster. See [Install the Confluent CLI](https://docs.confluent.io/confluent-cli/current/install.html). - [Schema Registry](../get-started/schema-registry.md#cloud-sr-config) must be enabled to use a Schema Registry-based format (for example, Avro, JSON_SR (JSON Schema), or Protobuf). - At least one source Kafka topic must exist in your Confluent Cloud cluster before creating the sink connector. ### Using the Confluent Cloud Console #### Step 1: Launch your Confluent Cloud cluster To create and launch a Kafka cluster in Confluent Cloud, see [Create a kafka cluster in Confluent Cloud](../get-started/index.md#cloud-create-kafka-cluster). #### Step 2: Add a connector In the left navigation menu, click **Connectors**. If you already have connectors in your cluster, click **+ Add connector**. #### Step 3: Select your connector Click the **HTTP Sink** connector card. ![HTTP Sink Connector Card](images/ccloud-http-sink-icon.png) #### Step 4: Enter the connector details #### NOTE * Ensure you have all your [prerequisites](#cc-http-sink-prereqs) completed. * An asterisk ( \* ) designates a required entry. At the **Add HTTP Sink Connector** screen, complete the following: ### Topic selection If you’ve already populated your Kafka topics, select the topics you want to connect from the **Topics** list. To create a new topic, click **+Add new topic**. ### Kafka access 1. Select the way you want to provide **Kafka Cluster credentials**. You can choose one of the following options: - **My account**: This setting allows your connector to globally access everything that you have access to. With a user account, the connector uses an API key and secret to access the Kafka cluster. This option is not recommended for production. - **Service account**: This setting limits the access for your connector by using a [service account](service-account.md#s3-cloud-service-account). This option is recommended for production. - **Use an existing API key**: This setting allows you to specify an API key and a secret pair. You can use an existing pair or create a new one. This method is not recommended for production environments. #### NOTE Freight clusters support only service accounts for Kafka authentication. 2. Click **Continue**. ### Authentication 1. Configure the authentication properties: - **HTTP URL**: Enter your **HTTP API URL**. Use an HTTP or HTTPS connection URL. For example, `http://eshost1:9200/api/messages` or `https://eshost3:9200/api/messages`. The connector sends the record value to the API as part of the request body. You can specify a static URL (for example, `http://eshost1:9200/api/messages`) or a dynamic URL (for example, `http://eshost1:9200/api/messages/${topic}/${key}`). You can also specify a dynamic URL using [fields from the Kafka record](#cc-http-sink-template-parameters). #### NOTE * Note that if the connection URL is HTTPS, HTTPS is used for all connections. A URL with no protocol is considered HTTP. * For additional information, see [HTTP Sink Connector](limits.md#http-sink-limits) limitations. - **Endpoint Authentication type**: Select an **Endpoint Authentication Type**. Defaults to `NONE`. - **BASIC**: The connector authenticates with a username and password. - **NONE** (the default): The endpoint requires no authentication. - **OAUTH2**: The connector authenticates using OAuth credentials. Supports [OAuth Client Credentials grant type](https://oauth.net/2/grant-types/) only. - **Auth username**: The username to be used with an endpoint requiring authentication. - **Auth password**: The password to be used with an endpoint requiring authentication. - **OAuth2 token URL**: The URL to be used for fetching the OAuth2 token in the field. **Client Credentials** is the only supported grant type. - **OAuth2 client ID**: The client ID used when fetching the OAuth2 token. - **OAuth2 secret**: The secret used when fetching the OAuth2 token. - **OAuth2 token property name**: The name of the property containing the OAuth2 token returned by the HTTP proxy. Defaults to `access_token`. - **OAuth2 auth mode mechanism**: Specifies how to encode `client_id` and `client_secret` in the OAuth2 authorization request. If set to `header`, the connector encodes credentials as an `Authorization: Basic ` HTTP header. If set to `url`, then `client_id` and `client_secret` are sent in body as URL encoded parameters. Defaults to `header`. - **OAuth2 scope**: The scope used when fetching OAuth2 token. If left empty, this parameter is not set in the authorization request. Defaults to `any`. - **Add JWT token**: Whether to generate and add a JSON Web Token (JWT) to a request. The signing algorithm used is the RS256 algorithm. If set to `true`, the JWT is added as a `jwt_token` request parameter. Defaults to `false`. When set to `true`, the following additional configuration properties appear: * **JWT keystore**: Click and upload the key store containing the private key used to sign the JWT. * **JWT keystore password**: Enter the password used to access the key store. * **JWT keystore type**: Sets the key store type to `JKS` or `PKCS12`. Defaults to `JKS`. * **JWT JSON claims**: Add any necessary [JWT claims](https://auth0.com/docs/secure/tokens/json-web-tokens/json-web-token-claims) as a JSON string. For example: ```json { "iss": "", "aud": "", "sub": "" } ``` #### NOTE The connector sets the JWT claim `iat` (issued at) to the JWT creation time (current time). The connector sets `exp` (expiry) to five minutes from time of issue. The claim `jti` is set to a random string. The connector overwrites these values if you add values for them here. - **JWT keystore**: Click and upload the key store containing the private key used to sign the JWT. - **JWT keystore password**: Enter the password used to access the key store. - **JWT keystore type**: Sets the key store type to `JKS` or `PKCS12`. Defaults to `JKS`. - **JWT JSON claims**: Add any necessary [JWT claims](https://auth0.com/docs/secure/tokens/json-web-tokens/json-web-token-claims) as a JSON string. - **Key password**: The password of the private key in the key store file. This is optional for client. - **Key Store**: The key store containing the server certificate. Only required if using HTTPS. - **Keystore Password**: The store password for the key store file. This is optional for a client and is only needed if `https.ssl.keystore.location` is configured. - **Trust store**: The trust store containing the server CA certificate. Only required if using HTTPS. - **Trust store password**: The trust store password containing the server CA certificate. Only required if using HTTPS. - **SSL Protocol**: The protocol to use for SSL connections. Defaults to `TLSv1.3`. - **Enable host verification**: Whether SSL host verification should be enabled. Defaults to `true`. 2. Click **Continue**. ### Configuration - **Input Kafka record value format**: Select the input Kafka record value format (data coming from the Kafka topic). Valid entires are AVRO, PROTOBUF, JSON_SR, JSON, or BYTES. A valid schema must be available in [Schema Registry](../get-started/schema-registry.md#cloud-sr-config) to use a schema-based message format (for example, Avro, JSON Schema, or Protobuf). ### **Show advanced configurations** - **Schema context**: Select a schema context to use for this connector, if using a schema-based data format. This property defaults to the **Default** context, which configures the connector to use the default schema set up for Schema Registry in your Confluent Cloud environment. A schema context allows you to use separate schemas (like schema sub-registries) tied to topics in different Kafka clusters that share the same Schema Registry environment. For example, if you select a non-default context, a **Source** connector uses only that schema context to register a schema and a **Sink** connector uses only that schema context to read from. For more information about setting up a schema context, see [What are schema contexts and when should you use them?](../sr/faqs-cc.md#faq-schema-contexts). - **Input Kafka record key format**: Sets the input Kafka record key format. Valid entries are BYTES or STRING. - **HTTP Request Method**: The request method to use. Options are `PUT`, `POST`, and `PATCH`. Defaults to `POST`. - **HTTP Headers**: HTTP headers to be included in all requests. Individual headers should be separated by the Header Separator. - **HTTP Headers Separator**: Separator character used in headers. - **Sensitive HTTP Headers**: Sensitive HTTP headers (for example, credentials) to be included in all requests. Individual headers should be separated by the Header Separator. - **OAuth2 Client Headers**: HTTP headers to be included in the OAuth2 client endpoint. Individual headers should be separated by OAuth2 Client Headers Separator. - **OAuth2 Client Headers Separator**: Separator character used in OAuth2 Client Headers. - **Behavior for null valued records**: How to handle records with a non-null key and a null value (that is, Kafka tombstone records). Valid options are `ignore`, `delete` and `fail`. Defaults to `ignore`. - **Behavior on errors**: Error handling behavior configuration for handling error responses from HTTP requests. Valid options are `ignore` and `fail`. Defaults to `ignore`. - **Report errors as**: Dictates the content of records produced to the error topic. If set to `error_string`, the value is a human readable string describing the failure. The error string includes the following information if available: HTTP response code, reason phrase, submitted payload, URL, response content, exception and error message. - **Retry on HTTP codes**: Comma-separated list of HTTP codes or range of codes to retry on. - **Maximum Retries**: The maximum number of times to retry on errors before failing the task. - **Retry Backoff (milliseconds)**: The time in milliseconds to wait following an error before a retry attempt is made. - **Connect timeout (milliseconds)**: The time in milliseconds to wait for a connection to be established. - **Request timeout (milliseconds)**: The time in milliseconds to wait for a request response from the server. - **Request Body Format**: Used to produce request body in either JSON or String format. - **Batch key pattern**: Pattern used to build the key for a given batch. - **Batch max size**: The number of records accumulated in a batch before the HTTP API is invoked. Note that batching is disabled if you use template [parameters](#cc-http-sink-template-parameters) other than `$key` and `$topic`. Defaults to `1`. - **Batch prefix**: Prefix added to record batches. This is applied once at the beginning of the batch of records. Only used when `request.body.format` is set to `string`. - **Batch suffix**: Suffix added to record batches. This is applied once at the end of the batch of records. Only used when `request.body.format` is set to `string`. - **Batch separator**: Separator for records in a batch. - **Batch json as array**: Whether or not to use an array to bundle JSON records. Only used when `request.body.format` is set to JSON. - **Regular expression patterns**: Regular expression patterns used for replacements in the message sent to the HTTP service. - **Regular expression replacements**: Regex replacements to use with the patterns in `regex.patterns`. - **Regular expression separator**: Separator character used in `regex.patterns` and `regex.replacements` property. - **Retry Backoff Policy**: The backoff policy to use in terms of retry - `CONSTANT_VALUE` or `EXPONENTIAL_WITH_JITTER`. **Additional Configs** - **value.converter.replace.null.with.default**: Whether to replace fields that have a default value and that are null to the default value. When set to true, the default value is used, otherwise null is used. Applicable for JSON Converter. - **Value Converter Schema ID Deserializer**: Sets the class name of the schema ID deserializer for values. The deserializer reads schema IDs from message headers. - **Value Converter Reference Subject Name Strategy**: Sets the subject reference name strategy for values. Valid entries are `DefaultReferenceSubjectNameStrategy` or `QualifiedReferenceSubjectNameStrategy`. You can use this strategy only with `PROTOBUF` format; the default strategy is `DefaultReferenceSubjectNameStrategy`. - **Schema ID For Value Converter**: Sets the schema ID to use for deserialization when using `ConfigSchemaIdDeserializer`. This lets you specify a fixed schema ID for deserializing message values. This property is applicable only when `value.converter.value.schema.id.deserializer` is set to `ConfigSchemaIdDeserializer`. - **value.converter.schemas.enable**: Include schemas within each of the serialized values. Input messages must contain schema and payload fields and may not contain additional fields. For plain JSON data, set this to false. Applicable for JSON Converter. - **errors.tolerance**: Use this property if you would like to configure the connector’s error handling behavior. WARNING: This property should be used with CAUTION for SOURCE CONNECTORS as it may lead to dataloss. If you set this property to ‘all’, the connector will not fail on errant records, but will instead log them (and send to DLQ for Sink Connectors) and continue processing. If you set this property to ‘none’, the connector task will fail on errant records. - **value.converter.ignore.default.for.nullables**: When set to true, this property ensures that the corresponding record in Kafka is NULL, instead of showing the default column value. Applicable for AVRO,PROTOBUF and JSON_SR Converters. - **Key Converter Schema ID Deserializer**: Sets the class name of the schema ID deserializer for keys. The deserializer reads schema IDs from message headers. - **Value Converter Decimal Format**: Specifies the `JSON` or `JSON_SR` serialization format for Connect `DECIMAL` logical type values with two allowed literals: `BASE64` to serialize `DECIMAL` logical types as base64 encoded binary data, and `NUMERIC` to serialize `DECIMAL` logical type values in `JSON` or `JSON_SR` as a number representing the decimal value. - **Schema GUID For Key Converter**: Sets the schema GUID to use for deserialization when using `ConfigSchemaIdDeserializer`. This lets you specify a fixed schema GUID for deserializing message keys. This property is applicable only when `key.converter.key.schema.id.deserializer` is set to `ConfigSchemaIdDeserializer`. - **Schema GUID For Value Converter**: Sets the schema GUID to use for deserialization when using `ConfigSchemaIdDeserializer`. This lets you specify a fixed schema GUID for deserializing message values. This property is applicable only when `value.converter.value.schema.id.deserializer` is set to `ConfigSchemaIdDeserializer`. - **Value Converter Connect Meta Data**: Enables the Connect converter to add its metadata to the output schema. Applies to Avro converters. - **Value Converter Value Subject Name Strategy**: Determines how to construct the subject name under which the value schema is registered with Schema Registry. - **Key Converter Key Subject Name Strategy**: Determines how to construct the subject name for key schema registration. - **Schema ID For Key Converter**: Sets the schema ID to use for deserialization when using `ConfigSchemaIdDeserializer`. This lets you specify a fixed schema ID for deserializing message keys. This property is applicable only when `key.converter.key.schema.id.deserializer` is set to `ConfigSchemaIdDeserializer`. **Auto-restart policy** - **Enable Connector Auto-restart**: Enables the auto-restart behavior of the connector and its task in the event of user-actionable errors. Defaults to `true`, enabling the connector to automatically restart in case of user-actionable errors. Set this property to `false` to disable auto-restart for failed connectors. If disabled, you must manually restart the connector. **Consumer configuration** - **Max poll interval(ms)**: Sets the maximum delay between subsequent consume requests to Kafka. Use this property to improve connector performance in cases when the connector cannot send records to the sink system. The default is 300,000 milliseconds (5 minutes). - **Max poll records**: Sets the maximum number of records to consume from Kafka in a single request. Use this property to improve connector performance in cases when the connector cannot send records to the sink system. The default is 500 records. **Transforms** - **Single Message Transformations**: To add a new SMT, see [Add transforms](single-message-transforms.md#cc-single-message-transforms-ui). For more information about unsupported SMTs, see [Unsupported transformations](single-message-transforms.md#cc-single-message-transforms-unsupported-transforms). **Processing position** - **Set offsets**: Click **Set offsets** to define a specific offset for this connector to begin procession data from. For more information on managing offsets, see [Manage offsets](offsets.md#connect-custom-offsets). See [Configuration Properties](#cc-http-sink-config-properties) for all property values and definitions. - Click **Continue**. ### Sizing Based on the number of topic partitions you select, you will be provided with a recommended number of tasks. 1. To change the number of recommended tasks, enter the number of [tasks](/platform/current/connect/concepts.html#tasks) for the connector to use in the **Tasks** field. 2. Click **Continue**. ### Review and Launch 1. Verify the connection details. 2. Click **Launch**. ![Launch the connector](images/ccloud-http-sink-status.png) The status for the connector should go from **Provisioning** to **Running**. ![Connector status](images/ccloud-http-sink-launch-connector.png) #### Step 5: Check for records Verify that records are being produced at the endpoint. For more information and examples to use with the Confluent Cloud API for Connect, see the [Confluent Cloud API for Connect Usage Examples](connect-api-section.md#ccloud-connect-api) section. ### Using the Confluent CLI To set up and run the connector using the Confluent CLI, complete the following steps. #### NOTE Make sure you have all your [prerequisites](#cc-http-sink-prereqs) completed. #### Step 1: List the available connectors Enter the following command to list available connectors: ```none confluent connect plugin list ``` #### Step 2: List the connector configuration properties Enter the following command to show the connector configuration properties: ```none confluent connect plugin describe ``` The command output shows the required and optional configuration properties. #### Step 3: Create the connector configuration file Create a JSON file that contains the connector configuration properties. The following example shows the required connector properties. ```json { "connector.class": "HttpSink", "input.data.format": "JSON", "name": "HttpSinkConnector_0", "kafka.auth.mode": "KAFKA_API_KEY", "kafka.api.key": "", "kafka.api.secret": "", "http.api.url": "http:://eshost1:9200/api/messages", "request.method": "POST", "tasks.max": "1", "topics": "orders", } ``` Note the following property definitions: * `"connector.class"`: Identifies the connector plugin name. * `"input.data.format"`: Sets the input Kafka record value format (data coming from the Kafka topic). Valid entries are **AVRO**, **JSON_SR**, **PROTOBUF**, **JSON**, or **BYTES**. You must have Confluent Cloud Schema Registry configured if using a schema-based message format (for example, Avro, JSON_SR (JSON Schema), or Protobuf). * `"name"`: Sets a name for your new connector. * `"kafka.auth.mode"`: Identifies the connector authentication mode you want to use. There are two options: `SERVICE_ACCOUNT` or `KAFKA_API_KEY` (the default). To use an API key and secret, specify the configuration properties `kafka.api.key` and `kafka.api.secret`, as shown in the example configuration (above). To use a [service account](service-account.md#s3-cloud-service-account), specify the **Resource ID** in the property `kafka.service.account.id=`. To list the available service account resource IDs, use the following command: ```bash confluent iam service-account list ``` For example: ```bash confluent iam service-account list Id | Resource ID | Name | Description +---------+-------------+-------------------+------------------- 123456 | sa-l1r23m | sa-1 | Service account 1 789101 | sa-l4d56p | sa-2 | Service account 2 ``` * `"http.api.url"`: Use an HTTP or HTTPS connection URL. For example, `http://eshost1:9200/api/messages` or `https://eshost3:9200/api/messages`. The connector sends the record value to the API as part of the request body. You can specify a static URL (for example, `http://eshost1:9200/api/messages`) or a dynamic URL (for example, `http://eshost1:9200/api/messages/${topic}/${key}`). You can also specify a dynamic URL using [fields from the Kafka record](#cc-http-sink-template-parameters). #### NOTE * Note that if the connection URL is HTTPS, HTTPS is used for all connections. A URL with no protocol is considered HTTP. * For additional information, see [HTTP Sink Connector](limits.md#http-sink-limits) limitations. * `"request.method"`: Enter an HTTP API Request Method: `PUT`, `POST`, or `PATCH`. Defaults to `POST`. * `"tasks.max"`: Enter the maximum number of [tasks](/platform/current/connect/concepts.html#tasks) for the connector to use. More tasks may improve performance (that is, consumer lag is reduced with multiple tasks running). * `"topics"`: Enter the topic name or a comma-separated list of topic names. **SMTs**: For details about adding SMTs using the Confluent CLI, see the [Single Message Transformations](single-message-transforms.md#cc-single-message-transforms) documentation. See [Configuration Properties](#cc-http-sink-config-properties) for all property values and descriptions. #### Step 3: Load the properties file and create the connector Enter the following command to load the configuration and start the connector: ```none confluent connect cluster create --config-file .json ``` For example: ```none confluent connect cluster create --config-file http-sink-config.json ``` Example output: ```none Created connector HttpSinkConnector_0 lcc-do6vzd ``` #### Step 4: Check the connector status. Enter the following command to check the connector status: ```none confluent connect cluster list ``` Example output: ```none ID | Name | Status | Type | Trace +------------+-------------------------------+---------+------+-------+ lcc-do6vzd | HttpSinkConnector_0 | RUNNING | sink | | ``` #### Step 5: Check for records Verify that records are populating the endpoint. For more information and examples to use with the Confluent Cloud API for Connect, see the [Confluent Cloud API for Connect Usage Examples](connect-api-section.md#ccloud-connect-api) section. ## Configuration Properties Use the following configuration properties with the fully managed connector. For self-managed connector property definitions and other details, see the connector docs in [Self-managed connectors for Confluent Platform](/platform/current/connect/kafka_connectors.html). #### NOTE These are properties for the fully managed cloud connector. If you are installing the connector locally for Confluent Platform, see [HTTP Sink Connector for Confluent Platform](https://docs.confluent.io/kafka-connectors/http/current/). ### Which topics do you want to get data from? `topics` : Identifies the topic name or a comma-separated list of topic names.
* Type: list * Importance: high ### Schema Config `schema.context.name` : Add a schema context name. A schema context represents an independent scope in Schema Registry. It is a separate sub-schema tied to topics in different Kafka clusters that share the same Schema Registry instance. If not used, the connector uses the default schema configured for Schema Registry in your Confluent Cloud environment.
* Type: string * Default: default * Importance: medium ### Input messages `input.data.format` : Sets the input Kafka record value format. Valid entries are AVRO, JSON_SR, PROTOBUF, JSON or BYTES. Note that you need to have Confluent Cloud Schema Registry configured if using a schema-based message format like AVRO, JSON_SR, and PROTOBUF.
* Type: string * Default: JSON * Importance: high `input.key.format` : Sets the input Kafka record key format. Valid entries are BYTES or STRING.
* Type: string * Default: BYTES * Valid Values: BYTES, STRING * Importance: high ### How should we connect to your data? `name` : Sets a name for your connector.
* Type: string * Valid Values: A string at most 64 characters long * Importance: high ### Kafka Cluster credentials `kafka.auth.mode` : Kafka Authentication mode. It can be one of KAFKA_API_KEY or SERVICE_ACCOUNT. It defaults to KAFKA_API_KEY mode, whenever possible.
* Type: string * Valid Values: SERVICE_ACCOUNT, KAFKA_API_KEY * Importance: high `kafka.api.key` : Kafka API Key. Required when kafka.auth.mode==KAFKA_API_KEY.
* Type: password * Importance: high `kafka.service.account.id` : The Service Account that will be used to generate the API keys to communicate with Kafka Cluster.
* Type: string * Importance: high `kafka.api.secret` : Secret associated with Kafka API key. Required when kafka.auth.mode==KAFKA_API_KEY.
* Type: password * Importance: high ### HTTP server details `http.api.url` : Specifies the API endpoint to which connector should write to.
* Type: string * Importance: high `request.method` : Specifies the HTTP request method (POST, PUT, PATCH) the connector should use for sending API request.
* Type: string * Default: POST * Importance: high `headers` : HTTP headers to be included in all requests. Individual headers should be separated by the Header Separator
* Type: string * Importance: high `header.separator` : Separator character used in headers
* Type: string * Importance: high `sensitive.headers` : Sensitive HTTP headers (eg: credentials) to be included in all requests. Individual headers should be separated by the Header Separator
* Type: password * Importance: high `behavior.on.null.values` : How to handle records with a non-null key and a null value (i.e. Kafka tombstone records). Valid options are `ignore`, `delete` and `fail`
* Type: string * Default: ignore * Importance: low ### HTTP server error handling `behavior.on.error` : Describes the error handling behavior configuration for handling error responses from HTTP requests. Accepted values are `ignore` and `fail`.
* Type: string * Default: ignore * Importance: medium `report.errors.as` : Dictates the content of records produced to the error topic. Accepted values are `error_string` and `http_response`. If set to `error_string`, the value would be a human readable string describing the failure. The value will include some or all of the following information if available: http response code, reason phrase, submitted payload, url, response content, exception and error message. If set to `http_response`, the value would be the plain response content for the request which failed to write the record. In both modes, any information about the failure will also be included in the error record’s headers.
* Type: string * Default: error_string * Importance: medium ### HTTP server batches `request.body.format` : Used to produce request body in either JSON or String format
* Type: string * Default: string * Importance: medium `batch.key.pattern` : Pattern used to build the key for a given batch. ${key} and ${topic} can be used to include message attributes here
* Type: string * Importance: high `batch.max.size` : The number of records accumulated in a batch before the HTTP API is invoked
* Type: int * Default: 1 * Importance: high `batch.prefix` : Prefix added to record batches. This is applied once at the beginning of the batch of records
* Type: string * Importance: high `batch.suffix` : Suffix added to record batches. This is applied once at the end of the batch of records
* Type: string * Importance: high `batch.separator` : Separator for records in a batch
* Type: string * Importance: high `batch.json.as.array` : Whether or not to use an array to bundle json records. Only used when request.body.format is set to json. This can be disabled only when batch.max.size is set to 1.
* Type: boolean * Importance: high ### HTTP server authentication `auth.type` : Specifies the authentication type of the API endpoint. Valid values are `NONE`, `BASIC`, `OAUTH2` (Client Credentials grant type only).
* Type: string * Default: NONE * Importance: high `connection.user` : The username to be used with an endpoint requiring authentication
* Type: string * Importance: high `connection.password` : The password to be used with an endpoint requiring authentication
* Type: password * Importance: high `oauth2.token.url` : The URL to be used for fetching OAuth2 token. Client Credentials is the only supported grant type.
* Type: string * Importance: high `oauth2.client.id` : The client id used when fetching OAuth2 token
* Type: string * Importance: high `oauth2.client.secret` : The secret used when fetching OAuth2 token
* Type: password * Importance: high `oauth2.token.property` : The name of the property containing the OAuth2 token returned by the http proxy.
* Type: string * Default: access_token * Importance: high `oauth2.client.auth.mode` : Specifies how to encode `client_id` and `client_secret` in the OAuth2 authorization request. If set to ‘header’, the credentials are encoded as an `'Authorization: Basic '` HTTP header. If set to ‘url’, then `client_id` and `client_secret` are sent in body as URL encoded parameters.
* Type: string * Default: header * Importance: low `oauth2.client.scope` : The scope used when fetching OAuth2 token. If empty, this parameter is not set in the authorization request
* Type: string * Default: any * Importance: low `oauth2.jwt.enabled` : Whether to generate and add JWT token to request. If selected, JWT token will be added as ‘jwt_token’ request param
* Type: boolean * Default: false * Importance: medium `oauth2.jwt.keystore.path` : Keystore containing private key to use to sign JWT.
* Type: password * Default: [hidden] * Importance: medium `oauth2.jwt.keystore.password` : Password to access keystore
* Type: password * Default: [hidden] * Importance: medium `oauth2.jwt.keystore.type` : JWT keystore type
* Type: string * Default: JKS * Importance: medium `oauth2.jwt.claimset` : JSON containing JWT claims
* Type: string * Default: “” * Importance: medium `oauth2.client.headers` : HTTP headers to be included in the OAuth2 client endpoint. Individual headers should be separated by OAuth2 Client Headers Separator
* Type: string * Importance: low `oauth2.client.header.separator` : Separator character used in OAuth2 Client Headers
* Type: string * Importance: low ### HTTP server retries `retry.on.status.codes` : The HTTP error codes to retry on. Comma-separated list of codes or range of codes to retry on. Ranges are specified with start and optional end code. Range boundaries are inclusive. For instance, ‘400-’ includes all codes greater than or equal to 400. ‘400-500’ includes codes from 400 to 500, including 500. Multiple ranges and single codes can be specified together to achieve fine grained control over retry behavior. For example, ‘404,408,500-’ will retry on 404 NOT FOUND, 408 REQUEST TIMEOUT, and all 5xx error codes
* Type: string * Default: 400- * Importance: medium `max.retries` : The maximum number of times to retry on errors before failing the task
* Type: int * Default: 3 * Importance: medium `retry.backoff.ms` : The initial duration in milliseconds to wait following an error before a retry attempt is made. Subsequent backoff attempts will be exponentially larger than the first duration. Note that this value is the initial backoff before retrying. After that, the connector will retry using exponential jitter. Jitter adds randomness to the exponential backoff algorithm to prevent synchronized retries.
* Type: int * Default: 3000 (3 seconds) * Valid Values: [100,…] * Importance: medium `http.connect.timeout.ms` : The time in milliseconds to wait for a connection to be established
* Type: int * Default: 30000 (30 seconds) * Importance: medium `http.request.timeout.ms` : The time in milliseconds to wait for a request response from the server
* Type: int * Default: 30000 (30 seconds) * Importance: medium `retry.backoff.policy` : The backoff policy to use in terms of retry - CONSTANT_VALUE or EXPONENTIAL_WITH_JITTER
* Type: string * Default: EXPONENTIAL_WITH_JITTER * Importance: medium ### HTTP server regular expressions `regex.patterns` : Regular expression patterns used for replacements in the message sent to the HTTP service. Multiple regular expression patterns can be specified, but must be separated by `regex.separator`
* Type: string * Importance: medium `regex.replacements` : Regex replacements to use with the patterns in `regex.patterns`. Multiple replacements can be specified, but must be separated by `regex.separator`. `${key}` and `${topic}` can be used here.
* Type: string * Importance: medium `regex.separator` : Separator character used in `regex.patterns` and `regex.replacements` property.
* Type: string * Importance: medium ### HTTP server SSL `https.ssl.key.password` : The password of the private key in the key store file. This is optional for client
* Type: password * Importance: high `https.ssl.keystorefile` : The key store containing server certificate. Only required if using https
* Type: password * Default: [hidden] * Importance: low `https.ssl.keystore.password` : The store password for the key store file. This is optional for a client and is only needed if https.ssl.keystore.location is configured
* Type: password * Importance: high `https.ssl.truststorefile` : The trust store containing server CA certificate. Only required if using https
* Type: password * Default: [hidden] * Importance: high `https.ssl.truststore.password` : The trust store password containing server CA certificate. Only required if using https
* Type: password * Importance: high `https.ssl.protocol` : The protocol to use for SSL connections
* Type: string * Default: TLSv1.3 * Importance: medium `https.host.verifier.enabled` : True if SSL host verification should be enabled
* Type: boolean * Default: true * Importance: medium ### Consumer configuration `max.poll.interval.ms` : The maximum delay between subsequent consume requests to Kafka. This configuration property may be used to improve the performance of the connector, if the connector cannot send records to the sink system. Defaults to 300000 milliseconds (5 minutes).
* Type: long * Default: 300000 (5 minutes) * Valid Values: [60000,…,1800000] for non-dedicated clusters and [60000,…] for dedicated clusters * Importance: low `max.poll.records` : The maximum number of records to consume from Kafka in a single request. This configuration property may be used to improve the performance of the connector, if the connector cannot send records to the sink system. Defaults to 500 records.
* Type: long * Default: 500 * Valid Values: [1,…,500] for non-dedicated clusters and [1,…] for dedicated clusters * Importance: low ### Number of tasks for this connector `tasks.max` : Maximum number of tasks for the connector.
* Type: int * Valid Values: [1,…] * Importance: high ### Additional Configs `key.converter.use.schema.guid` : The schema GUID to use for deserialization when using ConfigSchemaIdDeserializer. This allows you to specify a fixed schema GUID to be used for deserializing message keys. Only applicable when key.converter.key.schema.id.deserializer is set to ConfigSchemaIdDeserializer.
* Type: string * Importance: low `key.converter.use.schema.id` : The schema ID to use for deserialization when using ConfigSchemaIdDeserializer. This allows you to specify a fixed schema ID to be used for deserializing message keys. Only applicable when key.converter.key.schema.id.deserializer is set to ConfigSchemaIdDeserializer.
* Type: int * Importance: low `value.converter.connect.meta.data` : Allow the Connect converter to add its metadata to the output schema. Applicable for Avro Converters.
* Type: boolean * Importance: low `value.converter.use.schema.guid` : The schema GUID to use for deserialization when using ConfigSchemaIdDeserializer. This allows you to specify a fixed schema GUID to be used for deserializing message values. Only applicable when value.converter.value.schema.id.deserializer is set to ConfigSchemaIdDeserializer.
* Type: string * Importance: low `value.converter.use.schema.id` : The schema ID to use for deserialization when using ConfigSchemaIdDeserializer. This allows you to specify a fixed schema ID to be used for deserializing message values. Only applicable when value.converter.value.schema.id.deserializer is set to ConfigSchemaIdDeserializer.
* Type: int * Importance: low `errors.tolerance` : Use this property if you would like to configure the connector’s error handling behavior. WARNING: This property should be used with CAUTION for SOURCE CONNECTORS as it may lead to dataloss. If you set this property to ‘all’, the connector will not fail on errant records, but will instead log them (and send to DLQ for Sink Connectors) and continue processing. If you set this property to ‘none’, the connector task will fail on errant records.
* Type: string * Default: all * Importance: low `key.converter.key.schema.id.deserializer` : The class name of the schema ID deserializer for keys. This is used to deserialize schema IDs from the message headers.
* Type: string * Default: io.confluent.kafka.serializers.schema.id.DualSchemaIdDeserializer * Importance: low `key.converter.key.subject.name.strategy` : How to construct the subject name for key schema registration.
* Type: string * Default: TopicNameStrategy * Importance: low `value.converter.decimal.format` : Specify the JSON/JSON_SR serialization format for Connect DECIMAL logical type values with two allowed literals:
BASE64 to serialize DECIMAL logical types as base64 encoded binary data and
NUMERIC to serialize Connect DECIMAL logical type values in JSON/JSON_SR as a number representing the decimal value.
* Type: string * Default: BASE64 * Importance: low `value.converter.ignore.default.for.nullables` : When set to true, this property ensures that the corresponding record in Kafka is NULL, instead of showing the default column value. Applicable for AVRO,PROTOBUF and JSON_SR Converters.
* Type: boolean * Default: false * Importance: low `value.converter.reference.subject.name.strategy` : Set the subject reference name strategy for value. Valid entries are DefaultReferenceSubjectNameStrategy or QualifiedReferenceSubjectNameStrategy. Note that the subject reference name strategy can be selected only for PROTOBUF format with the default strategy being DefaultReferenceSubjectNameStrategy.
* Type: string * Default: DefaultReferenceSubjectNameStrategy * Importance: low `value.converter.replace.null.with.default` : Whether to replace fields that have a default value and that are null to the default value. When set to true, the default value is used, otherwise null is used. Applicable for JSON Converter.
* Type: boolean * Default: true * Importance: low `value.converter.schemas.enable` : Include schemas within each of the serialized values. Input messages must contain schema and payload fields and may not contain additional fields. For plain JSON data, set this to false. Applicable for JSON Converter.
* Type: boolean * Default: false * Importance: low `value.converter.value.schema.id.deserializer` : The class name of the schema ID deserializer for values. This is used to deserialize schema IDs from the message headers.
* Type: string * Default: io.confluent.kafka.serializers.schema.id.DualSchemaIdDeserializer * Importance: low `value.converter.value.subject.name.strategy` : Determines how to construct the subject name under which the value schema is registered with Schema Registry.
* Type: string * Default: TopicNameStrategy * Importance: low ### Auto-restart policy `auto.restart.on.user.error` : Enable connector to automatically restart on user-actionable errors.
* Type: boolean * Default: true * Importance: medium ## Frequently asked questions Find answers to frequently asked questions about the HTTP Sink connector for Confluent Cloud. ### Performance and consumer lag #### Why do I see consumer lag or `Commit of offsets timing out` errors? This occurs when the connector takes longer than `max.poll.interval.ms` to process the records specified in `max.poll.records`. When this timeout is exceeded, the task’s consumer is removed from the consumer group, causing the connector to reprocess messages and potentially send duplicates to the endpoint. **Common Causes:** * **Slow endpoint response:** The HTTP endpoint takes too long to process each request. * **No batching:** Records are sent individually (POST by POST) rather than in batches, causing each request to be processed separately. * **Endpoint load:** The target endpoint is under heavy load and cannot respond quickly. **Resolution:** 1. **Tune consumer properties:** Increase `max.poll.interval.ms` to give the connector more time to process records. For fully managed connectors, this is configured as `consumer.max.poll.interval.ms` in the connector settings. 2. **Reduce batch size:** Decrease `max.poll.records` to process fewer records per poll. For fully managed connectors, this is configured as `consumer.max.poll.records`. 3. **Enable batching:** If your endpoint supports arrays, configure `batch.max.size` to batch multiple records in a single HTTP request. 4. **Test endpoint performance:** Use tools like Postman or curl to measure how long your endpoint takes to respond to requests. #### Why am I seeing duplicate messages at my endpoint? The connector guarantees at least once delivery. If the connector fails to commit offsets before a timeout or rebalance occurs, it reprocesses messages from the last committed offset, resulting in duplicates at the endpoint. **Solution:** Follow the recommendations in the consumer lag section above to ensure offsets are committed successfully. ### OAuth and authentication #### Why does the connector not automatically refresh the OAuth2 token? The HTTP Sink connector refreshes OAuth2 tokens only when it receives an HTTP `401 (Unauthorized)` response and retries are enabled. By default, `retry.on.status.codes` is set to `"400-"` (all error codes >= 400), which includes 401. #### NOTE The connector does not refresh the OAuth2 token on an HTTP `403 (Forbidden)` response, even if `403` is included in `retry.on.status.codes`. A `403` is treated as an authorization failure (for example, insufficient scope or permissions) rather than an expired-token signal, so the request is retried with the same token. If your endpoint returns `403` for expired or invalid tokens instead of `401`, the connector cannot detect this and will not refresh the token automatically. **Issue:** If you override `retry.on.status.codes` and exclude `401`, token refresh does not work. **Resolution:** Ensure your `retry.on.status.codes` configuration includes `401`. For example: ```json "retry.on.status.codes": "401,403,429,500-" ``` #### Why do I get `invalid_scope` error with OAuth2 authentication? This error occurs when the `oauth2.client.scope` is rejected by your token endpoint. **Causes:** * The supplied scope is invalid for your endpoint. * If no scope is defined, the connector defaults to `"any"`, which may be rejected. **Resolution:** 1. Verify if your token endpoint requires a specific scope. 2. If no scope is required, set an empty string using the Confluent Cloud API or CLI: ```json "oauth2.client.scope": "" ``` Or using the CLI: ```bash confluent connect cluster update lcc- --config oauth2.client.scope="" ``` ### Batching and data processing #### Why is the connector not batching my records even though I set `batch.max.size`? The connector does not batch requests when messages contain different Kafka header values. This is because Kafka headers are passed as HTTP headers with each request, and messages with different headers cannot be batched together. **Solution:** * Ensure all messages in your topic have the same Kafka headers if you need batching. * If headers vary, consider whether your endpoint truly requires them, or if you can use SMTs to remove or standardize headers. #### Why does batching not work when I use template parameters in the URL? When you use [template parameters](#cc-http-sink-template-parameters) beyond `${topic}` and `${key}` (for example, `${order.id}` or `${customerId}`), each record can result in a unique URL. Batching is automatically disabled in this scenario because each request must go to a potentially different URL. **Solution:** This is expected behavior. If you need batching, use a static URL or limit template parameters to only `${topic}` and `${key}`. ### Error handling and monitoring #### How do I check if records are being sent to the error topic? For fully managed connectors, the error topic is automatically created with the naming pattern `error-lcc-` where `` is your connector ID. **Monitoring:** 1. In the Confluent Cloud Console, navigate to your cluster’s Topics view. 2. Look for the topic named `error-lcc-`. 3. Check if the topic has any recent production activity or retained bytes. 4. Similarly, check the success topic `success-lcc-` for successfully sent messages. **Headers:** Each message in the error or success topic includes headers showing the original record details: * `input_record_offset`: Offset of the original message * `input_record_timestamp`: Timestamp of the original message * `input_record_partition`: Partition of the original message * `input_record_topic`: Topic name of the original message #### How do I troubleshoot missing data at my endpoint? **Checklist:** 1. **Check error topic:** Verify if records were routed to `error-lcc-` instead of being sent successfully. 2. **Review connector logs:** Look for HTTP error codes, connection issues, or authentication failures. 3. **Test endpoint manually:** Use curl or Postman to verify your endpoint is reachable and accepting requests. 4. **Check for duplicates:** Verify if missing data might actually be present but with duplicates due to offset commit issues. ### Template parameters and URL construction #### Why does the connector fail when a field referenced in the URL does not exist? The connector throws a runtime exception if any field referenced in the HTTP API URL (`http.api.url`) does not exist in the Kafka record. **Solution:** * Ensure all fields used in template parameters exist in every record. * Use data validation or SMTs to ensure required fields are present before records reach the connector. ### Networking and connectivity #### Why do I see `UnknownHostException` or connection errors? This error indicates the connector cannot resolve or reach the target hostname. **Common Causes:** * **DNS resolution failure:** The hostname cannot be resolved. * **Firewall restrictions:** The endpoint is blocked by firewall rules. * **Network type mismatch:** For `PRIVATE_LINK` or `PCC` clusters, outbound traffic may not be configured. **Resolution:** 1. **Verify hostname:** Ensure the `http.api.url` hostname is correct and publicly resolvable (or resolvable from Confluent Cloud if using PrivateLink). 2. **Check network configuration:** Confirm your Confluent Cloud cluster’s network type and any required PrivateLink or VPC peering configurations. 3. **Allowlist IPs:** Ensure Confluent Cloud egress IPs for your region are allowlisted in your firewall or API gateway. 4. **Test connectivity:** Use curl or similar tools from a location with similar network access to verify the endpoint is reachable. For more information on network configuration, see [Prerequisites](#cc-http-sink-prereqs). ## Next Steps For an example that shows fully managed Confluent Cloud connectors in action with Confluent Cloud for Apache Flink, see the [Cloud ETL Demo](/platform/current/tutorials/examples/cloud-etl/docs/index.html). This example also shows how to use Confluent CLI to manage your resources in Confluent Cloud. [![image](images/topology.png)](https://docs.confluent.io/platform/current/tutorials/examples/cloud-etl/docs/index.html)