Merge pull request #43 from RADAR-base/release-0.2.2

blootsvoets · web-flow · commit 6842fa5efe35 · 2019-06-06T11:30:08.000+02:00
Release 0.2.2
diff --git a/README.md b/README.md
@@ -61,8 +61,6 @@ your Fitbit App client ID and client secret. The following tables shows the poss
 <tr>
 <td>fitbit.user.repository.url</td></td><td>URL for webservice containing user credentials. Only used if a webservice-based user repository is configured.</td></td><td>string</td></td><td>""</td></td><td></td></td><td>low</td></td></tr>
 <tr>
-<td>fitbit.max.users.per.poll</td></td><td>Maximum number of users to query in a single poll operation. Decrease this if memory constrains are pressing.</td></td><td>int</td></td><td>100</td></td><td>[1,...]</td></td><td>low</td></td></tr>
-<tr>
 <td>fitbit.intraday.steps.topic</td></td><td>Topic for Fitbit intraday steps</td></td><td>string</td></td><td>connect_fitbit_intraday_steps</td></td><td>non-empty string without control characters</td></td><td>low</td></td></tr>
 <tr>
 <td>fitbit.intraday.heart.rate.topic</td></td><td>Topic for Fitbit intraday heart_rate</td></td><td>string</td></td><td>connect_fitbit_intraday_heart_rate</td></td><td>non-empty string without control characters</td></td><td>low</td></td></tr>
diff --git a/build.gradle b/build.gradle
@@ -2,16 +2,16 @@ description = 'kafka-connect-rest-source'
 
 subprojects {
     ext {
-        kafkaVersion = '2.1.0-cp1'
-        confluentVersion = '5.1.0'
-        jacksonVersion = '2.9.8'
+        kafkaVersion = '2.2.0-cp2'
+        confluentVersion = '5.2.1'
+        jacksonVersion = '2.9.9'
     }
 
     apply plugin: 'java'
     apply plugin: 'java-library'
 
     group = 'org.radarcns'
-    version = '0.2.1'
+    version = '0.2.2'
 
     sourceCompatibility = 1.8
     targetCompatibility = 1.8
@@ -26,7 +26,7 @@ subprojects {
 }
 
 wrapper {
-    gradleVersion '4.10.2'
+    gradleVersion '5.4.1'
 }
 
 evaluationDependsOnChildren()
diff --git a/docker/rest-source-authorizer/docker-compose.yml b/docker/rest-source-authorizer/docker-compose.yml
@@ -0,0 +1,34 @@
+---
+version: '2.4'
+
+services:
+  radar-rest-source-auth-backend:
+    image: radarbase/radar-rest-source-auth-backend:1.0.0
+    depends_on:
+      - auth-postgresql
+    environment:
+      - SPRING_DATASOURCE_URL=jdbc:postgresql://auth-postgresql:5432/restsourceauth
+      - SPRING_DATASOURCE_USERNAME=radarcns
+      - SPRING_DATASOURCE_PASSWORD=radarcns
+      - REST_SOURCE_AUTHORIZER_SOURCE_CLIENTS_FILE_PATH=app-includes/rest_source_clients_configs.yml
+      - LOGGING_LEVEL_ORG_SPRINGFRAMEWORK_WEB=DEBUG
+      - APP_SLEEP=10 # gives time for the database to boot before the application
+    ports:
+      - "8080:8080"
+    volumes:
+      - ./etc:/app-includes
+
+  auth-postgresql:
+    image: postgres:10.4-alpine
+    environment:
+      - POSTGRES_USER=radarcns
+      - POSTGRES_PASSWORD=radarcns
+      - POSTGRES_DB=restsourceauth
+
+  radar-rest-source-authorizer:
+    image: radarbase/radar-rest-source-authorizer:1.0.0
+    depends_on:
+      - radar-rest-source-auth-backend
+      - auth-postgresql
+    ports:
+      - "3000:80"
diff --git a/docker/rest-source-authorizer/etc/rest_source_clients_configs.yml.template b/docker/rest-source-authorizer/etc/rest_source_clients_configs.yml.template
@@ -0,0 +1,7 @@
+rest_source_clients:
+  - source_type: FitBit
+    authorization_endpoint: https://www.fitbit.com/oauth2/authorize
+    token_endpoint: https://api.fitbit.com/oauth2/token
+    client_id: FITBITIT
+    client_secret: FITBITSECRET
+    scope: activity heartrate sleep profile
diff --git a/docker/source-fitbit.properties.template b/docker/source-fitbit.properties.template
@@ -6,6 +6,5 @@ rest.source.poll.interval.ms=5000
 rest.source.request.generator.class=org.radarbase.connect.rest.fitbit.request.FitbitRequestGenerator
 fitbit.api.client=?
 fitbit.api.secret=?
-fitbit.max.users.per.poll=10
 fitbit.user.repository.class=org.radarbase.connect.rest.fitbit.user.ServiceUserRepository
-fitbit.user.repository.url=http://radar-device-auth-backend:8080/
+fitbit.user.repository.url=http://localhost:8080/
diff --git a/gradle/wrapper/gradle-wrapper.jar b/gradle/wrapper/gradle-wrapper.jar
diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties
@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-4.10.2-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-5.4.1-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
diff --git a/gradlew b/gradlew
@@ -1,5 +1,21 @@
 #!/usr/bin/env sh
 
+#
+# Copyright 2015 the original author or authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
 ##############################################################################
 ##
 ##  Gradle start up script for UN*X
@@ -28,7 +44,7 @@ APP_NAME="Gradle"
 APP_BASE_NAME=`basename "$0"`
 
 # Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
-DEFAULT_JVM_OPTS=""
+DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"'
 
 # Use the maximum available, or set MAX_FD != -1 to use that value.
 MAX_FD="maximum"
diff --git a/gradlew.bat b/gradlew.bat
@@ -1,3 +1,19 @@
+@rem
+@rem Copyright 2015 the original author or authors.
+@rem
+@rem Licensed under the Apache License, Version 2.0 (the "License");
+@rem you may not use this file except in compliance with the License.
+@rem You may obtain a copy of the License at
+@rem
+@rem      http://www.apache.org/licenses/LICENSE-2.0
+@rem
+@rem Unless required by applicable law or agreed to in writing, software
+@rem distributed under the License is distributed on an "AS IS" BASIS,
+@rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+@rem See the License for the specific language governing permissions and
+@rem limitations under the License.
+@rem
+
 @if "%DEBUG%" == "" @echo off
 @rem ##########################################################################
 @rem
@@ -14,7 +30,7 @@ set APP_BASE_NAME=%~n0
 set APP_HOME=%DIRNAME%
 
 @rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
-set DEFAULT_JVM_OPTS=
+set DEFAULT_JVM_OPTS="-Xmx64m" "-Xms64m"
 
 @rem Find java.exe
 if defined JAVA_HOME goto findJavaFromJavaHome
diff --git a/kafka-connect-fitbit-source/build.gradle b/kafka-connect-fitbit-source/build.gradle
@@ -1,11 +1,7 @@
-repositories {
-    maven { url 'https://oss.jfrog.org/artifactory/oss-snapshot-local' }
-}
-
 dependencies {
     api project(':kafka-connect-rest-source')
     api group: 'io.confluent', name: 'kafka-connect-avro-converter', version: confluentVersion
-    api group: 'org.radarcns', name: 'radar-schemas-commons', version: '0.4.4-SNAPSHOT'
+    api group: 'org.radarcns', name: 'radar-schemas-commons', version: '0.5.0'
 
     implementation group: 'com.fasterxml.jackson.dataformat', name: 'jackson-dataformat-yaml', version: jacksonVersion
     implementation group: 'com.fasterxml.jackson.datatype', name: 'jackson-datatype-jsr310', version: jacksonVersion
@@ -14,9 +10,9 @@ dependencies {
     compileOnly group: 'org.apache.kafka', name: 'connect-api', version: kafkaVersion
     compileOnly group: 'com.fasterxml.jackson.core', name: 'jackson-databind', version: jacksonVersion
 
-    testImplementation group: 'org.junit.jupiter', name: 'junit-jupiter-api', version: '5.3.2'
-    testRuntimeOnly group: 'org.junit.jupiter', name: 'junit-jupiter-engine', version: '5.3.2'
-    testRuntimeOnly group: 'org.slf4j', name: 'slf4j-simple', version: '1.7.25'
+    testImplementation group: 'org.junit.jupiter', name: 'junit-jupiter-api', version: '5.4.2'
+    testRuntimeOnly group: 'org.junit.jupiter', name: 'junit-jupiter-engine', version: '5.4.2'
+    testRuntimeOnly group: 'org.slf4j', name: 'slf4j-simple', version: '1.7.26'
     testImplementation group: 'org.apache.kafka', name: 'connect-api', version: kafkaVersion
 }
 
diff --git a/kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/FitbitRestSourceConnectorConfig.java b/kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/FitbitRestSourceConnectorConfig.java
@@ -105,11 +105,6 @@ public class FitbitRestSourceConnectorConfig extends RestSourceConnectorConfig {
   private static final String FITBIT_TIME_ZONE_TOPIC_DEFAULT = "connect_fitbit_time_zone";
   private static final String FITBIT_TIME_ZONE_TOPIC_DISPLAY = "Time zone topic";
 
-  private static final String FITBIT_MAX_USERS_PER_POLL_CONFIG = "fitbit.max.users.per.poll";
-  private static final String FITBIT_MAX_USERS_PER_POLL_DOC = "Maximum number of users to query in a single poll operation. Decrease this if memory constrains are pressing.";
-  private static final int FITBIT_MAX_USERS_PER_POLL_DEFAULT = 100;
-  private static final String FITBIT_MAX_USERS_PER_POLL_DISPLAY = "Maximum users per poll";
-
   private static final String FITBIT_ACTIVITY_LOG_TOPIC_CONFIG = "fitbit.activity.log.topic";
   private static final String FITBIT_ACTIVITY_LOG_TOPIC_DOC = "Topic for Fitbit activity log.";
   private static final String FITBIT_ACTIVITY_LOG_TOPIC_DEFAULT = "connect_fitbit_activity_log";
@@ -152,7 +147,6 @@ public String toString() {
     };
 
     return RestSourceConnectorConfig.conf()
-
         .define(FITBIT_USERS_CONFIG,
             Type.LIST,
             Collections.emptyList(),
@@ -225,17 +219,6 @@ public String toString() {
             Width.SHORT,
             FITBIT_USER_REPOSITORY_URL_DISPLAY)
 
-        .define(FITBIT_MAX_USERS_PER_POLL_CONFIG,
-            Type.INT,
-            FITBIT_MAX_USERS_PER_POLL_DEFAULT,
-            Range.atLeast(1),
-            Importance.LOW,
-            FITBIT_MAX_USERS_PER_POLL_DOC,
-            group,
-            ++orderInGroup,
-            Width.SHORT,
-            FITBIT_MAX_USERS_PER_POLL_DISPLAY)
-
         .define(FITBIT_INTRADAY_STEPS_TOPIC_CONFIG,
             Type.STRING,
             FITBIT_INTRADAY_STEPS_TOPIC_DEFAULT,
@@ -362,10 +345,6 @@ public Headers getClientCredentials() {
     return clientCredentials;
   }
 
-  public long getMaxUsersPerPoll() {
-    return getInt(FITBIT_MAX_USERS_PER_POLL_CONFIG);
-  }
-
   public String getActivityLogTopic() {
     return getString(FITBIT_ACTIVITY_LOG_TOPIC_CONFIG);
   }
diff --git a/kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/route/FitbitPollingRoute.java b/kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/route/FitbitPollingRoute.java
@@ -109,7 +109,6 @@ public abstract class FitbitPollingRoute implements PollingRequestRoute {
   private Duration pollInterval;
   private Instant lastPoll;
   private String baseUrl;
-  private long maxUsersPerPoll;
   private Duration pollIntervalPerUser;
   private final Set<User> tooManyRequestsForUser;
   private Duration tooManyRequestsCooldown;
@@ -133,7 +132,6 @@ public void initialize(RestSourceConnectorConfig config) {
     FitbitRestSourceConnectorConfig fitbitConfig = (FitbitRestSourceConnectorConfig) config;
     this.pollInterval = fitbitConfig.getPollInterval();
     this.baseUrl = fitbitConfig.getUrl();
-    this.maxUsersPerPoll = fitbitConfig.getMaxUsersPerPoll();
     this.pollIntervalPerUser = fitbitConfig.getPollIntervalPerUser();
     this.tooManyRequestsCooldown = fitbitConfig.getTooManyRequestsCooldownInterval()
         .minus(getPollIntervalPerUser());
@@ -187,7 +185,6 @@ public Stream<FitbitRestRequest> requests() {
           .map(u -> new AbstractMap.SimpleImmutableEntry<>(u, nextPoll(u)))
           .filter(u -> lastPoll.isAfter(u.getValue()))
           .sorted(Comparator.comparing(Map.Entry::getValue))
-          .limit(maxUsersPerPoll)
           .flatMap(u -> {
             lastPollPerUser.put(u.getKey().getId(), lastPoll);
             return this.createRequests(u.getKey());
@@ -199,6 +196,15 @@ public Stream<FitbitRestRequest> requests() {
     }
   }
 
+
+  /** Get the time that this route should be polled again. */
+  @Override
+  public Instant getTimeOfNextRequest() {
+    return nextPolls()
+            .min(Comparator.naturalOrder())
+            .orElse(nearFuture());
+  }
+
   private Map<String, Object> getPartition(User user) {
     return partitions.computeIfAbsent(user.getId(),
         k -> generator.getPartition(routeName, user));
diff --git a/kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/user/ServiceUserRepository.java b/kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/user/ServiceUserRepository.java
@@ -30,13 +30,10 @@
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import javax.ws.rs.NotAuthorizedException;
-
 import okhttp3.HttpUrl;
 import okhttp3.MediaType;
 import okhttp3.OkHttpClient;
@@ -56,15 +53,17 @@ public class ServiceUserRepository implements UserRepository {
   private static final ObjectReader USER_LIST_READER = JSON_READER.forType(Users.class);
   private static final ObjectReader USER_READER = JSON_READER.forType(User.class);
   private static final ObjectReader OAUTH_READER = JSON_READER.forType(OAuth2UserCredentials.class);
-  private static final RequestBody EMPTY_BODY = RequestBody.create(
-      MediaType.parse("application/json; charset=utf-8"), "");
+  private static final RequestBody EMPTY_BODY =
+      RequestBody.create(MediaType.parse("application/json; charset=utf-8"), "");
+  private static final Duration FETCH_THRESHOLD = Duration.ofMinutes(1L);
+
   private final OkHttpClient client;
-  private HttpUrl baseUrl;
   private final Map<String, OAuth2UserCredentials> cachedCredentials;
-  private HashSet<String> containedUsers;
-  private Set< ? extends User> timedCachedUsers = new HashSet<>();
   private final AtomicReference<Instant> nextFetch = new AtomicReference<>(MIN_INSTANT);
-  private static final Duration FETCH_THRESHOLD = Duration.ofMinutes(1L);
+
+  private HttpUrl baseUrl;
+  private HashSet<String> containedUsers;
+  private Set<? extends User> timedCachedUsers = new HashSet<>();
 
   public ServiceUserRepository() {
     this.client = new OkHttpClient();
@@ -97,10 +96,11 @@ public Stream<? extends User> stream() throws IOException {
 
     logger.info("Requesting user information from webservice");
     Request request = requestFor("users" + "?source-type=FitBit").build();
-    this.timedCachedUsers = this.<Users>makeRequest(request, USER_LIST_READER)
-        .getUsers().stream()
-        .filter(u -> containedUsers.isEmpty() || containedUsers.contains(u.getId()))
-        .collect(Collectors.toSet());
+    this.timedCachedUsers =
+        this.<Users>makeRequest(request, USER_LIST_READER).getUsers().stream()
+            .filter(u -> u.isComplete()
+                && (containedUsers.isEmpty() || containedUsers.contains(u.getId())))
+            .collect(Collectors.toSet());
 
     return this.timedCachedUsers.stream();
   }
@@ -118,9 +118,7 @@ public String getAccessToken(User user) throws IOException, NotAuthorizedExcepti
 
   @Override
   public String refreshAccessToken(User user) throws IOException, NotAuthorizedException {
-    Request request = requestFor("users/" + user.getId() + "/token")
-      .post(EMPTY_BODY)
-      .build();
+    Request request = requestFor("users/" + user.getId() + "/token").post(EMPTY_BODY).build();
     OAuth2UserCredentials credentials = makeRequest(request, OAUTH_READER);
     cachedCredentials.put(user.getId(), credentials);
     return credentials.getAccessToken();
diff --git a/kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/user/User.java b/kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/user/User.java
@@ -23,18 +23,32 @@
 import org.radarcns.kafka.ObservationKey;
 
 public interface User {
+  static SchemaAndValue computeObservationKey(AvroData avroData, User user) {
+    return avroData.toConnectData(
+        ObservationKey.getClassSchema(),
+        new ObservationKey(user.getProjectId(), user.getUserId(), user.getSourceId()));
+  }
+
   String getId();
+
   String getExternalUserId();
+
   String getProjectId();
+
   String getUserId();
+
   Instant getStartDate();
+
   Instant getEndDate();
+
   String getSourceId();
+
   SchemaAndValue getObservationKey(AvroData avroData);
 
-  static SchemaAndValue computeObservationKey(AvroData avroData, User user) {
-    return avroData.toConnectData(
-          ObservationKey.getClassSchema(),
-          new ObservationKey(user.getProjectId(), user.getUserId(), user.getSourceId()));
+  default Boolean isComplete() {
+    return getEndDate() != null
+        && getStartDate() != null
+        && getProjectId() != null
+        && getUserId() != null;
   }
 }
diff --git a/kafka-connect-rest-source/build.gradle b/kafka-connect-rest-source/build.gradle
@@ -1,14 +1,14 @@
 dependencies {
-    api group: 'com.squareup.okhttp3', name: 'okhttp', version: '3.12.1'
+    api group: 'com.squareup.okhttp3', name: 'okhttp', version: '3.14.2'
 
     // Included in connector runtime
     compileOnly group: 'org.apache.kafka', name: 'connect-api', version: kafkaVersion
 
     testImplementation group: 'ch.qos.logback', name: 'logback-classic', version: '1.2.3'
-    testImplementation group: 'org.junit.jupiter', name: 'junit-jupiter-api', version: '5.3.2'
-    testRuntimeOnly group: 'org.junit.jupiter', name: 'junit-jupiter-engine', version: '5.3.2'
-    testImplementation group: 'org.mockito', name: 'mockito-core', version: '2.23.4'
-    testImplementation group: 'com.github.tomakehurst', name: 'wiremock', version: '2.20.0'
+    testImplementation group: 'org.junit.jupiter', name: 'junit-jupiter-api', version: '5.4.2'
+    testRuntimeOnly group: 'org.junit.jupiter', name: 'junit-jupiter-engine', version: '5.4.2'
+    testImplementation group: 'org.mockito', name: 'mockito-core', version: '2.27.0'
+    testImplementation group: 'com.github.tomakehurst', name: 'wiremock', version: '2.23.2'
 
     testImplementation group: 'org.apache.kafka', name: 'connect-api', version: kafkaVersion
 }
diff --git a/kafka-connect-rest-source/src/main/java/org/radarbase/connect/rest/RestSourceTask.java b/kafka-connect-rest-source/src/main/java/org/radarbase/connect/rest/RestSourceTask.java
