> ## Documentation Index
> Fetch the complete documentation index at: https://documentation.onesignal.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Confluent Cloud

> 将Confluent Cloud的自定义事件同步到OneSignal，基于实时用户行为触发自动化旅程和个性化消息活动。

export const PLATFORM_0 = "Confluent Cloud"

export const DATA_TYPE_0 = undefined

export const COLUMN_HEADER_0 = undefined

export const PROPERTIES_DESCRIPTION_0 = undefined

## 概览

OneSignal + Confluent Cloud集成支持从您的托管Kafka主题自动同步自定义事件到OneSignal。这让您能够基于通过Confluent Cloud流平台流动的实时用户行为数据触发自动化旅程和个性化消息活动。

***

## 要求

* 访问自定义事件（目前处于测试阶段）
* [升级账户计划](https://onesignal.com/pricing)（免费应用不可用）。

### Confluent Cloud要求

* **Confluent Cloud集群**，具有活跃的主题
* **API凭据**，具有对事件主题的读取权限
* **Schema Registry**（可选，用于结构化事件架构）
* **事件主题**，包含具有合适消息格式的行为数据

***

## 设置

<Steps>
  <Step title="在Confluent Cloud中创建API凭据">
    在您的Confluent Cloud控制台中为OneSignal生成API凭据：

    1. 在Confluent Cloud中导航到**数据集成 > API密钥**
    2. 点击**创建密钥**并选择**全局访问**
    3. 保存**API密钥**和**API密钥**（您在OneSignal中需要这些）
    4. 从集群设置中记录您的**Bootstrap服务器**端点
  </Step>

  <Step title="配置主题ACL（如果使用细粒度权限）">
    向OneSignal授予对包含事件数据的特定主题的读取权限：

    ```bash theme={null}
    confluent kafka acl create \
      --allow \
      --service-account <ONESIGNAL_SERVICE_ACCOUNT_ID> \
      --operation READ \
      --topic <EVENT_TOPIC_NAME>

    confluent kafka acl create \
      --allow \
      --service-account <ONESIGNAL_SERVICE_ACCOUNT_ID> \
      --operation DESCRIBE \
      --topic <EVENT_TOPIC_NAME>
    ```
  </Step>

  <Step title="在OneSignal中添加集成">
    在 OneSignal 中，前往 **数据 > 集成**，然后点击 **添加集成**。

    选择**Confluent Cloud**并提供：

    * **Bootstrap服务器**：您的Confluent Cloud集群端点
    * **API密钥**：Confluent Cloud API密钥
    * **API密钥**：Confluent Cloud API密钥
    * **消费者组**：OneSignal的唯一组ID（例如，`onesignal-events`）
    * **Schema Registry URL**（可选）：如果使用Confluent Schema Registry
  </Step>

  <Step title="配置事件主题">
    指定包含事件数据的Confluent Cloud主题：

    * **主题名称**：要消费的主题的逗号分隔列表（例如，`user-events,purchase-events`）
    * **事件格式**：JSON、Avro或Protobuf消息格式
    * **Schema Registry**：如果使用结构化架构，请启用

    您的事件消息应包含：

    * 事件名称/类型（String）
    * 用户标识符（String）
    * 事件时间戳（Long/ISO格式）
    * 其他事件属性（嵌套JSON）
  </Step>

  <Step title="测试连接">
    点击**测试连接**以验证OneSignal能够连接到您的Confluent Cloud集群并消费事件消息。
  </Step>
</Steps>

***

### 事件数据映射

将您的 {PLATFORM_0} {DATA_TYPE_0} 映射到 OneSignal 的自定义事件格式：

| OneSignal 字段  | {COLUMN_HEADER_0} | 描述                         | 必需 |
| ------------- | ----------------- | -------------------------- | -- |
| `name`        | `event_name`      | 事件标识符                      | 是  |
| `external_id` | `user_id`         | 用户标识符                      | 是  |
| `timestamp`   | `event_timestamp` | 事件发生时间                     | 否  |
| `properties`  | `event_data`      | {PROPERTIES_DESCRIPTION_0} | 否  |

***

## 高级配置

### Schema Registry集成

利用Confluent Schema Registry处理结构化事件数据：

```json theme={null}
{
  "schema": "user_event_schema_v1",
  "data": {
    "event_name": "purchase",
    "user_id": "user_12345",
    "event_timestamp": 1640995200000,
    "properties": {
      "product_id": "prod_abc123",
      "amount": 29.99,
      "currency": "USD"
    }
  }
}
```

### 消费者组管理

OneSignal创建专用的消费者组来跟踪消息偏移量：

* **自动提交**：成功处理后自动提交偏移量
* **错误处理**：记录失败消息并具有重试机制
* **扩展**：跨OneSignal消费者实例平衡分区

### 实时处理

Confluent Cloud实现近实时事件激活：

* **低延迟**：事件在发布后几秒钟内处理
* **高吞吐量**：每秒处理数千个事件
* **故障容错**：内置复制和自动故障转移

<Warning>
  确保您的Confluent Cloud集群具有足够的吞吐量容量来处理OneSignal的消费率以及您的其他消费者。
</Warning>

***

## 常见问题

### OneSignal多久从Confluent Cloud消费一次事件？

OneSignal实时消费事件，当它们到达您的主题时，延迟很小（通常少于5秒）。

### 我可以同时从多个主题消费吗？

是的，OneSignal可以并行从多个主题消费。在配置中将主题名称指定为逗号分隔的列表。

### 如果OneSignal无法连接到Confluent Cloud会发生什么？

OneSignal将使用指数退避重试连接。一旦连接恢复，事件消费将自动恢复。