.. _azure-search-overview:

Azure Search Sink Connector for |cp|
------------------------------------

The |kconnect-long| Azure Search Sink connector allows moving data from |ak-tm|
to Azure Search. It writes each event from a topic in |ak| as document to an
index in Azure Search.

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

--------
Features
--------

The Azure Search connector offers the following features:

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

* **Reporting**: The connector writes the `HTTP responses <https://docs.microsoft.com/en-us/rest/api/searchservice/addupdate-or-delete-documents#request>`__ from Azure Search to success and error topics for each individual record.

* **Automatic Retries**: Occasionally, there may be `issues <https://docs.microsoft.com/en-us/rest/api/searchservice/http-status-codes>`__ writing to the Azure 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.

-------------
Prerequisites
-------------

The following are required to run the |kconnect-long| Azure Search Sink Connector:

* |ak| Broker: |cp| 3.3.0 or above
* |kconnect|: |cp| 4.1.0 or above
* Java 1.8

.. installation includes

.. include:: ../includes/connector-install-hub.rst

.. codewithvars:: bash


   confluent-hub install confluentinc/kafka-connect-azure-search:latest

.. include:: ../includes/connector-install-version.rst

.. codewithvars:: bash


   confluent-hub install confluentinc/kafka-connect-azure-search:1.0.0-preview


------------------------------
Install the connector manually
------------------------------

`Download and extract the ZIP file <https://www.confluent.io/connector/kafka-connect-azure-search/#download>`_ for your
connector and then follow the manual connector installation :ref:`instructions <connect_install_connectors>`.

-------
License
-------

.. include:: ../includes/enterprise-license.rst

See :ref:`azure-search-connector-license-config` for license properties and :ref:`azure-search-license-topic-configuration` for information about the license topic.

-----------
Quick Start
-----------

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

.. include:: ../../includes/install-cli-prereqs.rst

#.  Before starting the connector, create and deploy an Azure Search service.

    * Navigate to the Microsoft `Azure Portal <https://portal.azure.com/>`__.
    * Create a Search service following this `Azure Search quickstart guide <https://docs.microsoft.com/en-us/azure/search/search-create-service-portal>`__.
    * Create an index in the service following this `index quickstart 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 Search should now be set up for the connector.

    .. note:: 
      
      Make sure the index has the default name ``hotels-sample-index`` and only has the fields ``HotelId``, ``HotelName``, ``Description``. All others should be deleted.

#.  Install the connector through the :ref:`confluent_hub_client`.

    .. codewithvars:: bash

        # run from your CP installation directory
        confluent-hub install confluentinc/kafka-connect-azure-seach:latest

#.  Start |cp| using the :ref:`Confluent CLI <cli>` commands.

    .. include:: ../../includes/cli-new.rst

    .. codewithvars:: bash

        |confluent_start|

#.  Produce test data to the ``hotels-sample`` topic in |ak|.

    Start the Avro console producer to import a few records to |ak|:

    .. codewithvars:: bash

        <path-to-confluent>/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:

    .. codewithvars:: 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 |ak| topic ``hotels-sample`` in Avro format.

#.  Create a ``azure-search.json`` file with the following contents:

    .. codewithvars:: 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
          }
        }

    .. caution:: 
      
      Do not forget to change the ``azure.search.service.name`` and ``azure.search.api.key`` values in the JSON file to the copied name and admin key respectively. Find your `azure.search.service.name <https://docs.microsoft.com/en-us/azure/search/search-create-service-portal#name-the-service>`__.
      

#.  Load the Azure Search Sink Connector.

    .. include:: ../../includes/confluent-local-consume-limit.rst

    .. codewithvars:: bash

        |confluent_load| azure-search|dash| -d path/to/azure-search.json

    .. important:: Don't use the :ref:`Confluent CLI <cli>` commands in production environments.

#.  Confirm that the connector is in a ``RUNNING`` state.

    .. codewithvars:: bash

        |confluent_status| azure-search

#.  Confirm that the messages were delivered to the result topic in Kafka

    .. codewithvars:: bash

       |confluent_consume| test-result|dash| --from-beginning

#.  Confirm that the messages were delivered to Azure Search. Login to the service and check that the index ``hotel-samples-index`` contains the three written records from before.

#.  Cleanup resources

    * Delete the connector

      .. codewithvars:: bash

        |confluent_unload| azure-search

    * Stop |cp|

      .. codewithvars:: bash

        |confluent_stop|

    * Delete the created Azure Search service and its resource group in the Azure portal.

------------------------
Additional Documentation
------------------------

.. toctree::
    :titlesonly:

    connector_config
    changelog