<a id="amps-source-connector"></a>

# AMPS Source Connector [Deprecated] for Confluent Platform

The Kafka Connect Advanced Message Processing System (AMPS) Source connector
allows you to export data from AMPS to Apache Kafka®. The connector subscribes to
messages from an AMPS topic and writes this data to a Kafka topic.

## Features

The AMPS Source connector supports the following features:

- [At least once delivery](#amps-source-connector-at-least-once-delivery)
- [Supports one task](#amps-source-supports-one-task)
- [Message header forwarding](#amps-source-connector-message-header-forwarding)
- [Timestamp processing](#amps-source-connector-timestamp-processing)
- [Kerberos authenticated AMPS installations](#amps-source-connector-kerberos-amps-installations)

<a id="amps-source-connector-at-least-once-delivery"></a>

### At least once delivery

The connector guarantees that messages from the AMPS topic are delivered at
least once to the corresponding Kafka topic. In the case of a connector
restart, there may be duplicate records in the Kafka topic.

<a id="amps-source-supports-one-task"></a>

### Supports one task

The AMPS Source connector supports running only one task.

<a id="amps-source-connector-message-header-forwarding"></a>

### Message header forwarding

The AMPS message data is sent in its `ByteArray` format as Kafka records values
to Kafka. It also optionally forwards AMPS message headers as Kafka record
headers. One can implement a custom `Converter` or use [Single
Message Transforms (SMTs)](/platform/current/connect/concepts.html#transforms) to convert
`ByteArray` values to any compatible format.

<a id="amps-source-connector-timestamp-processing"></a>

### Timestamp processing

Any Kafka record produced by the connector has the same timestamp as the
corresponding AMPS message, which translates to the time when AMPS processed the
message. If such a processed time isn’t available, the connector uses Kafka’s
current system time. For example, the current system time is used for AMP Out of
Focus (`oof`) messages.

<a id="amps-source-connector-kerberos-amps-installations"></a>

### Kerberos authenticated AMPS installations

The connector supports Kerberos authenticated AMPS installations.

<a id="amps-source-sow-topic-mapping"></a>

### State of the World (SOW) topic mapping

The messages from a SOW topic in AMPS are modeled similarly to a
[KTable](/platform/current/streams/concepts.html#ktable). More precisely, the value in a
record should be interpreted as an UPDATE of the last value for the same record
key, if applicable. If the same record key doesn’t exist, the update should be
considered an INSERT.

Additionally, `null` record values, or *tombstone* records, are interpreted as
DELETE for the record’s key. Note that the `oof` option must be set in the
subscription in order to record message deletes on AMPS. See [config
property](connector_options.md#amps-source-config-options) `amps.options` for more details.

The SOW key from the AMPS message is used as the Kafka record key which
facilitates this interpretation. Note that the corresponding Kafka topic
[log compacted](/platform/current/kafka/design.html#log-compaction) should closely model
a SOW topic in AMPS.

#### NOTE
Since AMPS doesn’t guarantee any particular order for SOW messages, no
ordering can be guaranteed for SOW records in Kafka.

<a id="amps-source-logged-topic-mapping"></a>

### Transactional logged topic mapping

A message from a transactional logged topic in AMPS is modeled as is. The
value in a record is the AMPS message data. The key for an AMPS message from a
transactional logged topic is the AMPS topic name. This ensures all the messages
go to a single partition of Kafka. The connector supports two different bookmarks
to subscribe to such a AMPS topic - `NOW` and `EPOCH`. For more details, see
[config property](connector_options.md#amps-source-config-options) `amps.bookmark`.

## Limitations

The connector currently doesn’t support historical SOW queries. That is,
`amps.topic.type` [configuration](connector_options.md#amps-source-config-options) can either
be SOW or Transactional logged.

## Install the AMPS Source 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.
- If you want to install the connector using Confluent Marketplace, you must install
  the [Confluent Hub Client](https://docs.confluent.io/home/connect/confluent-hub/client.html). This
  is installed by default with Confluent Enterprise.
- cp| 3.3.0 or later, or Kafka 0.11.0 or later
- Java 1.8
- AMPS Java Client: The connector is built with version 5.3.0.1. Versions 5.3.x.x
  should work.
- AMPS Server: Tested with version 5.3; versions 5.3.x.x recommended. Earlier
  versions should work because AMPS ensures backward compatibility. However,
  with old versions there is the potential for missing features.
- If you are running a multi-node Connect cluster, the AMPS connector and
  AMPS Client JAR must be installed on every Connect worker in the cluster.
  For more information, see [AMPS Client JAR](#amps-source-client-jar).

### Install the connector using the Confluent CLI

To install the `latest` connector version using [Confluent Hub Client](https://www.confluent.io/hub/), navigate to your Confluent Platform installation directory
and run the following command:

```bash
confluent connect plugin install confluentinc/kafka-connect-amps: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-amps:1.0.0
```

### Install the connector manually

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

<a id="amps-source-client-jar"></a>

## AMPS Client JAR

The AMPS Source connector uses the [AMPS Java Client API](https://www.crankuptheamps.com/documentation/client-apis/java/) to connect
to AMPS. In order for this to work, the connector must have the *AMPS Client*
for Java.

1. Click [Here](https://devnull.crankuptheamps.com/releases/amps/clients/java/5.3.0.1/amps-java-client-5.3.0.1.tar.gz) to download AMPS Java Client, version 5.3.0.1.
2. Place the JAR file into the `share/confluent-hub-components/confluentinc-kafka-connect-amps/lib` directory in your Confluent Platform installation **on each of the Connect worker nodes**.
3. Restart all of the Connect worker nodes.

#### NOTE
The `share/confluent-hub-components/confluentinc-kafka-connect-amps/lib`
directory referenced above is for Confluent Platform when this connector is installed
through Confluent Hub. If you are using a different installation, find the
location where the Confluent connector JAR files are located and place the
AMPS Client JAR file into the same directory.

## 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_options.md#amps-source-connector-license-config) for license properties and
[License topic configuration](connector_options.md#amps-source-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 AMPS Source Connector for Confluent Platform](connector_options.md#amps-source-config-options).

## Quick Start

In this quick start guide, the AMPS Source connector is used to consume messages
from an SOW topic called `Orders` on AMPS that has Kerberos authentication
enabled. It then sends these messages as records to a Kafka topic named
`AMPS_Orders` with headers being forwarded from the AMPS messages.

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

**Prerequisites:**

- [Confluent Platform](/platform/current/installation/installing_cp/index.html) is installed and services are running by using the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) commands.

  #### NOTE
  This quick start assumes that you are using the [Confluent
  CLI](https://docs.confluent.io/confluent-cli/current/index.html) commands, but [standalone
  installations](/platform/current/installation/installing_cp/index.html) are also supported. By
  default ZooKeeper, Apache Kafka®, Schema Registry, Kafka Connect REST API, and Kafka Connect
  are started with the `confluent local start` command. Note that as
  of Confluent Platform 7.5, ZooKeeper is deprecated for new deployments. Confluent recommends
  KRaft mode for new deployments.
- Kafka and Schema Registry are running locally on the default ports.

### Install AMPS locally and publish messages

1. Get the AMPS distribution and follow the [Getting Started](https://devnull.crankuptheamps.com/documentation/html/develop/user-guide/html/chapters/getting_started.html) guide to set up AMPS. Create the AMPS config file (for example `config.xml`) and set it to the following :
   ```bash
   <?xml version="1.0" encoding="UTF-8"?>

   <AMPSConfig>

     <Name>AMPS-Quick-Start</Name>

     <Admin>
       <InetAddr>localhost:8085</InetAddr>
     </Admin>

     <Transports>
       <Transport>
         <Name>any-tcp</Name>
         <Type>tcp</Type>
         <InetAddr>9007</InetAddr>
         <Protocol>amps</Protocol>
       </Transport>
     </Transports>

     <SOW>
       <Topic>
         <Name>Orders</Name>
         <FileName>./sow/%n.sow</FileName>
         <MessageType>json</MessageType>
         <Key>/id</Key>
       </Topic>
     </SOW>

   </AMPSConfig>
   ```
2. Run `bin/ampServer <path-to-config.xml>` from the AMPS installation folder to start the AMPS server.
3. Use the `spark` utility provided by AMPS to quickly publish few records to the `Orders` topic.
   ```bash
   # run from the AMPS installation folder
   $ bin/spark publish -server localhost:9007 -topic Orders -type json
   {"id": 1, "order": "Apples"}
   {"id": 2, "order": "Oranges"}
   total messages published: 2 (Infinity/s)
   ```

### Start the AMPS source connector

1. Create a JSON file for the AMPS Source connector (for example, `amps-source.json`).
   ```bash
   // substitute <> with your config
   {
     "name": "amps-source-connector",
     "config": {
         "connector.class": "io.confluent.connect.amps.AmpsSourceConnector",
         "tasks.max": "1",
         "kafka.topic" : "AMPS_Orders",
         "forward.kafka.headers": "true",
         "amps.servers": "<amps-servers-uri>",
         "amps.topic": "Orders",
         "amps.topic.type": "sow",
         "amps.command": "sow_and_subscribe",
         "amps.kerberos.keytab": "<path-to-user-keytab>",
         "amps.kerberos.principal": "<user-principal>",
         "amps.options": "timestamp,oof",
         "confluent.topic.bootstrap.servers": "localhost:9092",
         "confluent.topic.replication.factor": 1,
         "confluent.license": "<license>", // leave it empty for evaluation license
     }
   }
   ```
2. Load the AMPS source connector.
   ```bash
   confluent local load amps --config amps-source.json
   ```
3. Confirm that the connector is in a `RUNNING` state.
   ```bash
   confluent local status amps-source-connector
   ```
4. Confirm the messages were delivered to the configured topic in Kafka.
   ```bash
   confluent local consume AMPS_Orders --from-beginning
   ```

## Batch Size Tuning

The [Configuration Reference for AMPS Source Connector for Confluent Platform](connector_options.md#amps-source-config-options) provides options to tune both the AMPS
batch size, `amps.batch.size` and the Kafka batch size, `batch.size`.

* **AMPS Batch Size:** Tuning the AMPS batch size may produce better network utilization and improve the overall performance of the connector. The larger the AMPS batch size, the more messages AMPS sends to the network layer at a time, which can result in fewer packets being sent. For smaller messages, you should set this configuration to a higher value, like `50` or `100`. However, for large messages, even though a large AMPS batch size value may improve network utilization, the overall performance of the connector may decrease as these require larger sized buffers to store them.
* **Kafka Batch Size:** Tuning the Kafka batch size is recommended for large topics on AMPS. You should set the Kafka batch size configuration to a higher value, like `1000`. This is recommended for extremely large SOW topics; that is, topics with upwards of a million messages.