This project contains a Kafka Connect source connector for a general REST API, and one for Fitbit in particular. The documentation of the Kafka Connect REST source still needs to be done.
This repository relies on a recent version of docker and docker-compose as well as an installation of Java 17 or later.
Generally, this component is installed with RADAR-Kubernetes. It uses Docker image radarbase/kafka-connect-rest-fitbit-source.
First, register a Fitbit App with Fitbit. It should be either a server app, for multiple users, or a personal app for a single user. With the server app, you need to request access to intraday API data.
For every Fitbit user you want access to, copy docker/fitbit-user.yml.template to a file in
docker/users/. Get an access token and refresh token for the user using for example the
Fitbit OAuth 2.0 tutorial page.
For automatic configuration for multiple users, please take a look at scripts/REDCAP-FITBIT-AUTH-AUTO/README.md.
Copy docker/source-fitbit.properties.template to docker/source-fitbit.properties and enter
your Fitbit App client ID and client secret. The following tables shows the possible properties.
| Name | Description | Type | Default | Valid Values | Importance |
|---|---|---|---|---|---|
| rest.source.poll.interval.ms | How often to poll the source URL. | long | 60000 | low | |
| rest.source.base.url | Base URL for REST source connector. | string | high | ||
| rest.source.destination.topics | The list of destination topics for the REST source connector. | list | "" | high | |
| rest.source.topic.selector | The topic selector class for REST source connector. | class | org.radarbase.connect.rest.selector.SimpleTopicSelector | Class extending org.radarbase.connect.rest.selector.TopicSelector | high |
| rest.source.payload.converter.class | Class to be used to convert messages from REST calls to SourceRecords | class | org.radarbase.connect.rest.converter.StringPayloadConverter | Class extending org.radarbase.connect.rest.converter.PayloadToSourceRecordConverter | low |
| rest.source.request.generator.class | Class to be used to generate REST requests | class | org.radarbase.connect.rest.single.SingleRequestGenerator | Class extending org.radarbase.connect.rest.request.RequestGenerator | low |
| fitbit.users | The user ID of Fitbit users to include in polling, separated by commas. Non existing user names will be ignored. If empty, all users in the user directory will be used. | list | "" | high | |
| fitbit.api.client | Client ID for the Fitbit API | string | non-empty string | high | |
| fitbit.api.secret | Secret for the Fitbit API client set in fitbit.api.client. | password | high | ||
| fitbit.user.poll.interval | Polling interval per Fitbit user per request route in seconds. | int | 150 | medium | |
| fitbit.api.intraday | Set to true if the client has permissions to Fitbit Intraday API, false otherwise. | boolean | false | medium | |
| fitbit.user.repository.class | Class for managing users and authentication. | class | org.radarbase.connect.rest.fitbit.user.YamlUserRepository | Class extending org.radarbase.connect.rest.fitbit.user.UserRepository | medium |
| fitbit.user.dir | Directory containing Fitbit user information and credentials. Only used if a file-based user repository is configured. | string | /var/lib/kafka-connect-fitbit-source/users | low | |
| fitbit.user.repository.url | URL for webservice containing user credentials. Only used if a webservice-based user repository is configured. | string | "" | low | |
| fitbit.user.repository.client.id | Client ID for connecting to the service repository. | string | "" | medium | |
| fitbit.user.repository.client.secret | Client secret for connecting to the service repository. | string | "" | medium | |
| fitbit.user.repository.oauth2.token.url | OAuth 2.0 token url for retrieving client credentials. | string | "" | medium | |
| fitbit.intraday.steps.topic | Topic for Fitbit intraday steps | string | connect_fitbit_intraday_steps | non-empty string without control characters | low |
| fitbit.intraday.heart.rate.topic | Topic for Fitbit intraday heart_rate | string | connect_fitbit_intraday_heart_rate | non-empty string without control characters | low |
| fitbit.sleep.stages.topic | Topic for Fitbit sleep stages | string | connect_fitbit_sleep_stages | non-empty string without control characters | low |
| fitbit.sleep.classic.topic | Topic for Fitbit sleep classic data | string | connect_fitbit_sleep_classic | non-empty string without control characters | low |
| fitbit.time.zone.topic | Topic for Fitbit profile time zone | string | connect_fitbit_time_zone | non-empty string without control characters | low |
| fitbit.activity.log.topic | Topic for Fitbit activity log. | string | connect_fitbit_activity_log | non-empty string without control characters | low |
| fitbit.intraday.calories.topic | Topic for Fitbit intraday calories | string | connect_fitbit_intraday_calories | non-empty string without control characters | low |
| fitbit.user.firebase.collection.fitbit.name | Firestore Collection for retrieving Fitbit Auth details. Only used when a Firebase based user repository is used. | string | fitbit | low | |
| fitbit.user.firebase.collection.user.name | Firestore Collection for retrieving User details. Only used when a Firebase based user repository is used. | string | users | low |
If the ManagementPortal is used to authenticate against the user repository, please add an OAuth client to ManagementPortal with the following properties:
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
Finally set the fitbit.user.repository.oauth.token.url to http://managementportal-app:8080/managementportal/oauth/token.
Now you can run a full Kafka stack using
docker-compose up -d --buildInspect the progress with docker-compose logs -f radar-fitbit-connector. To inspect data
that is coming out of the requests, run
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_rateThe following diagrams shows the flow of the Fitbit source connector. The fitbit source connector is a Kafka Connect source connector that polls the Fitbit API for data. The data is then converted to Avro records and sent to Kafka topics.
On startup, the fitbit connector simply starts up and schedules its regular polling tasks.
sequenceDiagram
participant connector as Fitbit Source Connector
participant kafka as Kafka
connector ->> kafka: Check Kafka readiness (optional)
connector ->> connector: Schedule polling tasks
The Fitbit connector operates by regularly polling the user repository, and regularly polling all configured users for data
sequenceDiagram
participant connector as Fitbit Source Connector
participant userRepo as User Repository (rest-source-auth)
participant fitbit as Fitbit API
participant kafka as Kafka
note over connector: Get users (every 5 minutes)
connector ->> userRepo: Get users @ /users?source-type=FitBit
note over connector: For each user (every 5 seconds)
connector ->> connector: What data should be fetched?
connector ->> userRepo: Get fitbit access token @ users/<id>/token
connector ->> fitbit: Get required data @ api.fitbit.com/1/user/<id>/<data-type>/date/<daterange>
connector ->> kafka: Send data
kafka ->> connector: 200 OK
connector ->> connector: Update offset times
Code should be formatted using the Google Java Code Style Guide. If you want to contribute a feature or fix browse our issues, and please make a pull request.