[[Tryitoutlocally-Tryitoutlocally]]
= Try it out locally

[[Tryitoutlocally-RunKafka]]
== Run Kafka

First, get a locally running kafka instance by following Apache Kafka https://kafka.apache.org/quickstart[quickstart guide]. This usually boils down to:

.Set KAFKA_HOME
[source,bash]
----
export KAFKA_HOME=<your kafka install dir>
----

.Start Zookeeper cluster
[source,bash]
----
$KAFKA_HOME/bin/zookeeper-server-start.sh $KAFKA_HOME/config/zookeeper.properties
----

.Start Kafka broker
[source,bash]
----
$KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/config/server.properties
----

.Create "mytopic" topic
[source,bash]
----
$KAFKA_HOME/bin/kafka-topics.sh --create \
  --zookeeper localhost:2181 \
  --replication-factor 1 \
  --partitions 1 \
  --topic mytopic
----

For using the quickstart we'll use the `plugin.path` property, so you'll have to add a path for your connectors.

Open your configuration file located at `$KAFKA_HOME/config/connect-standalone.properties`

and add a property at the end

[source,bash]
----
# Set to a list of filesystem paths separated by commas (,) to enable class loading isolation for plugins
# (connectors, converters, transformations). The list should consist of top level directories that include 
# any combination of: 
# a) directories immediately containing jars with plugins and their dependencies
# b) uber-jars with plugins and their dependencies
# c) directories immediately containing the package directory structure of classes of plugins and their dependencies
# Note: symlinks will be followed to discover dependencies or plugins.
# Examples: 
# plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,
plugin.path=/home/connectors
----

At this point you're able to run the connectors quickstart.

Next, run Camel kafka connectors source and/or syncs:

You can use these Kafka utilities to listen or produce from a Kafka topic:

.Run an Kafka Consumer
[source,bash]
----
$KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic mytopic --from-beginning
----

.Run an interactive CLI kafka producer
[source,bash]
----
$KAFKA_HOME/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic mytopic
----

[[Tryitoutlocally-TryExamples]]
== Try some examples

For the following examples you need to fetch the `camel-kafka-connector` project and https://github.com/apache/camel-kafka-connector/blob/master/README.adoc#build-the-project[build] it locally by running `./mvnw package` from the root of the project. Look into the `config` and `docs/examples` directories for the configuration files (`*.properties`) of the examples showcased here.

[[Tryitoutlocally-SimpleLogger]]
=== Simple logger (sink)

First thing to do, is unzip or untar the camel-log-kafka-connector archive in the `plugin.path` location. After building the project you should have in `connectors/camel-log-kafka-connector/target/` a .zip file named `camel-log-kafka-connector-0.6.0-SNAPSHOT-package.zip`

[source,bash]
----
> cd /home/connectors/
> cp connectors/camel-log-kafka-connector/target/camel-log-kafka-connector-0.6.0-SNAPSHOT-package.zip .
> unzip camel-log-kafka-connector-0.6.0-SNAPSHOT-package.zip
----

.Run the default sink, just a camel logger:
[source,bash]
----
$KAFKA_HOME/bin/connect-standalone.sh $KAFKA_HOME/config/connect-standalone.properties docs/examples/CamelSinkConnector.properties 
----

[[Tryitoutlocally-Timer]]
=== Timer (source)

First thing to do, is unzip or untar the camel-timer-kafka-connector archive in the `plugin.path` location. After building the project you should have in `connectors/camel-timer-kafka-connector/target/` a .zip file named `camel-timer-kafka-connector-0.6.0-SNAPSHOT-package.zip`

[source,bash]
----
> cd /home/connectors/
> cp connectors/camel-timer-kafka-connector/target/camel-log-kafka-connector-0.6.0-SNAPSHOT-package.zip .
> unzip camel-timer-kafka-connector-0.6.0-SNAPSHOT-package.zip
----

This is an example of a _source_ that produces a message every second to `mytopic`.

.Run the default source, just a camel timer:
[source,bash]
----
$KAFKA_HOME/bin/connect-standalone.sh $KAFKA_HOME/config/connect-standalone.properties docs/examples/CamelSourceConnector.properties
----

[[Tryitoutlocally-AwsKinesis]]
=== AWS Kinesis (source)

First thing to do, is unzip or untar the camel-aws-kinesis-kafka-connector archive in the `plugin.path` location. After building the project you should have in `connectors/camel-aws-kinesis-kafka-connector/target/` a .zip file named `camel-aws-kinesis-kafka-connector-0.6.0-SNAPSHOT-package.zip`

[source,bash]
----
> cd /home/connectors/
> cp connectors/camel-aws-kinesis-kafka-connector/target/camel-aws-kinesis-kafka-connector-0.6.0-SNAPSHOT-package.zip .
> unzip camel-aws-kinesis-kafka-connector-0.6.0-SNAPSHOT-package.zip
----

This example consumes from AWS Kinesis data stream and transfers the payload to `mytopic` topic in Kafka.

Adjust properties in `examples/CamelAWSKinesisSourceConnector.properties` for your environment, you need to configure access key, secret key and region by setting `camel.component.aws-kinesis.configuration.access-key=youraccesskey`, `camel.component.aws-kinesis.configuration.secret-key=yoursecretkey` and `camel.component.aws-kinesis.configuration.region=yourregion`.

.Run the AWS Kinesis source:
[source,bash]
----
$KAFKA_HOME/bin/connect-standalone.sh $KAFKA_HOME/config/connect-standalone.properties docs/examples/CamelAWSKinesisSourceConnector.properties
----

[[Tryitoutlocally-AWSSQSSink]]
=== AWS SQS (sink)

First thing to do, is unzip or untar the camel-aws-sqs-kafka-connector archive in the `plugin.path` location. After building the project you should have in `connectors/camel-aws-sqs-kafka-connector/target/` a .zip file named `camel-aws-sqs-kafka-connector-0.6.0-SNAPSHOT-package.zip`

[source,bash]
----
> cd /home/connectors/
> cp connectors/camel-aws-sqs-kafka-connector/target/camel-aws-sqs-kafka-connector-0.6.0-SNAPSHOT-package.zip .
> unzip camel-aws-sqs-kafka-connector-0.6.0-SNAPSHOT-package.zip
----

This example consumes from Kafka topic `mytopic` and transfers the payload to AWS SQS.

Adjust properties in `examples/CamelAWSSQSSinkConnector.properties` for your environment, you need to configure access key, secret key and region by setting `camel.component.aws-sqs.configuration.access-key=youraccesskey`, `camel.component.aws-sqs.configuration.secret-key=yoursecretkey` and `camel.component.aws-sqs.configuration.region=yourregion`

.Run the AWS SQS sink:
[source,bash]
----
$KAFKA_HOME/bin/connect-standalone.sh $KAFKA_HOME/config/connect-standalone.properties docs/examples/CamelAWSSQSSinkConnector.properties
----

[[Tryitoutlocally-AWSSQSSource]]
=== AWS SQS (source)

First thing to do, is unzip or untar the camel-aws-sqs-kafka-connector archive in the `plugin.path` location. After building the project you should have in `connectors/camel-aws-sqs-kafka-connector/target/` a .zip file named `camel-aws-sqs-kafka-connector-0.6.0-SNAPSHOT-package.zip`

[source,bash]
----
> cd /home/connectors/
> cp connectors/camel-aws-sqs-kafka-connector/target/camel-aws-sqs-kafka-connector-0.6.0-SNAPSHOT-package.zip .
> unzip camel-aws-sqs-kafka-connector-0.6.0-SNAPSHOT-package.zip
----

This example consumes from AWS SQS queue `mysqs` and transfers the payload to `mytopic` topic in Kafka.

Adjust properties in `examples/CamelAWSSQSSourceConnector.properties` for your environment, you need to configure access key, secret key and region by setting `camel.component.aws-sqs.configuration.access-key=youraccesskey`, `camel.component.aws-sqs.configuration.secret-key=yoursecretkey` and `camel.component.aws-sqs.configuration.region=yourregion`

.Run the AWS SQS source:
[source,bash]
----
$KAFKA_HOME/bin/connect-standalone.sh $KAFKA_HOME/config/connect-standalone.properties docs/examples/CamelAWSSQSSourceConnector.properties
----

[[Tryitoutlocally-AWSSNSSink]]
=== AWS SNS (sink)

First thing to do, is unzip or untar the camel-aws-sqs-kafka-connector archive in the `plugin.path` location. After building the project you should have in `connectors/camel-aws-sns-kafka-connector/target/` a .zip file named `camel-aws-sns-kafka-connector-0.6.0-SNAPSHOT-package.zip`

[source,bash]
----
> cd /home/connectors/
> cp connectors/camel-aws-sns-kafka-connector/target/camel-aws-sns-kafka-connector-0.6.0-SNAPSHOT-package.zip .
> unzip camel-aws-sns-kafka-connector-0.6.0-SNAPSHOT-package.zip
----

This example consumes from `mytopic` Kafka topic and transfers the payload to AWS SNS `topic` topic.

Adjust properties in `examples/CamelAWSSNSSinkConnector.properties` for your environment, you need to configure access key, secret key and region by setting `camel.component.aws-sns.configuration.access-key=youraccesskey`, `camel.component.aws-sns.configuration.secret-key=yoursecretkey` and `camel.component.aws-sns.configuration.region=yourregion`

.Run the AWS SNS sink:
[source,bash]
----
$KAFKA_HOME/bin/connect-standalone.sh $KAFKA_HOME/config/connect-standalone.properties docs/examples/CamelAWSSNSSinkConnector.properties
----

[[Tryitoutlocally-AWSSNSSource]]
=== AWS S3 (source)

First thing to do, is unzip or untar the camel-aws-sqs-kafka-connector archive in the `plugin.path` location. After building the project you should have in `connectors/camel-aws-s3-kafka-connector/target/` a .zip file named `camel-aws-s3-kafka-connector-0.6.0-SNAPSHOT-package.zip`

[source,bash]
----
> cd /home/connectors/
> cp connectors/camel-aws-s3-kafka-connector/target/camel-aws-s3-kafka-connector-0.6.0-SNAPSHOT-package.zip .
> unzip camel-aws-s3-kafka-connector-0.6.0-SNAPSHOT-package.zip
----

This example fetches objects from AWS S3 in the `camel-kafka-connector` bucket and transfers the payload to `mytopic` Kafka topic. This example shows how to implement a custom converter converting from bytes received from S3 to Kafka's `SchemaAndValue`.

Adjust properties in `examples/CamelAWSS3SourceConnector.properties` for your environment, you need to configure access key, secret key and region by adding `camel.component.aws-s3.configuration.access-key=youraccesskey`, `camel.component.aws-s3.configuration.secret-key=yoursecretkey` and `camel.component.aws-s3.configuration.region=yourregion`

.Run the AWS S3 source:
[source,bash]
----
$KAFKA_HOME/bin/connect-standalone.sh $KAFKA_HOME/config/connect-standalone.properties docs/examples/CamelAWSS3SourceConnector.properties
----

[[Tryitoutlocally-CassandraQL]]
=== Apache Cassandra

First thing to do, is unzip or untar the camel-aws-sqs-kafka-connector archive in the `plugin.path` location. After building the project you should have in `connectors/camel-cql-kafka-connector/target/` a .zip file named `camel-cql-kafka-connector-0.6.0-SNAPSHOT-package.zip`

[source,bash]
----
> cd /home/connectors/
> cp connectors/camel-cql-kafka-connector/target/camel-cql-kafka-connector-0.6.0-SNAPSHOT-package.zip .
> unzip camel-cql-kafka-connector-0.6.0-SNAPSHOT-package.zip
----

This examples require a running Cassandra instance, for simplicity the steps below show how to start Cassandra using Docker. First you'll need to run a Cassandra instance:

[source,bash]
----
docker run --name master_node --env MAX_HEAP_SIZE='800M' -dt oscerd/cassandra
----

Next, check and make sure Cassandra is running:

[source,bash]
----
docker exec -ti master_node /opt/cassandra/bin/nodetool status
Datacenter: datacenter1
=======================
Status=Up/Down
|/ State=Normal/Leaving/Joining/Moving
--  Address     Load       Tokens       Owns (effective)  Host ID                               Rack
UN  172.17.0.2  251.32 KiB  256          100.0%            5126aaad-f143-43e9-920a-0f9540a93967  rack1
----

To populate the database using to the `cqlsh` tool, you'll need a local installation of Cassandra. Download and extract the Apache Cassandra distribution to a directory. We reference the Cassandra installation directory with `LOCAL_CASSANDRA_HOME`. Here we use version 3.11.4 to connect to the Cassandra instance we started using Docker.

[source,bash]
----
<LOCAL_CASSANDRA_HOME>/bin/cqlsh $(docker inspect --format='{{ .NetworkSettings.IPAddress }}' master_node)
----

Next, execute the following script to create keyspace `test`, the table `users` and insert one row into it.

[source,bash]
----
create keyspace test with replication = {'class':'SimpleStrategy', 'replication_factor':3};
use test;
create table users ( id int primary key, name text );
insert into users (id,name) values (1, 'oscerd');
quit;
----

In the configuration `.properties` file we use below the IP address of the Cassandra master node needs to be configured, replace the value `172.17.0.2` in the `camel.source.url` or `localhost` in `camel.sink.url` configuration property with the IP of the master node obtained from Docker. Each example uses a different `.properties` file shown in the command line to run the example.

[source,bash]
----
docker inspect --format='{{ .NetworkSettings.IPAddress }}' master_node
----

[[Tryitoutlocally-CassandraQLSource]]
==== Apache Cassandra (source)

This example polls Cassandra via CSQL (`select * from users`) in the `test` keyspace and transfers the result to the `mytopic` Kafka topic. 

.Run the Cassandra CQL source:
[source,bash]
----
$KAFKA_HOME/bin/connect-standalone.sh $KAFKA_HOME/config/connect-standalone.properties docs/examples/CamelCassandraQLSourceConnector.properties
----

[[Tryitoutlocally-CassandraQLSink]]
==== Apache Cassandra (sink)

First thing to do, is unzip or untar the camel-aws-sqs-kafka-connector archive in the `plugin.path` location. After building the project you should have in `connectors/camel-cql-kafka-connector/target/` a .zip file named `camel-cql-kafka-connector-0.6.0-SNAPSHOT-package.zip`

[source,bash]
----
> cd /home/connectors/
> cp connectors/camel-cql-kafka-connector/target/camel-cql-kafka-connector-0.6.0-SNAPSHOT-package.zip .
> unzip camel-cql-kafka-connector-0.6.0-SNAPSHOT-package.zip
----

This example adds data to the `users` table in Cassandra from the data consumed from the `mytopic` Kafka topic. Notice how the `name` column is populated from the Kafka message using CQL command `insert into users...`.

.Run the Cassandra CQL sink:
[source,bash]
----
$KAFKA_HOME/bin/connect-standalone.sh $KAFKA_HOME/config/connect-standalone.properties docs/examples/CamelCassandraQLSinkConnector.properties
----

[[Tryitoutlocally-ElasticsearchSink]]
=== Elasticsearch (sink)

First thing to do, is unzip or untar the camel-aws-sqs-kafka-connector archive in the `plugin.path` location. After building the project you should have in `connectors/camel-elasticsearch-rest-kafka-connector/target/` a .zip file named `camel-elasticsearch-rest-kafka-connector-0.6.0-SNAPSHOT-package.zip`

[source,bash]
----
> cd /home/connectors/
> cp connectors/camel-elasticsearch-rest-kafka-connector/target/camel-elasticsearch-rest-kafka-connector-0.6.0-SNAPSHOT-package.zip .
> unzip camel-elasticsearch-rest-kafka-connector-0.6.0-SNAPSHOT-package.zip
----

This example passes data from `mytopic` Kafka topic to `sampleIndexName` index in Elasticsearch. Adjust properties in `docs/examples/CamelElasticSearchSinkConnector.properties` to reflect your environment, for example change the `hostAddresses` to a valid Elasticsearch instance hostname and port.

For the index operation, it might be necessary to provide or implement a `transformer`. A sample configuration would be similar to the one below:

[source,bash]
----
transforms=ElasticSearchTransformer
----

This is the sample Transformer used in the integration test code that transforms Kafka's ConnectRecord to a Map:

[source,bash]
----
transforms.ElasticSearchTransformer.type=org.apache.camel.kafkaconnector.elasticsearch.sink.transforms.ConnectRecordValueToMapTransformer
----

This is a configuration for the sample transformer that defines the key used in the map:

[source,bash]
----
transforms.ElasticSearchTransformer.key=MyKey
----

When the configuration is ready run the sink with:

.Run the Elasticsearch sink:
[source,bash]
----
$KAFKA_HOME/bin/connect-standalone.sh $KAFKA_HOME/config/connect-standalone.properties docs/examples/CamelElasticSearchSinkConnector.properties
----

[[Tryitoutlocally-FileSink]]
=== File (sink)

First thing to do, is unzip or untar the camel-aws-sqs-kafka-connector archive in the `plugin.path` location. After building the project you should have in `connectors/camel-file-kafka-connector/target/` a .zip file named `camel-file-kafka-connector-0.6.0-SNAPSHOT-package.zip`

[source,bash]
----
> cd /home/connectors/
> cp connectors/camel-file-kafka-connector/target/camel-file-kafka-connector-0.6.0-SNAPSHOT-package.zip .
> unzip camel-file-kafka-connector-0.6.0-SNAPSHOT-package.zip
----

This example appends data from `mytopic` Kafka topic to a file in `/tmp/kafkaconnect.txt`.

.Run the file sink:
[source,bash]
----
$KAFKA_HOME/bin/connect-standalone.sh $KAFKA_HOME/config/connect-standalone.properties docs/examples/CamelFileSinkConnector.properties
----

[[Tryitoutlocally-HttpSink]]
=== HTTP (sink)

First thing to do, is unzip or untar the camel-aws-sqs-kafka-connector archive in the `plugin.path` location. After building the project you should have in `connectors/camel-http-kafka-connector/target/` a .zip file named `camel-http-kafka-connector-0.6.0-SNAPSHOT-package.zip`

[source,bash]
----
> cd /home/connectors/
> cp connectors/camel-http-kafka-connector/target/camel-http-kafka-connector-0.6.0-SNAPSHOT-package.zip .
> unzip camel-http-kafka-connector-0.6.0-SNAPSHOT-package.zip
----

This example sends data from `mytopic` Kafka topic to a HTTP service. Adjust properties in `docs/examples/CamelHttpSinkConnector.properties` for your environment, for example configuring the `camel.sink.url`. 

.Run the http sink:
[source,bash]
----
$KAFKA_HOME/bin/connect-standalone.sh $KAFKA_HOME/config/connect-standalone.properties docs/examples/CamelHttpSinkConnector.properties
----

[[Tryitoutlocally-JMSSource]]
=== JMS (source)

First thing to do, is unzip or untar the camel-aws-sqs-kafka-connector archive in the `plugin.path` location. After building the project you should have in `connectors/camel-sjms2-kafka-connector/target/` a .zip file named `camel-sjsm2-kafka-connector-0.6.0-SNAPSHOT-package.zip`

[source,bash]
----
> cd /home/connectors/
> cp connectors/camel-sjsm2-kafka-connector/target/camel-sjms2-kafka-connector-0.6.0-SNAPSHOT-package.zip .
> unzip camel-sjsm2-kafka-connector-0.6.0-SNAPSHOT-package.zip
----

These are the basic connectors. For camel-sjms2 we have a bunch of provided dependencies we need to add in our path, so run the following commands:

[source,bash]
----
> cd /home/connectors/camel-sjms2-kafka-connector
> wget https://repo1.maven.org/maven2/org/apache/activemq/activemq-client/5.15.11/activemq-client-5.15.11.jar
> wget https://repo1.maven.org/maven2/org/apache/geronimo/specs/geronimo-jms_2.0_spec/1.0-alpha-2/geronimo-jms_2.0_spec-1.0-alpha-2.jar
> wget https://repo1.maven.org/maven2/org/apache/geronimo/specs/geronimo-annotation_1.0_spec/1.1.1/geronimo-annotation_1.0_spec-1.1.1.jar
> wget https://repo1.maven.org/maven2/javax/management/j2ee/management-api/1.1-rev-1/management-api-1.1-rev-1.jar
> wget https://repo1.maven.org/maven2/org/fusesource/hawtbuf/hawtbuf/1.11/hawtbuf-1.11.jar
```

This example receives messages from a JMS queue named `myqueue` and transfers them to `mytopic` Kafka topic. In this example ActiveMQ is used and it's configured to connect to the broker running on `localhost:61616`. Adjust properties in `examples/CamelJmsSourceConnector.properties` for your environment, for example configuring username and password by setting `camel.component.sjms2.connection-factory.userName=yourusername` and `camel.component.sjms2.connection-factory.password=yourpassword` or change the `camel.component.sjms2.connection-factory` and `camel.component.sjms2.connection-factory.brokerURL` to reflect your JMS implementation and URL.

.Run the JMS source:
[source,bash]
----
$KAFKA_HOME/bin/connect-standalone.sh $KAFKA_HOME/config/connect-standalone.properties docs/examples/CamelJmsSourceConnector.properties
----

[[Tryitoutlocally-JMSSink]]
=== JMS (sink)

This example receives messages from `mytopic` Kafka topic and transfers them to JMS queue named `myqueue`. In this example ActiveMQ is used and it's configured to connect to the broker running on `localhost:61616`. You can adjust properties in `examples/CamelJmsSinkConnector.properties` for your environment, for example configure username and password by adding `camel.component.sjms2.connection-factory.userName=yourusername` and `camel.component.sjms2.connection-factory.password=yourpassword` or change the `camel.component.sjms2.connection-factory` and `camel.component.sjms2.connection-factory.brokerURL` to reflect your JMS implementation and URL.

.Run the JMS sink:
[source,bash]
----
$KAFKA_HOME/bin/connect-standalone.sh $KAFKA_HOME/config/connect-standalone.properties docs/examples/CamelJmsSinkConnector.properties
----

[[Tryitoutlocally-TelegramSource]]
=== Telegram (source)

First thing to do, is unzip or untar the camel-aws-sqs-kafka-connector archive in the `plugin.path` location. After building the project you should have in `connectors/camel-telegram-kafka-connector/target/` a .zip file named `camel-telegram-kafka-connector-0.6.0-SNAPSHOT-package.zip`

[source,bash]
----
> cd /home/connectors/
> cp connectors/camel-telegram-kafka-connector/target/camel-telegram-kafka-connector-0.6.0-SNAPSHOT-package.zip .
> unzip camel-telegram-kafka-connector-0.6.0-SNAPSHOT-package.zip
----

This example transfers messages sent to Telegram bot to the `mytopic` Kafka topic. Adjust to set telegram bot token in `examples/CamelTelegramSourceConnector.properties` to reflect your bot's token.

.Run the telegram source:
[source,bash]
----
$KAFKA_HOME/bin/connect-standalone.sh $KAFKA_HOME/config/connect-standalone.properties docs/examples/CamelTelegramSourceConnector.properties
----

