.. _activemq-sink-connector:

|kconnect-long| ActiveMQ Sink Connector
=======================================

The ActiveMQ Sink Connector is used to move messages from |ak| to an `ActiveMQ <http://activemq.apache.org/>`_ cluster.

.. note::
    If you are required to use the Java Naming and Directory Interface™ (JNDI) to connect to ActiveMQ,
    there is a general :ref:`JMS Sink Connector <jms-sink-connector>` available that uses a
    JNDI-based mechanism to connect to the JMS broker.

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

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

* |ak| Broker: |cp| 3.3.0 or above, or |ak| 0.11.0 or above
* |kconnect|: |cp| 4.1.0 or above, or |ak| 1.1.0 or above (requires header support in |kconnect|)
* `ActiveMQ 5.x <https://activemq.apache.org/components/classic>`_
* Java 1.8

Install ActiveMQ Sink Connector
-------------------------------

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

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

.. codewithvars:: bash

   confluent-hub install confluentinc/kafka-connect-activemq-sink:latest

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

.. codewithvars:: bash

   confluent-hub install confluentinc/kafka-connect-activemq-sink:1.0.0-preview

--------------------------
Install Connector Manually
--------------------------

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

License
-------

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

See :ref:`activemq_sink_connector_license_config` for license properties and :ref:`activemq_sink_license-topic-configuration` for information about the license topic.

Client Library JARs
-------------------

The |kconnect-long| ActiveMQ connector includes all of the client libraries required to work with ActiveMQ.

..  note:: The ActiveMQ Sink Connector uses the ``org.apache.activemq:activemq-client:5.14.4`` client library.

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

This quick start uses the ActiveMQ Sink Connector to consume records from |ak| and send them to an ActiveMQ broker.

#.  `Install ActiveMQ <https://activemq.apache.org/getting-started#installation-procedure-for-unix>`_

#.  `Start ActiveMQ <https://activemq.apache.org/getting-started#starting-activemq>`_

#.  Install the connector through the `Confluent Hub Client <https://docs.confluent.io/current/connect/managing/confluent-hub/client.html>`_.

    .. codewithvars:: bash

        # run from your Confluent Platform installation directory
        confluent-hub install confluentinc/kafka-connect-activemq-sink:latest

#.  Start |cp|.

    .. codewithvars:: bash

        |confluent_start|

#.  `Produce <https://docs.confluent.io/current/cli/command-reference/confluent-produce.html>`_ test data to the ``sink-messages`` topic in |ak|.

    .. codewithvars:: bash

        seq 10 | |confluent_produce| sink-messages

#.  Create a ``activemq-sink.json`` file with the following contents:

    .. codewithvars:: json

        {
          "name": "AMQSinkConnector",
          "config": {
            "connector.class": "io.confluent.connect.jms.ActiveMqSinkConnector",
            "tasks.max": "1",
            "topics": "sink-messages",
            "activemq.url": "tcp://localhost:61616",
            "activemq.username": "connectuser",
            "activemq.password": "connectuser",
            "jms.destination.type": "queue",
            "jms.destination.name": "connector-quickstart",
            "key.converter": "org.apache.kafka.connect.storage.StringConverter",
            "value.converter": "org.apache.kafka.connect.storage.StringConverter",
            "confluent.topic.bootstrap.servers": "localhost:9092",
            "confluent.topic.replication.factor": "1"
          }
        }

#.  Load the ActiveMQ Sink Connector.

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

    .. codewithvars:: bash

        |confluent_load| jms|dash| -d activemq-sink.json

    .. important:: Don't use the :ref:`cli` in production environments.

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

    .. codewithvars:: bash

        |confluent_status| AMQSinkConnector

#.  Navigate to the `ActiveMQ Admin UI <http://localhost:8161/admin>`_ or use the following ActiveMQ CLI command to confirm the messages were delivered to the ``connector-quickstart`` queue.

    .. tip:: The default credentials for the ActiveMQ Admin UI are ``admin``/``admin``
    
    .. codewithvars:: bash

        ./bin/activemq consumer --destination connector-quickstart --messageCount 10


.. include:: ../../includes/jms-message-formats.rst

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

.. toctree::
  :maxdepth: 1

  connector_config
  changelog