Kafka Connect REST源连接器与Fitbit源连接器

本项目包含Kafka Connect的REST API通用源连接器，以及专门用于Fitbit的源连接器。目前REST源连接器的文档仍在完善中，以下主要介绍Fitbit源连接器的使用。

Fitbit源连接器

概述

Fitbit源连接器是Kafka Connect的源组件，用于通过Fitbit API将Fitbit用户的生理和活动数据（如心率、步数、睡眠等）接入Kafka，支持单用户（个人应用）和多用户（服务器应用）场景，提供灵活的配置和数据路由能力。

核心功能

支持Fitbit OAuth 2.0认证，自动管理access token和refresh token
适配单用户（个人应用）与多用户（服务器应用）数据采集
采集多种Fitbit数据类型：日内步数、心率、卡路里，睡眠阶段，活动日志，用户时区等
可自定义Kafka主题，实现不同类型数据的分离存储
支持文件型、Web服务型（如ManagementPortal）或Firebase用户仓库，灵活管理用户认证信息

使用场景

适用于需要将Fitbit用户健康数据接入Kafka的场景，如健康监测平台、科研数据分析系统、用户行为分析平台等，尤其适合需要批量处理多用户Fitbit数据的应用。

安装要求

运行该连接器需满足：

Docker及docker-compose（最新版本）
Java 11或更高版本

使用方法

1. 注册Fitbit应用

访问Fitbit开发者平台注册应用：
- 多用户场景选择服务器应用，单用户场景选择个人应用
- 服务器应用需额外申请日内API数据访问权限（用于获取高频数据）

2. 配置用户认证

手动配置用户（单用户/少量用户）

复制用户配置模板：将项目中的docker/fitbit-user.yml.template复制到docker/users/目录，重命名为<用户名>.yml
获取用户令牌：通过Fitbit OAuth 2.0交互式教程获取用户的access token和refresh token，填入上述文件

自动配置用户（多用户）

参考scripts/REDCAP-FITBIT-AUTH-AUTO/README.md中的自动化认证流程，支持批量用户配置

3. 配置连接器属性

复制属性模板：将docker/source-fitbit.properties.template复制为docker/source-fitbit.properties
填写应用凭证：在配置文件中填入Fitbit应用的client ID和client secret
（可选）调整其他配置参数，参数说明如下表：

名称	描述	类型	默认值	有效值	重要性
rest.source.poll.interval.ms	轮询源URL的频率（毫秒）	long	60000	-	低
rest.source.base.url	REST源连接器的基础URL	string	-	-	高
rest.source.destination.topics	REST源连接器的目标主题列表	list	""	-	高
rest.source.topic.selector	REST源连接器的主题选择器类	class	org.radarbase.connect.rest.selector.SimpleTopicSelector	继承自org.radarbase.connect.rest.selector.TopicSelector的类	高
rest.source.payload.converter.class	REST响应转SourceRecords的转换器类	class	org.radarbase.connect.rest.converter.StringPayloadConverter	继承自org.radarbase.connect.rest.converter.PayloadToSourceRecordConverter的类	低
rest.source.request.generator.class	REST请求生成器类	class	org.radarbase.connect.rest.single.SingleRequestGenerator	继承自org.radarbase.connect.rest.request.RequestGenerator的类	低
fitbit.users	需轮询的Fitbit用户ID列表（逗号分隔，空则包含用户目录所有用户）	list	""	-	高
fitbit.api.client	Fitbit API的客户端ID	string	-	非空字符串	高
fitbit.api.secret	Fitbit API客户端的密钥	password	-	-	高
fitbit.user.poll.interval	每个用户每个请求路由的轮询间隔（秒）	int	150	-	中
fitbit.api.intraday	是否有权限访问Fitbit日内API	boolean	false	-	中
fitbit.user.repository.class	用户认证管理类	class	org.radarbase.connect.rest.fitbit.user.YamlUserRepository	继承自org.radarbase.connect.rest.fitbit.user.UserRepository的类	中
fitbit.user.dir	文件型用户仓库的用户凭证目录	string	/var/lib/kafka-connect-fitbit-source/users	-	低
fitbit.user.repository.url	Web服务型用户仓库的URL	string	""	-	低
fitbit.user.repository.client.id	Web服务用户仓库的客户端ID	string	""	-	中
fitbit.user.repository.client.secret	Web服务用户仓库的客户端密钥	string	""	-	中
fitbit.user.repository.oauth2.token.url	Web服务用户仓库的OAuth 2.0令牌URL	string	""	-	中
fitbit.intraday.steps.topic	日内步数数据的Kafka主题	string	connect_fitbit_intraday_steps	无控制字符的非空字符串	低
fitbit.intraday.heart.rate.topic	日内心率数据的Kafka主题	string	connect_fitbit_intraday_heart_rate	无控制字符的非空字符串	低
fitbit.sleep.stages.topic	睡眠阶段数据的Kafka主题	string	connect_fitbit_sleep_stages	无控制字符的非空字符串	低
fitbit.sleep.classic.topic	传统睡眠数据的Kafka主题	string	connect_fitbit_sleep_classic	无控制字符的非空字符串	低
fitbit.time.zone.topic	用户时区数据的Kafka主题	string	connect_fitbit_time_zone	无控制字符的非空字符串	低
fitbit.activity.log.topic	活动日志数据的Kafka主题	string	connect_fitbit_activity_log	无控制字符的非空字符串	低
fitbit.intraday.calories.topic	日内卡路里数据的Kafka主题	string	connect_fitbit_intraday_calories	无控制字符的非空字符串	低
fitbit.user.firebase.collection.fitbit.name	Firebase用户仓库的Fitbit认证集合名	string	fitbit	-	低
fitbit.user.firebase.collection.user.name	Firebase用户仓库的用户信息集合名	string	users	-	低

4. 集成ManagementPortal用户仓库（可选）

若使用ManagementPortal管理用户，需在ManagementPortal中创建OAuth客户端，配置如下：

Client ID: fitbit.user.repository.client.id
Client Secret: fitbit.user.repository.client.secret
Scope: SUBJECT.READ MEASUREMENT.CREATE
Resources: res_restAuthorizer
Grant types: client_credentials
Access Token validity: 600
Refresh Token validity: 0

并设置fitbit.user.repository.oauth.token.url为[***]

5. 启动与验证

启动服务

shell
docker-compose up -d --build

查看运行日志

shell
docker-compose logs -f radar-fitbit-connector

验证数据接入

通过Kafka Avro控制台消费者查看数据（以心率数据为例）：

shell
docker-compose exec schema-registry-1 kafka-avro-console-consumer \
  --bootstrap-server kafka-1:9092,kafka-2:9092,kafka-3:9092 \
  --from-beginning \
  --topic connect_fitbit_intraday_heart_rate

（将主题名替换为实际需要验证的数据类型，如connect_fitbit_intraday_steps表示步数数据）