# Kafka-Connect

The *kafka-connect* repository offers configurable source and sink kafka connectors that can be used to replicate data.

## Framework Concepts

Kafka Connect is a framework to stream data into and out of Apache Kafka. It can be used to stream data to or from commonly used systems such as relational databases, blob storages(AWS S3), message queues and other various sources and destiantions(sinks). In order to efficiently work Kafka Connect, it is helpful to establish a few major concepts which can be further read in [this document](https://docs.confluent.io/platform/current/connect/concepts.html).

- Source connectors will attach and consume(read) data from existing datastores and produce(write) kafka messages to topics.
- Sink connectors that are configured to consume(read) the messages from those topics will produce(write) data to the destination.

### Source Connector Examples

Generic Source connector configuration options: https://kafka.apache.org/24/documentation.html#sourceconnectconfigs

- [debezium_mysql_source](./debezium_mysql_source)
- [s3_source](./s3_source)

### Sinks Connector Examples

Generic Sink connector configuration options: https://kafka.apache.org/24/documentation.html#sinkconnectconfigs

- [snowflake sink](./snowflake_sink)
- [neo4j_sink](./neo4j_sink)
- [elasticsearch_sink](./elasticsearch_sink)

## Connector configuration

Each Kafka-connect connector has a set of **generic configuration properties**:
- [Sink configuration properties](https://docs.confluent.io/platform/current/installation/configuration/connect/sink-connect-configs.html#sink-configuration-properties)
- [Source configuration properties](https://docs.confluent.io/platform/current/installation/configuration/connect/source-connect-configs.html#source-configuration-properties)

Additionally there are properties defined by connector developers which usually described in documentation for particular connectors.

In most cased the above configuration will be sufficient, but there might be cases when it is not and you need to override some of the internal consumer or producer configuration properties. By default, source and sink connectors inherit their client configurations from the worker configuration. Within the worker configuration, properties that have a prefix of `producer.` or `consumer.` are used to create clients for all source and sink connectors, respectively.

If you want to override producer or consumer properties for a specific connector, enable client overrides in the worker configuration and then use `producer.override.*` for a source connector config and `consumer.override.*` for a sink connector config.

- [Producer configs](https://kafka.apache.org/documentation/#producerconfigs)
- [Consumer configs](https://kafka.apache.org/documentation/#consumerconfigs)

If there is lack of information about how those configuration work in particular prop description additional documentation can be found in apache kafka sources, e.g. for connect producer, consumer or dustributed config:
- [KafkaConsumer](https://github.com/apache/kafka/blob/2.6/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java)
- [KafkaProducer](https://github.com/apache/kafka/blob/2.6/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java)
- [DistributedConfig](https://github.com/apache/kafka/blob/2.6/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java)

> Note: If you change the connector classname or any major config.json file changes then you should delete the connector before deploying using `curl -X DELETE "https://qa-kafka-connect-s3-source-dtt.theorchard.io/connectors/s3_source"`. Only then your changes will reflect on the task info url (`/connectors/s3_source/tasks`).

## Run Kafka-Connect cluster locally

The process is pretty much straightforward.
1. Clone the repo
2. `cd` into one of the connect clusters dir (i.e. `neo4j_sink`)
3. Copy `.env.shadow` to `.env` and populate the file
4. [Generate temporary dev AWS credentials](https://github.com/theorchard/collab/tree/master/jcarrion/aws-creds-generator)
5. Enable TheOrchard VPN
6. Run `docker compose up --build`

## Terraforming Kafka Connectors

Pick your source and sink connector types and make copies of the terraform module blocks, configured for your app. Copy an existing terraform module in [terraform-infra/{env}/kafka-connect](https://github.com/theorchard/terraform-infra/tree/master/qa/kafka-infra) for your connector type. Rename the new module and update the config for your app.

## Deployments

If you are creating new connector type, prior to QA/PROD deployment you need to map connector's directory name in this repository to the service name declared in its terraform configuration in the [Jenkinsfile](https://github.com/theorchard/kafka-connect/blob/master/Jenkinsfile#L1).

Add your service to be deployed in [Jenkinsfile](https://github.com/theorchard/kafka-connect/blob/e5053c9d8e5eacfa9a5d56830bd4d3338f3918a0/Jenkinsfile#L112) with configuration details like 

**Configuration values:**

 - project (required): The project that this service is based on. This must map to one of the projects defined in the PROJECTS mapping above.
 - containerName (optional): The name of the main container in the task. Defaults to service name.
 - deployToProd (optional): Whether or not to deploy to prod. This takes precedence over the DEPLOY_TO_PROD parameter.
 - integrationTestJob (optional): The name of an integration test job to run for this service.
 - integrationTestJobParameters (optional): A list of parameters to pass to the integration test job.
 - prodAccount (optional): The name of the production account for this service. This must map to one of the accounts defined in the ACCOUNTS mapping above.
 - qaAccount (optional): The name of the QA account for this service. This must map to one of the accounts defined in the ACCOUNTS mapping above.
 
 
## Conventions

- `${CONNECTOR_NAME}`: this should be datasource*name + underscore + connector_type, e.g. \_artrelations_sink*, _snowflake_source_
- Set defaults for environment variables in Dockerfile
- Add a docker-compose.yml file that provides a working configuration for local development
- Add a .env.shadow file, listing environment variables that should be overridden specifically during local development
- .env files should not be loaded in QA, Prod, or in AWS dev. Instead, environment variables should be specified in the task or worker definition at runtime
- Connectors should be written in such a way so that we can safely run multiple instances of it with different settings, e.g. different bucket/topic combinations in the case of an S3 Sink connector.
- Only one connector, and moreover one instance of that connector, should run in a container. Do not pack multiple connectors in a single task or container. If you have two S3 Sink connectors with different bucket/topic combinations, they should run as separate containers/tasks.
- Connect tasks should run in headless mode wherever possible to ensure immutability.
- Make sure `${CONNECT_GROUP_ID}` is unique.
- By default, when updating connector configurations, output is suppressed; to view the raw curl output, set the environment variable `DEBUG_MODE` to a non-empty value.

## Credentials

Credentials should not be stored in plaintext wherever possible. Generally speaking, connector support falls into 3 categories, since not all support our preferred method. The implementations involve, in order of our preference and security:

- Use the [Lenses AWS secrets plugin/provider](https://github.com/lensesio/secret-provider) to dynamically pull secrets from aws secrets manager, store them as references, and do not persist them in Kafka topics. This is the preferred approach where supported.

  - Example: https://github.com/theorchard/kafka-connect/blob/master/elasticsearch_sink/entry_point.sh#L6-L8 + https://github.com/theorchard/kafka-connect/blob/master/elasticsearch_sink/entry_point.sh#L35-L45

- Use the awscli to pull secrets from aws secrets manager, write them to a file at runtime, use the kafka [file config provider](https://docs.confluent.io/current/connect/security.html#fileconfigprovider) to reference this file, and do not persist them in Kafka topics.
  - Example: https://github.com/theorchard/kafka-connect/blob/master/debezium_mysql_source/entry_point.sh#L34-L42 + https://github.com/theorchard/kafka-connect/blob/master/debezium_mysql_source/entry_point.sh#L98-L101
- Use the Fargate secrets block to provide values from aws secrets manager as runtime environment variables. While this works fine from the container image and code security perspective, these are unfortunately persisted in config messages in Kafka topics. This should only be used in situations where the connector does not support either of the previous approaches.
  - Example: https://github.com/theorchard/terraform-infra/blob/master/dev/kafka-infra/snowflake_sink/main.tf#L73-L80 + https://github.com/theorchard/kafka-connect/blob/master/snowflake_sink/entry_point.sh#L49-L53

## Entrypoint

Connectors should have an entrypoint script that wraps the run command, sets conditional runtime parameters (e.g. if qa, KEY=VALUE) and programmatically sets values that can be automated (e.g. `${CONNECT_GROUP_ID}`)

## Using AVRO serialization

You need to configure your Kafka connector to connect to the schema registry, if you want to start using AVRO serialization.
Please, take a look at [snowflake sink](./snowflake_sink) connector for an example.
The following environment variables are required:
  - `CONNECT_VALUE_CONVERTER=io.confluent.connect.avro.AvroConverter` - specify AVRO value converter.
  - `CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL=https://dev-schema-registry.dev.theorchard.io` - schema registry URL.
Currently it is only deployed on DEV environment.
  - `SCHEMA_REGISTRY_AUTH=false` - (optional) schema registry authentication is currently disabled by default.
  - `SCHEMA_REGISTRY_CREDENTIALS=user:password` - (optional) credentials are not currently used.