Kafka Connector 故障排除¶

第 1 步：查看表的 COPY 历史记录¶

查询目标表的加载活动历史记录。有关信息，请参阅 COPY_HISTORY 视图。如果 COPY_HISTORY 输出不包含一组预期文件，请查询更早的时间段。如果文件是早期文件的副本，则在尝试加载原始文件时，加载历史记录可能已记录该活动。STATUS 列指明一组特定文件是已加载、部分加载还是加载失败。FIRST_ERROR_MESSAGE 列提供尝试部分加载或加载失败的原因。

Kafka Connector 将其无法加载的文件移动到与目标表关联的暂存区。引用表暂存区的语法为 @[namespace.]%table_name。

使用 LIST 列出位于表暂存区的所有文件。

例如：

LIST @mydb.public.%mytable;

文件名采用以下格式之一。表中描述了生成每种格式的条件：

以下各节提供了解决每种文件类型问题的说明：

第 2 步：分析 Kafka Connector 日志文件¶

如果 COPY_HISTORY 视图没有数据加载记录，则分析 Kafka Connector 的日志文件。连接器将事件写入日志文件。请注意，Snowflake Kafka Connector 与所有 Kafka Connector 插件共享相同的日志文件。此日志文件的名称和位置应位于您的 Kafka Connect 配置文件中。有关更多信息，请参阅为您的 Apache Kafka 软件提供的文档。

在 Kafka Connector 日志文件中搜索与 Snowflake 相关的错误消息。大多数消息都将使用字符串 ERROR 并包含文件名 com.snowflake.kafka.connector...，以便于查找这些消息。

您可能遇到的错误包括：

配置错误:

该错误的可能原因：

• 连接器没有订阅该主题的适当信息。

• 连接器没有写入 Snowflake 表的适当信息（例如，用于身份验证的密钥对可能出错）。

请注意，Kafka Connector 会验证其参数。对于每个不兼容的配置参数，连接器都会抛出一个错误。错误消息已写入 Kafka Connect 集群的日志文件中。如果您怀疑存在配置问题，请检查该日志文件中的错误。

读取错误:

连接器可能因以下原因无法从 Kafka 读取数据：

• Kafka 或 Kafka Connect 可能无法运行。

• 该消息可能尚未发送。

• 该消息可能已被删除（过期）。

写入错误（暂存区）:

该错误的可能原因：

• 暂存区权限不足。

• 暂存区空间不足。

• 暂存区已删除。

• 其他用户或进程向暂存区写入意外文件。

写入错误（表）:

该错误的可能原因：

• 表权限不足。

第 3 步：检查 Kafka Connect¶

如果 Kafka 连接日志文件中未报告错误，请检查 Kafka Connect。有关故障排除说明，请参阅 Apache Kafka 软件供应商提供的文档。

主题分区和偏移相同的重复行¶

使用 Kafka Connector 1.4 版本（或更高版本）加载数据时，目标表中主题分区和偏移相同的重复行可能表明加载操作超过 300000 毫秒（300 秒）的默认执行超时值。要验证原因，请检查 Kafka Connect 日志文件中是否存在以下错误：

org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member.

This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.

at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.sendOffsetCommitRequest(ConsumerCoordinator.java:1061)

要解决该错误，请在 Kafka 配置文件（例如 <kafka_dir>/config/connect-distributed.properties）中更改以下 任一 属性：

consumer.max.poll.interval.ms

将执行超时值延长至 900000 （900 秒）。

consumer.max.poll.records

将每次操作加载的记录数减少至 50。

Failure in Streaming Channel Offset Migration Response 错误代码：5023¶

连接器升级到 v2.1.0 版本（或更高版本）时，Snowpipe Streaming Channel 名称格式有所更改。因此，检测先前提交的偏移量信息的逻辑将找不到任何先前提交的偏移量信息。这将表现为以下异常：

com.snowflake.kafka.connector.internal.SnowflakeKafkaConnectorException: [SF_KAFKA_CONNECTOR] Exception: Failure in Streaming Channel Offset Migration Response Error Code: 5023

Detail: Streaming Channel Offset Migration from Source to Destination Channel has no/invalid response, please contact Snowflake Support

Message: Snowflake experienced a transient exception, please retry the migration request.

要解决此错误，请在 Kafka 配置文件（例如 <kafka_dir>/config/connect-distributed.properties）中添加以下配置属性：

enable.streaming.channel.offset.migration

通过将其设置为 false 禁用自动偏移迁移。

配置连接器以支持多个主题¶

我们在单个 Kafka Cconnector 实例支持大量主题（每个主题都有多个分区）时遇到了问题。连接器的配置虽然看似有效，却导致无休止的重新平衡循环，无法将任何数据引入 Snowflake。这个问题是 Snowpipe Streaming 引入模式（snowflake.ingestion.method=SNOWPIPE_STREAMING）所特有的，但准则也适用于 Snowpipe 引入模式（snowflake.ingestion.method=SNOWPIPE）。该问题通过反复记录此日志消息在日志文件中表现出来：

[Worker-xyz] [timestamp] INFO [my-connector|task-id] [SF_INGEST] Channel is marked as closed

当您配置连接器以通过 regex 引入主题时通常会发生这种情况。我们建议对 Kafka 配置文件应用以下选项（例如 :file : <kafka_dir>/ config/connect-distributed.properties)：

consumer.override.partition.assignment.strategy

将任务的分区分配策略配置为 org.apache.kafka.clients.consumer.CooperativeStickyAssignor – 这可以确保引入通道均匀分配到可用任务，从而降低重新平衡的风险。

tasks.max

每个连接器的实例化任务数量不应超过可用的 CPU 数量 – 底层驱动程序基于可用 CPU 实现节流机制。并发请求的增加会增加系统的内存压力，但也会延长插入处理时间，直接导致连接器新号丢失。

就连接器的超时值而言，有一组配置属性对其有直接影响：

consumer.override.heartbeat.interval.ms

定义监控线程（每个任务都有一个关联线程）向 Kafka 发送信号的频率。默认值为 3000 ms，但如果系统负载较高 – 您可以尝试将其增加到 5000 ms。

consumer.override.session.timeout.ms

定义代理在假定使用者处于无效状态并尝试重新平衡之前将等待多长时间。此设置通常应是信号间隔时间的 3 倍，因此，如果您将信号间隔时间配置为 5000 ms,请将此设置为 15000 ms。

consumer.override.max.poll.interval.ms

定义从底层 Kafka 调用 poll() 的最大时间间隔。轮询间隔的时间基本上映射到处理批量数据的连接器（包括上传到 Snowflake 并提交）。在有多个任务处理数据的场景中，底层 Snowflake Connection 可能会开始限制请求，从而导致处理时间更长。根据您的场景，可以将此值提高至长达 20 分钟 (1200000 ms） – 特别是当您通过用于引入的大量初始记录计数启动连接器时。

consumer.override.rebalance.timeout.ms

当发生重新平衡时，如果每个任务包含大量通道，每个通道都有很多底层逻辑需要确定从哪里恢复处理。此代码是顺序执行的，因此每个任务的通道越多，初始设置所需的时间就越长。将此属性配置为足够大的值，以便每个通道都能完成其初始化。3 分钟（180000 ms）是一个不错的起点。

还需要注意连接器的可用堆内存这在多个连接器同时运行或单个连接器从多个主题引入数据的场景中尤为重要。每个主题的分区映射到单个通道，因此需要内存。

请确保通过 Xmx 设置调整 Kafka Connect 进程内存设置。其中一种方法是定义 KAFKA_OPTS 环境变量并对其进行相应设置（即 KAFKA_OPTS=-Xmx4G）。

文件清理器意外清除文件¶

如果将 Kafka Connector 与 SNOWPIPE 一起使用，在从多个主题将数据引入到单个表中时您可能会遇到问题。如果您的配置没有 snowflake.topic2table.map 条目，或者主题与表之间存在 1:1 的映射，则此问题不适用。

Kafka Connector 正在生成包含上载到暂存区的记录的文件。这些文件按照以下模式格式化：snowflake_kafka_connector_<connector-name>_stage_<table-name>/<connector-name>/<table-name>/<partition-id>/<low-watermark>_<high-watermark>_<timestamp>.json.gz。问题位于 <partition-id> 中：如果多个主题将数据加载到单个表中，则 partition-id 值很可能出现重复。在正常连接器操作中，这不是问题。然而，如果连接器重新启动或重新平衡，清理进程可能会误地将已加载到暂存区（但尚未引入）的文件与错误的分区关联起来，并决定删除它们，这可能会导致数据丢失事件。

2.5.0 版本的连接器通过在 partition-id 中包含源主题的哈希码来解决此问题，确保文件名唯一且精确匹配单个主题的分区。此修复程序默认启用 – snowflake.snowpipe.stageFileNameExtensionEnabled – 并且仅影响目标表在 snowflake.topic2table.map 中被多次列出的配置。

如果您的配置受到此功能的影响，您可能会将陈旧的文件上传到暂存区。当连接器启动时，将会检查暂存区是否包含此类文件。您需要查找以 NOTE: For table 开头的日志条目，然后是检测到的文件列表。

您也可以手动检查暂存区中是否有受影响的文件：

1. 查找受影响的暂存区：

   show stages like 'snowflake_kafka_connector%<your table name>';
   

   Copy

2. 列出暂存区文件：

   list @<your stage name> pattern = '.+/<your-table-name>/[0-9]{1,4}/[0-9]+_[0-9]+_[0-9]+\.json\.gz$';
   

   Copy

上面的命令列出了所有与表的暂存区匹配并且分区 IDs 处于 0-9999 范围内的文件。这些文件将不再被引入，因此您可以下载或将其删除。