<a id="azure-search-overview"></a>

# Azure Cognitive Search Sink Connector for Confluent Platform

The Kafka Connect Azure Cognitive Search Sink connector allows moving data
from Apache Kafka® to Azure Cognitive Search. It writes each event from a topic in
Kafka as document to an index in Azure Cognitive Search.

The connector leverages Azure Cognitive Search’s REST API to send records as
documents.

#### IMPORTANT
<!-- WARNING: THIS IS A SHARED FILE AND THE SOURCE IS LOCATED IN DOCS-COMMON. DO NOT ADD TO ANY OTHER REPO. -->

Effective July 6, 2025, only self-managed connector versions that meet or exceed the minimum version listed on the
[Supported Connector Versions](https://docs.confluent.io/platform/7.8/connect/supported-connector-version.html#supported-connector-versions-till-cp-7-8)
page receive support from Confluent. Older, unsupported connector versions have been removed from Confluent Marketplace and
are no longer available for download.

## Features

The Azure Cognitive Search Sink connector includes the following features:

- [At least once delivery](#azure-search-sink-at-least-once-delivery)
- [Dead Letter Queue](#azure-search-sink-dead-letter-queue)
- [Multiple tasks](#azure-search-sink-multiple-tasks)
- [Ordered writes](#azure-search-sink-ordered-writes)
- [Reporting](#azure-search-sink-reporting)
- [Automatic retries](#azure-search-sink-automatic-retries)
- [Client-side encryption](#azure-search-sink-sink-csfle-sm)

<a id="azure-search-sink-at-least-once-delivery"></a>

### At least once delivery

This connector guarantees that records from the Kafka topic are delivered at
least once.

<a id="azure-search-sink-dead-letter-queue"></a>

### Dead Letter Queue

This connector supports the Dead Letter Queue (DLQ) functionality. For
information about accessing and using the DLQ, see [Confluent Platform
Dead Letter Queue](/platform/current/connect/concepts.html#dead-letter-queue).

<a id="azure-search-sink-multiple-tasks"></a>

### Multiple tasks

The Azure Cognitive Search Sink connector supports running one or more tasks. You can
specify the number of tasks in the `tasks.max` configuration parameter. This
can lead to performance gains when multiple files need to be parsed.

<a id="azure-search-sink-ordered-writes"></a>

### Ordered writes

The connector writes records in exactly the same order that it receives them.
And for uniqueness, the Kafka coordinates (topic, partition, and offset) can be
used as the document key. Otherwise, the connector can use the record key as the
document key.

<a id="azure-search-sink-reporting"></a>

### Reporting

The connector writes the [HTTP responses](https://docs.microsoft.com/en-us/rest/api/searchservice/addupdate-or-delete-documents#request)
from Azure Cognitive Search to success and error topics for each individual
record.

<a id="azure-search-sink-automatic-retries"></a>

### Automatic retries

Occasionally, there may be [issues](https://docs.microsoft.com/en-us/rest/api/searchservice/http-status-codes)
writing to the Azure Cognitive Search service and the connector will retry all
retry-able requests. The maximum amount of time that the connector spends
retrying can be specified by the `max.retry.ms` config.

<a id="azure-search-sink-sink-csfle-sm"></a>

### Client-side encryption

This connector supports Client-Side Field Level Encryption (CSFLE) and Client-Side Payload Encryption (CSPE). For more information, see [Manage Client-Side Encryption](https://docs.confluent.io/platform/current/connect/manage-csfle.html).

## Limitations

The connector does not currently support [Single Message Transformations (SMTs)](https://docs.confluent.io/platform/current/connect/transforms/overview.html)
that modify the topic name. Additionally, the following transformations are not
allowed:

- `io.debezium.transforms.ByLogicalTableRouter`
- `io.debezium.transforms.outbox.EventRouter`
- `org.apache.kafka.connect.transforms.RegexRouter`
- `org.apache.kafka.connect.transforms.TimestampRouter`
- `io.confluent.connect.transforms.MessageTimestampRouter`
- `io.confluent.connect.transforms.ExtractTopic$Key`
- `io.confluent.connect.transforms.ExtractTopic$Value`

## License

You can use this connector for a 30-day trial period without a license key.

After 30 days, you must purchase a connector subscription which includes [Confluent enterprise license](/platform/current/installation/license.html#enterprise-subscription-license) keys to subscribers, along with [enterprise-level support](https://www.confluent.io/subscription/) for Confluent Platform and your connectors. If you are a subscriber, you can contact [Confluent Support](https://support.confluent.io/) for more information.

See [Confluent Platform license](connector_config.md#azure-search-connector-license-config) for license properties and
[Confluent license topic configuration](connector_config.md#azure-search-license-topic-configuration) for information about the license
topic.

## Configuration Properties

For a complete list of configuration properties for this connector, see
[Configuration Reference for Azure Cognitive Search Sink Connector for Confluent Platform](connector_config.md#azure-search-overview-config).

For an example of how to get Kafka Connect connected to [Confluent Cloud](/cloud/current/index.html), see
[Connect Self-Managed Kafka Connect to Confluent Cloud](/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster).

## Install the Azure Cognitive Search Sink Connector

You can install this connector by using the [confluent connect
plugin
install](https://docs.confluent.io/confluent-cli/current/command-reference/connect/plugin/confluent_connect_plugin_install.html)
command, or by manually downloading the ZIP file.

### Prerequisites

- You must install the connector on every machine where Connect will run.
- Kafka Broker: Confluent Platform 3.3.0 or later.
- Connect: Confluent Platform 4.1.0 or later.
- Java 1.8.
- An installation of the latest (`latest`) connector version.

  To install the `latest` connector version, navigate to your Confluent Platform
  installation directory and run the following command:
  ```bash
  confluent connect plugin install confluentinc/kafka-connect-azure-search:latest
  ```

  You can install a specific version by replacing `latest` with a version
  number as shown in the following example:
  ```bash
  confluent connect plugin install confluentinc/kafka-connect-azure-search:1.1.4
  ```

### Install the connector manually

[Download and extract the ZIP file](https://www.confluent.io/hub/confluentinc/kafka-connect-azure-search) for
your connector and then follow the manual connector installation
[instructions](/kafka-connectors/self-managed/userguide.html#connect-installing-plugins).

## Quick Start

This quick start uses the Azure Cognitive Search Sink connector to consume
records and write them as documents to an Azure Cognitive Search service.

Prerequisites
: - [Confluent Platform](/platform/current/installation/index.html)
  - [Confluent CLI](https://docs.confluent.io/confluent-cli/current/installing.html) (requires separate installation)

1. Before starting the connector, create and deploy an Azure Cognitive Search service.
   * Navigate to the Microsoft [Azure Portal](https://portal.azure.com/).
   * Create a Search service following this [Azure Cognitive Search quick start guide](https://docs.microsoft.com/en-us/azure/search/search-create-service-portal).
   * Create an index in the service following this [index quick start guide](https://docs.microsoft.com/en-us/azure/search/search-get-started-portal).
   * Copy the admin key and the Search service name from the portal and save it for later. Azure Cognitive Search should now be set up for the connector.

   #### NOTE
   Ensure the index has the default name `hotels-sample-index` and only has the fields `HotelId`, `HotelName`, `Description`. All others should be deleted.
2. Install the connector through the [Confluent Hub Client](/kafka-connectors/self-managed/confluent-hub/client.html).
   ```bash
   # run from your CP installation directory
   confluent connect plugin install confluentinc/kafka-connect-azure-seach:latest
   ```
3. Start Confluent Platform using the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) commands.
   ```bash
   confluent local start
   ```
4. Produce test data to the `hotels-sample` topic in Kafka.

   Start the Avro console producer to import a few records to Kafka:
   ```bash
   ${CONFLUENT_HOME}/bin/kafka-avro-console-producer --broker-list localhost:9092 --topic hotels-sample \
   --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"HotelName","type":"string"},{"name":"Description","type":"string"}]}' \
   --property key.schema='{"type":"string"}' \
   --property "parse.key=true" \
   --property "key.separator=,"
   ```

   Then in the console producer, enter:
   ```bash
   "marriotId",{"HotelName": "Marriot", "Description": "Marriot description"}
   "holidayinnId",{"HotelName": "HolidayInn", "Description": "HolidayInn description"}
   "motel8Id",{"HotelName": "Motel8", "Description": "motel8 description"}
   ```

   The three records entered are published to the Kafka topic `hotels-sample` in Avro format.
5. Create a `azure-search.json` file with the following contents:
   ```bash
   {
     "name": "azure-search",
     "config": {
       "topics": "hotels-sample",
       "tasks.max": "1",
       "connector.class": "io.confluent.connect.azure.search.AzureSearchSinkConnector",
       "key.converter": "io.confluent.connect.avro.AvroConverter",
       "key.converter.schema.registry.url": "http://localhost:8081",
       "value.converter": "io.confluent.connect.avro.AvroConverter",
       "value.converter.schema.registry.url": "http://localhost:8081",
       "confluent.topic.bootstrap.servers": "localhost:9092",
       "confluent.topic.replication.factor": "1",
       "azure.search.service.name": "<the created Search service name>",
       "azure.search.api.key": "<the copied api key>",
       "index.name": "${topic}-index",
       "reporter.bootstrap.servers": "localhost:9092",
       "reporter.error.topic.name": "test-error",
       "reporter.error.topic.replication.factor": 1,
       "reporter.error.topic.key.format": "string",
       "reporter.error.topic.value.format": "string",
       "reporter.result.topic.name": "test-result",
       "reporter.result.topic.key.format": "string",
       "reporter.result.topic.value.format": "string",
       "reporter.result.topic.replication.factor": 1
     }
   }
   ```

   #### NOTE
   For details about using this connector with Kafka Connect Reporter, see
   [Connect
   Reporter](/kafka-connectors/self-managed/userguide.html#userguide-connect-reporter).
6. Load the Azure Cognitive Search Sink connector.
   ```bash
   confluent local load azure-search --config path/to/azure-search.json
   ```

   #### IMPORTANT
   Don’t use the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) commands in
   production environments.
7. Confirm that the connector is in a `RUNNING` state.
   ```bash
   confluent local status azure-search
   ```
8. Confirm that the messages were delivered to the result topic in Kafka
   ```bash
   confluent local consume test-result --from-beginning
   ```
9. Confirm that the messages were delivered to Azure Cognitive Search.
10. Log in to the service and check that the index `hotel-samples-index`
    contains the three written records from before.
11. Clean up resources:
    1. Delete the connector
       ```bash
       confluent local unload azure-search
       ```
    2. Stop Confluent Platform
       ```bash
       confluent local stop
       ```
    3. Delete the created Azure Cognitive Search service and its resource group
       in the Azure portal.