Platform: Kafka source and destination connectors (#421)

Paul-Cornell · web-flow · commit 0373070d8cea · 2025-01-30T09:52:43.000-08:00
diff --git a/mint.json b/mint.json
@@ -487,6 +487,7 @@
                 "platform/sources/elasticsearch",
                 "platform/sources/google-cloud",
                 "platform/sources/google-drive",
+                "platform/sources/kafka",
                 "platform/sources/mongodb",
                 "platform/sources/onedrive",
                 "platform/sources/outlook",
@@ -507,6 +508,7 @@
               "platform/destinations/delta-table",
               "platform/destinations/elasticsearch",
               "platform/destinations/google-cloud",
+              "platform/destinations/kafka",
               "platform/destinations/milvus",
               "platform/destinations/mongodb",
               "platform/destinations/neo4j",
@@ -536,6 +538,7 @@
                   "platform/api/sources/elasticsearch",
                   "platform/api/sources/google-cloud",
                   "platform/api/sources/google-drive",
+                  "platform/api/sources/kafka",
                   "platform/api/sources/mongodb",
                   "platform/api/sources/onedrive",
                   "platform/api/sources/outlook",
@@ -556,6 +559,7 @@
                   "platform/api/destinations/delta-table",
                   "platform/api/destinations/elasticsearch",
                   "platform/api/destinations/google-cloud",
+                  "platform/api/destinations/kafka",
                   "platform/api/destinations/milvus",
                   "platform/api/destinations/mongodb",
                   "platform/api/destinations/neo4j",
diff --git a/platform/api/destinations/kafka.mdx b/platform/api/destinations/kafka.mdx
@@ -0,0 +1,30 @@
+---
+title: Kafka
+---
+
+Send processed data from Unstructured to Kafka.
+
+The requirements are as follows.
+
+import KafkaPrerequisites from '/snippets/general-shared-text/kafka.mdx';
+
+<KafkaPrerequisites />
+
+To create or change a Kafka destination connector, see the following examples.
+
+import KafkaAPIRESTCreate from '/snippets/destination_connectors/kafka_rest_create.mdx';
+import KafkaAPIRESTChange from '/snippets/destination_connectors/kafka_rest_change.mdx';
+
+<CodeGroup>
+    <KafkaAPIRESTCreate />
+    <KafkaAPIRESTChange />
+</CodeGroup>
+
+Replace the preceding placeholders as follows:
+
+import KafkaAPIPlaceholders from '/snippets/general-shared-text/kafka-api-placeholders.mdx';
+
+<KafkaAPIPlaceholders />
+
+To change a connector, replace `<connector-id>` with the source connector's unique ID. 
+To get this ID, see [List source connectors](/platform/api/overview#list-source-connectors).
diff --git a/platform/api/destinations/overview.mdx b/platform/api/destinations/overview.mdx
@@ -20,6 +20,7 @@ For the list of specific settings, see:
 - [Delta Table](/platform/api/destinations/delta-table) (`destination_type=delta_table`)
 - [Elasticsearch](/platform/api/destinations/elasticsearch) (`destination_type=elasticsearch`)
 - [Google Cloud Storage](/platform/api/destinations/google-cloud) (`destination_type=gcs`)
+- [Kafka](/platform/api/destinations/kafka) (`destination_type=kafka-cloud`)
 - [Milvus](/platform/api/destinations/milvus) (`destination_type=milvus`)
 - [MongoDB](/platform/api/destinations/mongodb) (`destination_type=mongodb`)
 - [Neo4j](/platform/api/destinations/neo4j) (`destination_type=neo4j`)
diff --git a/platform/api/sources/kafka.mdx b/platform/api/sources/kafka.mdx
@@ -0,0 +1,30 @@
+---
+title: Kafka
+---
+
+Ingest your files into Unstructured from Kafka.
+
+The requirements are as follows.
+
+import KafkaPrerequisites from '/snippets/general-shared-text/kafka.mdx';
+
+<KafkaPrerequisites />
+
+To create or change a Kafka source connector, see the following examples.
+
+import KafkaAPIRESTCreate from '/snippets/source_connectors/kafka_rest_create.mdx';
+import KafkaAPIRESTChange from '/snippets/source_connectors/kafka_rest_change.mdx';
+
+<CodeGroup>
+    <KafkaAPIRESTCreate />
+    <KafkaAPIRESTChange />
+</CodeGroup>
+
+Replace the preceding placeholders as follows:
+
+import KafkaAPIPlaceholders from '/snippets/general-shared-text/kafka-api-placeholders.mdx';
+
+<KafkaAPIPlaceholders />
+
+To change a connector, replace `<connector-id>` with the source connector's unique ID. 
+To get this ID, see [List source connectors](/platform/api/overview#list-source-connectors).
diff --git a/platform/api/sources/overview.mdx b/platform/api/sources/overview.mdx
@@ -21,6 +21,7 @@ For the list of specific settings, see:
 - [Elasticsearch](/platform/api/sources/elasticsearch) (`source_type=elasticsearch`)
 - [Google Cloud Storage](/platform/api/sources/google-cloud) (`source_type=gcs`)
 - [Google Drive](/platform/api/sources/google-drive) (`source_type=google_drive`)
+- [Kafka](/platform/api/sources/kafka) (`source_type=kafka-cloud`)
 - [MongoDB](/platform/api/sources/mongodb) (`source_type=mongodb`)
 - [OneDrive](/platform/api/sources/onedrive) (`source_type=onedrive`)
 - [Outlook](/platform/api/sources/outlook) (`source_type=outlook`)
diff --git a/platform/connectors.mdx b/platform/connectors.mdx
@@ -19,6 +19,7 @@ The Unstructured Platform supports connecting to the following source and destin
 - [Elasticsearch](/platform/sources/elasticsearch)
 - [Google Cloud Storage](/platform/sources/google-cloud)
 - [Google Drive](/platform/sources/google-drive)
+- [Kafka](/platform/sources/kafka)
 - [MongoDB](/platform/sources/mongodb)
 - [OneDrive](/platform/sources/onedrive)
 - [Outlook](/platform/sources/outlook)
@@ -41,6 +42,7 @@ If your source is not listed here, you might still be able to connect Unstructur
 - [Delta Table](/platform/destinations/delta-table)
 - [Elasticsearch](/platform/destinations/elasticsearch)
 - [Google Cloud Storage](/platform/destinations/google-cloud)
+- [Kafka](/platform/destinations/kafka)
 - [Milvus](/platform/destinations/milvus)
 - [MongoDB](/platform/destinations/mongodb)
 - [Neo4j](/platform/destinations/neo4j)
diff --git a/platform/destinations/kafka.mdx b/platform/destinations/kafka.mdx
@@ -0,0 +1,26 @@
+---
+title: Kafka
+---
+
+Send processed data from Unstructured to Kafka.
+
+The requirements are as follows.
+
+import KafkaPrerequisites from '/snippets/general-shared-text/kafka.mdx';
+
+<KafkaPrerequisites />
+
+To create the destination connector:
+
+1. On the sidebar, click **Connectors**.
+2. Click **Destinations**.
+3. Cick **New** or **Create Connector**.
+4. Give the connector some unique **Name**.
+5. In the **Provider** area, click **Kafka**.
+6. Click **Continue**. 
+7. Follow the on-screen instructions to fill in the fields as described later on this page.
+8. Click **Save and Test**.
+
+import KafkaFields from '/snippets/general-shared-text/kafka-platform.mdx';
+
+<KafkaFields />
diff --git a/platform/destinations/overview.mdx b/platform/destinations/overview.mdx
@@ -24,6 +24,7 @@ To create a destination connector:
    - [Delta Table](/platform/destinations/delta-table)
    - [Elasticsearch](/platform/destinations/elasticsearch)
    - [Google Cloud Storage](/platform/destinations/google-cloud)
+   - [Kafka](/platform/destinations/kafka)
    - [Milvus](/platform/destinations/milvus)
    - [MongoDB](/platform/destinations/mongodb)
    - [Neo4j](/platform/destinations/neo4j)
diff --git a/platform/sources/kafka.mdx b/platform/sources/kafka.mdx
@@ -0,0 +1,26 @@
+---
+title: Kafka
+---
+
+Ingest your files into Unstructured from Kafka.
+
+The requirements are as follows. 
+
+import KafkaPrerequisites from '/snippets/general-shared-text/kafka.mdx';
+
+<KafkaPrerequisites />
+
+To create the source connector:
+
+1. On the sidebar, click **Connectors**.
+2. Click **Sources**.
+3. Cick **New** or **Create Connector**.
+4. Give the connector some unique **Name**.
+5. In the **Provider** area, click **Kafka**.
+6. Click **Continue**. 
+7. Follow the on-screen instructions to fill in the fields as described later on this page.
+8. Click **Save and Test**.
+
+import KafkaFields from '/snippets/general-shared-text/kafka-platform.mdx';
+
+<KafkaFields />
diff --git a/platform/sources/overview.mdx b/platform/sources/overview.mdx
@@ -26,6 +26,7 @@ To create a source connector:
    - [Elasticsearch](/platform/sources/elasticsearch)
    - [Google Cloud Storage](/platform/sources/google-cloud)
    - [Google Drive](/platform/sources/google-drive)
+   - [Kafka](/platform/sources/kafka)
    - [MongoDB](/platform/sources/mongodb)
    - [OneDrive](/platform/sources/onedrive)
    - [Outlook](/platform/sources/outlook)
diff --git a/snippets/destination_connectors/kafka_rest_change.mdx b/snippets/destination_connectors/kafka_rest_change.mdx
@@ -0,0 +1,19 @@
+```bash REST (Change)
+curl --request 'PUT' --location \
+"$UNSTRUCTURED_API_URL/destinations/<connector-id>" \
+--header 'accept: application/json' \
+--header "unstructured-api-key: $UNSTRUCTURED_API_KEY" \
+--header 'content-type: application/json' \
+--data \
+'{
+    "config": {
+        "bootstrap_server": "<bootstrap-server>",
+        "port": <port>,
+        "group_id": "<group-id>",
+        "kafka_api_key": "<kafka-api-key>",
+        "secret": "<secret>",
+        "topic": "<topic>",
+        "num_messages_to_consume": <num-messages-to-consume>
+    },
+}'
+```
diff --git a/snippets/destination_connectors/kafka_rest_create.mdx b/snippets/destination_connectors/kafka_rest_create.mdx
@@ -0,0 +1,20 @@
+```bash REST (Create)
+curl --request 'POST' --location \
+"$UNSTRUCTURED_API_URL/destinations" \
+--header 'accept: application/json' \
+--header "unstructured-api-key: $UNSTRUCTURED_API_KEY" \
+--header 'content-type: application/json' \
+--data \
+'{
+    "name": "<name>",
+    "type": "kafka-cloud",
+    "config": {
+        "bootstrap_server": "<bootstrap-server>",
+        "port": <port>,
+        "group_id": "<group-id>",
+        "kafka_api_key": "<kafka-api-key>",
+        "secret": "<secret>",
+        "topic": "<topic>",
+        "num_messages_to_consume": <num-messages-to-consume>
+    },
+}'
diff --git a/snippets/general-shared-text/kafka-api-placeholders.mdx b/snippets/general-shared-text/kafka-api-placeholders.mdx
@@ -0,0 +1,8 @@
+- `<name>` (_required_) - A unique name for this connector.
+- `<bootstrap-server>` - The hostname of the bootstrap Kafka cluster to connect to.
+- `<port>` - The port number of the bootstrap Kafka cluster to connect to. The default is `9092` if not otherwise specified.
+- `<group-id>` - The ID of the consumer group. A consumer group is a way to allow a pool of consumers to divide the consumption of data over topics and partitions. The default is `default_group_id` if not otherwise specified.
+- `<kafka-api-key>` - For authentication, the API key for access to the cluster.
+- `<secret>` - For authentication, the secret for access to the cluster.
+- `<topic>` - The name of the topic to read messages from or write messages to on the cluster.
+- `<num-messages-to-consume>` - The maximum number of messages that the consumer will try to consume. The default is `100` if not otherwise specified.
diff --git a/snippets/general-shared-text/kafka-cli-api.mdx b/snippets/general-shared-text/kafka-cli-api.mdx
@@ -24,6 +24,6 @@ Additional settings include:
 - `--confluent` (CLI) or `confluent` (Python): True to indicate that the cluster is running Confluent Kafka.
 - `--num-messages-to-consume` (CLI) or `num_messages_to_consume` (Python): The maximum number of messages to get from the topic. The default is `1` if not otherwise specified.
 - `--timeout` (CLI) or `timeout` (Python): The maximum amount of time to wait for the response of a request to the topic, expressed in seconds. The default is `1.0` if not otherwise specified.
-- `--group-id` (CLI) or `group_id` (Python): The ID of the consumer group, if any, that is associated with the target Kafka cluser. 
+- `--group-id` (CLI) or `group_id` (Python): The ID of the consumer group, if any, that is associated with the target Kafka cluster. 
   (A consumer group is a way to allow a pool of consumers to divide the consumption of data 
   over topics and partitions.) The default is `default_group_id` if not otherwise specified.
diff --git a/snippets/general-shared-text/kafka-platform.mdx b/snippets/general-shared-text/kafka-platform.mdx
@@ -0,0 +1,13 @@
+Fill in the following fields:
+
+- **Name** (_required_): A unique name for this connector.
+- **Bootstrap Server** (_required_): The hostname of the bootstrap Kafka cluster to connect to.
+- **Port**: The port number of the cluster.
+- **Group ID**: The ID of the consumer group, if any, that is associated with the target Kafka cluster. 
+  (A consumer group is a way to allow a pool of consumers to divide the consumption of data 
+  over topics and partitions.) The default is `default_group_id` if not otherwise specified.
+- **Topic** (_required_): The unique name of the topic to read messages from and write messages to on the cluster.
+- **Number of messages to consume**: The maximum number of messages to get from the topic. The default is `100` if not otherwise specified.
+- **Batch Size**: The maximum number of messages to send in a single batch. The default is `100` if not otherwise specified.
+- **API Key** (_required_): The Kafka API key value.
+- **Secret** (_required_): The secret value for the Kafka API key.
diff --git a/snippets/general-shared-text/kafka.mdx b/snippets/general-shared-text/kafka.mdx
@@ -1,7 +1,7 @@
-- A Kafka cluster, available through providers such as [Confluent Cloud](https://www.confluent.io/confluent-cloud).
+- A Kafka cluster in [Confluent Cloud](https://www.confluent.io/confluent-cloud). 
+  ([Create a cluster](https://docs.confluent.io/cloud/current/clusters/create-cluster.html#create-ak-clusters).)
 
-  The following video shows how to 
-  [create a Kafka cluster in Confluent Cloud](https://docs.confluent.io/cloud/current/get-started/index.html#step-1-create-a-ak-cluster-in-ccloud):
+  The following video shows how to set up a Kafka cluster in Confluent Cloud:
 
   <iframe
   width="560"
@@ -13,10 +13,8 @@
   allowfullscreen
   ></iframe>
 
-- The [hostname and port number](https://docs.confluent.io/cloud/current/faq.html#how-do-i-view-cluster-details-with-ccloud-console-short) 
-  of the bootstrap Kafka server to connect to for the cluster. Note that a colon (`:`) separates the hostname from the port number.
-  [Learn more](https://docs.confluent.io/cloud/current/clusters/create-cluster.html#view-ak-clusters).
-- The name of the [topic](https://docs.confluent.io/cloud/current/get-started/index.html#step-2-create-a-ak-topic) 
-  to read messages from and write messages to on the cluster. 
-- The [API key and secret](https://docs.confluent.io/cloud/current/security/authenticate/workload-identities/service-accounts/api-keys/manage-api-keys.html#add-an-api-key) 
-  for the cluster.
+- The [hostname and port number](https://docs.confluent.io/cloud/current/clusters/create-cluster.html#view-a-ak-cluster) of the bootstrap Kafka cluster to connect to..
+- The name of the topic to read messages from or write messages to on the cluster. 
+  [Create a topic](https://docs.confluent.io/cloud/current/client-apps/topics/index.html#create-topics). 
+  [Access available topics](https://docs.confluent.io/cloud/current/client-apps/topics/index.html#create-topics).
+- For authentication, an [API key and secret](https://docs.confluent.io/cloud/current/security/authenticate/workload-identities/service-accounts/api-keys/manage-api-keys.html#add-an-api-key).
diff --git a/snippets/source_connectors/kafka_rest_change.mdx b/snippets/source_connectors/kafka_rest_change.mdx
@@ -0,0 +1,19 @@
+```bash REST (Change)
+curl --request 'PUT' --location \
+"$UNSTRUCTURED_API_URL/sources/<connector-id>" \
+--header 'accept: application/json' \
+--header "unstructured-api-key: $UNSTRUCTURED_API_KEY" \
+--header 'content-type: application/json' \
+--data \
+'{
+    "config": {
+        "bootstrap_server": "<bootstrap-server>",
+        "port": <port>,
+        "group_id": "<group-id>",
+        "kafka_api_key": "<kafka-api-key>",
+        "secret": "<secret>",
+        "topic": "<topic>",
+        "num_messages_to_consume": <num-messages-to-consume>
+    },
+}'
+```
diff --git a/snippets/source_connectors/kafka_rest_create.mdx b/snippets/source_connectors/kafka_rest_create.mdx
@@ -0,0 +1,21 @@
+```bash REST (Create)
+curl --request 'POST' --location \
+"$UNSTRUCTURED_API_URL/sources" \
+--header 'accept: application/json' \
+--header "unstructured-api-key: $UNSTRUCTURED_API_KEY" \
+--header 'content-type: application/json' \
+--data \
+'{
+    "name": "<name>",
+    "type": "kafka-cloud",
+    "config": {
+        "bootstrap_server": "<bootstrap-server>",
+        "port": <port>,
+        "group_id": "<group-id>",
+        "kafka_api_key": "<kafka-api-key>",
+        "secret": "<secret>",
+        "topic": "<topic>",
+        "num_messages_to_consume": <num-messages-to-consume>
+    },
+}'
+```
