Kafka Connect REST Source and Fitbit Source

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.

Fitbit source connector

Installation

This repository relies on a recent version of docker and docker-compose as well as an installation of Java 17 or later.

Usage

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 --build

Inspect 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_rate

Flows

The 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.

Initialization

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

Loading

Regular operation

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

Loading

Contributing

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.