> ## Documentation Index
> Fetch the complete documentation index at: https://docs.arc.cdata.com/llms.txt
> Use this file to discover all available pages before exploring further.
# Apache Kafka Connector Setup
> Connect Arc to Apache Kafka for data integration and synchronization.
The Apache Kafka connector allows you to integrate Apache Kafka into your data flow by pushing or pulling data from Apache Kafka.
## Key Capabilities
* Real-time message streaming with produce and consume capabilities
* Multiple authentication schemes including SASL and Kerberos
* Topic-based message organization with consumer group management
* Configurable read duration to control how long to wait for messages to arrive
Follow the steps below to connect Arc to Apache Kafka.
## Prerequisites
To establish a connection with Apache Kafka, at a minimum users must have:
* An Apache Kafka environment that consists of a valid Apache Kafka host server and server manager (the broker, which is commonly Zookeeper or KRaft).
* Network connectivity from the Apache Kafka host to all Apache Kafka servers and server managers in the cluster containing the targeted topics.
* Topics created in Apache Kafka to send and receive messages from the server.
* A clear network path from Arc to the Apache Kafka server.
CData does not provide setup assistance or support for the creation or configuration of the Apache Kafka server or server manager. For information on the setup of a Apache Kafka environment, consult the [Apache Kafka](https://kafka.apache.org/41/) documentation.
## Establish a Connection
To allow Arc to use data from Apache Kafka, you must first establish a connection to Apache Kafka. There are two ways to establish this connection:
* Add a Apache Kafka connector to your flow. Then, in the **Settings** tab, click **Create** next to the Connection drop-down list.
* Open the Arc **Settings** page, then open the **Connections** tab. Click **Add**, select **Apache Kafka**, and click **Next**.
- The login process is only required the first time the connection is created.
- Connections to Apache Kafka can be reused across multiple Apache Kafka connectors.
### Enter Connection Settings
After opening a new connection dialogue, follow these steps:
1. Provide the requested information:
* **Name** — The static name of the connection.
* **Type** — This is always set to Apache Kafka.
* **Auth Scheme** — The authorization scheme to use for the connection. Options are **Auto**, **None**, **Plain**, **Scram**, and **Kerberos**.
* **User** — (All schemes except **None**) The Apache Kafka username to use for logging in.
* **Password** — (All schemes except **None**) The password for the user entered above.
* **Bootstrap Servers** — The host/port pair to use for establishing the initial connection to Apache Kafka. If you are connecting to Confluent Cloud, you can find this on the Cluster settings.
2. Optionally, click **Advanced** to open the drop-down menu of advanced connection settings. In most cases, you should not need these.
3. Click **Test Connection** to ensure that Arc can connect to Apache Kafka with the provided information. If an error occurs, check all fields and try again.
4. Click **Add Connection** to finalize the connection.
5. In the **Connection** drop-down list of the connector configuration pane, select the newly-created connection.
6. In the **Topic** field, enter the Apache Kafka topic that you want to target.
7. Click **Save Changes**.
There are also data source-specific authentication and configuration options on the **Advanced** tab. They are not all described in this documentation, but you can find detailed information for your data source on the [Online Help Files](https://www.cdata.com/kb/help/) page of the CData website.
## Select an Action
After establishing a connection to Apache Kafka, you must choose the action that the Apache Kafka connector should perform. The table below outlines each action and where it belongs in an Arc flow:
| Action | Description | Position in Flow |
| ------------------- | ---------------------------------------------------------------------------------------------------- | ---------------- |
| [Produce](#produce) | Accepts input data from a file or another connector and sends it to Apache Kafka. | End |
| [Consume](#consume) | Checks the queue for messages and sends any data that it gets down the flow through the Output path. | Middle |
### Produce
The **Produce** action sends input data to Apache Kafka. This data can come from other connectors or from files that you manually upload to the Input tab of the Apache Kafka connector. The Apache Kafka connector sends the input data to the topic you entered in the **Topic** field of the **Configuration** section.
#### Advanced Message Key and Partition Configuration
The Apache Kafka connector supports specifying message keys and partitions when producing messages to Kafka topics, providing granular control over message distribution and ordering in multi-partition environments.
##### Message Key Configuration
You can specify the Kafka message key using the Arc message header `MessageKey` or by using the **Key Header** field in the [Advanced Settings](#advanced-settings) section of the **Advanced** tab to tell Arc which header on the input message to use to determine the message key. This key determines which partition the message is sent to and enables message ordering guarantees within partitions.
##### Partition Configuration
You can control which partition receives your message using these optional settings:
* **Partition Header:** Defines the message header that the connector uses to set the partition. The default header name is `MessagePartition` but you can override it with any message header that contains the desired partition.
* **Partition:** Provide the partition number in the [Advanced Settings](#advanced-settings) section on the **Advanced** tab.
- If a partition header and specific partition number are both present, the message header takes precedence.
- If neither a message key nor a partition is specified, Kafka distributes messages across all partitions in the target topic.
### Consume
The **Consume** action checks for messages in the Apache Kafka queue for the topic you entered in the **Topic** field of the **Configuration** section. You must set the following fields for this action:
* **Consumer Group ID**: Specifies which group the consumers created by the connector should belong to.
* **Read Duration**: The length of time (in seconds) that the connector waits for messages to arrive. The connector waits the full duration, regardless of the number of messages received.
Data processed through the **Consume** connector goes to the Output tab and travels down to the next steps of the Arc flow.
## Additional Connection Configuration
### Advanced Tab
#### Advanced Settings
| Setting | Description |
| ------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **Partition** | The partition number to send messages to. If this and **Partition Header** are supplied, the latter takes precedence. See [Partition Configuration](#partition-configuration) for details on this and the following setting. |
| **Partition Header** | The name of the Arc message header used to specify the target partition. The default is `MessagePartition`. |
| **Key Header** | The name of the Arc message header used to specify the message key. The default is `MessageKey`. See [Message Key Configuration](#message-key-configuration) for details. |
| **Local File Scheme** | A scheme for assigning filenames to messages that are output by the connector. You can use macros in your filenames dynamically to include information such as identifiers and timestamps. For more information, see [Macros](#macros). |
| **Other Driver Settings** | Hidden properties that are used only in specific use cases. |
#### Message
The **Save to Sent Folder** and **Sent Folder Scheme** settings are deprecated and hidden by default. They appear only on connectors where they were previously enabled or configured to a non-default value. To retain copies of successfully processed files, right-click the connector in the flow, choose **Show Success Path**, and connect the success path to a [File connector](./file).
| Setting | Description |
| ---------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------ |
| **Save to Sent Folder**（非推奨） | Copies files processed by the connector to the Sent folder. Disabled by default. Files saved to the Sent folder are not covered by Encryption at Rest. |
| **Sent Folder Scheme**（非推奨） | Groups files in the Sent folder by the selected time interval. Only relevant when **Save to Sent Folder** is enabled. |
#### Logging
| Setting | Description |
| ------------------------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **Log Level** | The verbosity of logs generated by the connector. When you request support, set this to **Debug**. |
| **Log Subfolder Scheme** | Instructs the connector to group files in the Logs folder according to the selected interval. The **Weekly** option (which is the default) instructs the connector to create a new subfolder each week and store all logs for the week in that folder. Leaving this setting blank tells the connector to save all logs directly in the Logs folder. For connectors that process many transactions, using subfolders helps keep logs organized and improves performance. |
| **Log Messages** | Check this to have the log entry for a processed file include a copy of the file itself. If you disable this, you might not be able to download a copy of the file from the **Input** or **Output** tabs. |
#### Miscellaneous
*Miscellaneous settings are for specific use cases.*
| Setting | Description |
| ------------------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **Other Settings** | Enables you to configure hidden connector settings in a semicolon-separated list (for example, `setting1=value1;setting2=value2`). Normal connector use cases and functionality should not require the use of these settings. |
### Automation Tab
#### Automation Settings
*Settings related to the automatic processing of files by the connector.*
| Setting | Description |
| ------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **Send** | Whether files arriving at the connector are automatically sent. |
| **Retry Interval** | The number of minutes before a failed send is retried. |
| **Max Attempts** | The maximum number of times the connector processes the file. Success is measured based on a successful server acknowledgement. If you set this to 0, the connector retries the file indefinitely. |
| **Receive** | Whether the connector should automatically query the data source. |
| **Interval** | The interval between automatic query attempts. |
| **Minutes Past the Hour** | The minutes offset for an hourly schedule. Only applicable when the interval setting above is set to *Hourly*. For example, if this value is set to 5, the automation service downloads at 1:05, 2:05, 3:05, etc. |
| **Time** | The time of day that the attempt should occur. Only applicable when the interval setting above is set to *Daily*, *Weekly*, or *Monthly*. |
| **Day** | The day on which the attempt should occur. Only applicable when the interval setting above is set to *Weekly* or *Monthly*. |
| **Minutes** | The number of minutes to wait before attempting the download. Only applicable when the interval setting above is set to *Minute*. |
| **Cron Expression** | A five-position string representing a cron expression that determines when the attempt should occur. Only applicable when the interval setting above is set to *Advanced*. |
#### Performance
*Settings related to the allocation of resources to the connector.*
| Setting | Description |
| --------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| **Max Workers** | The maximum number of worker threads consumed from the threadpool to process files on this connector. If set, this overrides the default setting on the [Performance Settings](./Performance-Settings.html) portion of the [Advanced Settings](./Advanced-Settings.html) page. |
| **Max Files** | The maximum number of files sent by each thread assigned to the connector. If set, this overrides the default setting on the [Performance Settings](./Performance-Settings.html) portion of the [Advanced Settings](./Advanced-Settings.html) page. |
### Alerts Tab
*Settings related to configuring alerts and Service Level Agreements (SLAs).*
#### Connector Email Settings
Before you can execute SLAs, you need to set up email alerts for notifications. Clicking **Configure Alerts** opens a new browser window to the [Settings page](./Settings-Page.html) where you can set up system-wide alerts. See [Alerts](./Settings-Page.html#alerts) for more information.
#### Service Level Agreement (SLA) Settings
SLAs enable you to configure the volume you expect connectors in your flow to send or receive, and to set the time frame in which you expect that volume to be met. Arc sends emails to warn the user when an SLA is not met, and marks the SLA as *At Risk*, which means that if the SLA is not met soon, it will be marked as *Violated*. This gives the user an opportunity to step in and determine the reasons the SLA is not being met, and to take appropriate actions. If the SLA is still not met at the end of the at-risk time period, the SLA is marked as violated, and the user is notified again.
To define an SLA, click **Add Expected Volume Criteria**.
* If your connector has separate send and receive actions, use the radio buttons to specify which direction the SLA pertains to.
* Set **Expect at least** to the minimum number of transactions (the volume) you expect to be processed, then use the **Every** fields to specify the time frame.
* By default, the SLA is in effect every day. To change that, uncheck **Everyday** then check the boxes for the days of the week you want.
* Use **And set status to 'At Risk'** to indicate when the SLA should be marked as at risk.
* By default, notifications are not sent until an SLA is in violation. To change that, check **Send an 'At Risk' notification**.
The following example shows an SLA configured for a connector that expects to receive 1000 files every day Monday-Friday. An at-risk notification is sent 1 hour before the end of the time period if the 1000 files have not been received.
You can turn off SLA alerts if necessary. This can be useful during maintenance windows. Click **Settings** on the navbar, then navigate to **Alerts > General Alerts**. Click the tablet and pencil icon to edit, and uncheck the **SLA Alerts** setting.
## Macros
Using macros in file naming strategies can enhance organizational efficiency and contextual understanding of data. By incorporating macros into filenames, you can dynamically include relevant information such as identifiers, timestamps, and header information, providing valuable context to each file. This helps ensure that filenames reflect details important to your organization.
Arc supports these macros, which all use the following syntax: `%Macro%`.
| Macro | Description |
| ----------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| ConnectorID | Evaluates to the ConnectorID of the connector. |
| Ext | Evaluates to the file extension of the file currently being processed by the connector. |
| Filename | Evaluates to the filename (extension included) of the file currently being processed by the connector. |
| FilenameNoExt | Evaluates to the filename (without the extension) of the file currently being processed by the connector. |
| MessageId | Evaluates to the MessageId of the message being output by the connector. |
| RegexFilename:*pattern* | Applies a RegEx pattern to the filename of the file currently being processed by the connector. |
| Header:*headername* | Evaluates to the value of a targeted header (`headername`) on the current message being processed by the connector. |
| LongDate | Evaluates to the current datetime of the system in long-handed format (for example, Wednesday, January 24, 2024). |
| ShortDate | Evaluates to the current datetime of the system in a yyyy-MM-dd format (for example, 2024-01-24). |
| DateFormat:*format* | Evaluates to the current datetime of the system in the specified format (`format`). See [Sample Date Formats](./Date-Formatters.html#date-formats-with-literal-characters) for the available datetime formats |
| Vault:*vaultitem* | Evaluates to the value of the specified vault item. |
### Examples
Some macros, such as %Ext% and %ShortDate%, do not require an argument, but others do. All macros that take an argument use the following syntax: `%Macro:argument%`
Here are some examples of the macros that take an argument:
* %Header:headername%: Where `headername` is the name of a header on a message.
* %Header:mycustomheader% resolves to the value of the `mycustomheader` header set on the input message.
* %Header:ponum% resolves to the value of the `ponum` header set on the input message.
* %RegexFilename:pattern%: Where `pattern` is a regex pattern. For example, `%RegexFilename:^([\\w][A-Za-z]+)%` matches and resolves to the first word in the filename and is case insensitive (`test_file.xml` resolves to `test`).
* %Vault:vaultitem%: Where `vaultitem` is the name of an item in the [vault](./Global-Settings-Vault.html). For example, `%Vault:companyname%` resolves to the value of the `companyname` item stored in the vault.
* %DateFormat:format%: Where `format` is an accepted date format (see [Sample Date Formats](./Date-Formatters.html#sample-date-formats) for details). For example, `%DateFormat:yyyy-MM-dd-HH-mm-ss-fff%` resolves to the date and timestamp on the file.
You can also create more sophisticated macros, as shown in the following examples:
* Combining multiple macros in one filename: `%DateFormat:yyyy-MM-dd-HH-mm-ss-fff%%EXT%`
* Including text outside of the macro: `MyFile_%DateFormat:yyyy-MM-dd-HH-mm-ss-fff%`
* Including text within the macro: `%DateFormat:'DateProcessed-'yyyy-MM-dd_'TimeProcessed-'HH-mm-ss%`