Merge pull request #44 from RADAR-base/release-0.2.3

Release 0.2.3

Author: blootsvoets

README.md

@@ -53,7 +53,9 @@ your Fitbit App client ID and client secret. The following tables shows the poss
 <tr>
 <td>fitbit.api.secret</td></td><td>Secret for the Fitbit API client set in fitbit.api.client.</td></td><td>password</td></td><td></td></td><td></td></td><td>high</td></td></tr>
 <tr>
-<td>fitbit.api.intraday</td></td><td>Set to true if the client has permissions to Fitbit Intraday API, false otherwise.</td></td><td>boolean</td></td><td>true</td></td><td></td></td><td>medium</td></td></tr>
+<td>fitbit.user.poll.interval</td></td><td>Polling interval per Fitbit user per request route in seconds.</td></td><td>int</td></td><td>150</td></td><td></td></td><td>medium</td></td></tr>
+<tr>
+<td>fitbit.api.intraday</td></td><td>Set to true if the client has permissions to Fitbit Intraday API, false otherwise.</td></td><td>boolean</td></td><td>false</td></td><td></td></td><td>medium</td></td></tr>
 <tr>
 <td>fitbit.user.repository.class</td></td><td>Class for managing users and authentication.</td></td><td>class</td></td><td>org.radarbase.connect.rest.fitbit.user.YamlUserRepository</td></td><td>Class extending org.radarbase.connect.rest.fitbit.user.UserRepository</td></td><td>medium</td></td></tr>
 <tr>

build.gradle

@@ -10,8 +10,8 @@ subprojects {
     apply plugin: 'java'
     apply plugin: 'java-library'
 
-    group = 'org.radarcns'
-    version = '0.2.2'
+    group = 'org.radarbase'
+    version = '0.2.3'
 
     sourceCompatibility = 1.8
     targetCompatibility = 1.8

kafka-connect-fitbit-source/build.gradle

@@ -1,7 +1,7 @@
 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.5.0'
+    api group: 'org.radarcns', name: 'radar-schemas-commons', version: '0.5.1'
 
     implementation group: 'com.fasterxml.jackson.dataformat', name: 'jackson-dataformat-yaml', version: jacksonVersion
     implementation group: 'com.fasterxml.jackson.datatype', name: 'jackson-datatype-jsr310', version: jacksonVersion

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/FitbitRestSourceConnectorConfig.java

@@ -70,6 +70,14 @@ public class FitbitRestSourceConnectorConfig extends RestSourceConnectorConfig {
   private static final boolean FITBIT_API_INTRADAY_ACCESS_DEFAULT = false;
   private static final String FITBIT_API_INTRADAY_ACCESS_DISPLAY = "Is Fitbit Intraday API available?";
 
+  public static final String FITBIT_USER_POLL_INTERVAL = "fitbit.user.poll.interval";
+  private static final String FITBIT_USER_POLL_INTERVAL_DOC = "Polling interval per Fitbit user per request route in seconds.";
+  // 150 requests per hour -> 2.5 per minute. There are currently 5 paths, that limits us to 1
+  // call per route per 2 minutes.
+  private static final int FITBIT_USER_POLL_INTERVAL_DEFAULT = 150;
+  private static final String FITBIT_USER_POLL_INTERVAL_DISPLAY = "Per-user per-route polling interval.";
+
+
   public static final String FITBIT_USER_CREDENTIALS_DIR_CONFIG = "fitbit.user.dir";
   private static final String FITBIT_USER_CREDENTIALS_DIR_DOC = "Directory containing Fitbit user information and credentials. Only used if a file-based user repository is configured.";
   private static final String FITBIT_USER_CREDENTIALS_DIR_DISPLAY = "User directory";
@@ -178,9 +186,19 @@ public String toString() {
             Width.SHORT,
             FITBIT_API_SECRET_DISPLAY)
 
+        .define(FITBIT_USER_POLL_INTERVAL,
+            Type.INT,
+            FITBIT_USER_POLL_INTERVAL_DEFAULT,
+            Importance.MEDIUM,
+            FITBIT_USER_POLL_INTERVAL_DOC,
+            group,
+            ++orderInGroup,
+            Width.SHORT,
+            FITBIT_USER_POLL_INTERVAL_DISPLAY)
+
         .define(FITBIT_API_INTRADAY_ACCESS_CONFIG,
             Type.BOOLEAN,
-            true,
+            FITBIT_API_INTRADAY_ACCESS_DEFAULT,
             Importance.MEDIUM,
             FITBIT_API_INTRADAY_ACCESS_DOC,
             group,
@@ -352,4 +370,12 @@ public String getActivityLogTopic() {
   public boolean hasIntradayAccess() {
     return getBoolean(FITBIT_API_INTRADAY_ACCESS_CONFIG);
   }
+
+  public Duration getPollIntervalPerUser() {
+    return Duration.ofSeconds(getInt(FITBIT_USER_POLL_INTERVAL));
+  }
+
+  public Duration getTooManyRequestsCooldownInterval() {
+    return Duration.ofHours(1);
+  }
 }

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/route/FitbitIntradayHeartRateRoute.java

@@ -29,12 +29,11 @@
 import org.radarbase.connect.rest.fitbit.user.UserRepository;
 
 public class FitbitIntradayHeartRateRoute extends FitbitPollingRoute {
-  private static final String ROUTE_NAME = "heart_rate";
   private final FitbitIntradayHeartRateAvroConverter converter;
 
   public FitbitIntradayHeartRateRoute(FitbitRequestGenerator generator,
       UserRepository userRepository, AvroData avroData) {
-    super(generator, userRepository, ROUTE_NAME);
+    super(generator, userRepository, "heart_rate");
     this.converter = new FitbitIntradayHeartRateAvroConverter(avroData);
   }
 

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/route/FitbitPollingRoute.java

@@ -140,13 +140,15 @@ public void initialize(RestSourceConnectorConfig config) {
 
   @Override
   public void requestSucceeded(RestRequest request, SourceRecord record) {
+    lastPollPerUser.put(((FitbitRestRequest) request).getUser().getId(), lastPoll);
     String userKey = ((FitbitRestRequest) request).getUser().getId();
     Instant offset = Instant.ofEpochMilli((Long) record.sourceOffset().get(TIMESTAMP_OFFSET_KEY));
     offsets.put(userKey, offset);
   }
 
   @Override
   public void requestEmpty(RestRequest request) {
+    lastPollPerUser.put(((FitbitRestRequest) request).getUser().getId(), lastPoll);
     FitbitRestRequest fitbitRequest = (FitbitRestRequest) request;
     Instant endOffset = fitbitRequest.getDateRange().end().toInstant();
     if (DAYS.between(endOffset, lastPoll) >= HISTORICAL_TIME_DAYS) {
@@ -160,7 +162,16 @@ public void requestFailed(RestRequest request, Response response) {
     if (response != null && response.code() == 429) {
       User user = ((FitbitRestRequest)request).getUser();
       tooManyRequestsForUser.add(user);
-      Instant backOff = lastPoll.plus(getTooManyRequestsCooldown());
+      String cooldownString = response.header("Retry-After");
+      Duration cooldown = getTooManyRequestsCooldown();
+      if (cooldownString != null) {
+        try {
+          cooldown = Duration.ofSeconds(Long.parseLong(cooldownString));
+        } catch (NumberFormatException ex) {
+          cooldown = getTooManyRequestsCooldown();
+        }
+      }
+      Instant backOff = lastPoll.plus(cooldown);
       lastPollPerUser.put(user.getId(), backOff);
       logger.info("Too many requests for user {}. Backing off until {}",
           user, backOff.plus(getPollIntervalPerUser()));
@@ -185,10 +196,7 @@ public Stream<FitbitRestRequest> requests() {
           .map(u -> new AbstractMap.SimpleImmutableEntry<>(u, nextPoll(u)))
           .filter(u -> lastPoll.isAfter(u.getValue()))
           .sorted(Comparator.comparing(Map.Entry::getValue))
-          .flatMap(u -> {
-            lastPollPerUser.put(u.getKey().getId(), lastPoll);
-            return this.createRequests(u.getKey());
-          })
+          .flatMap(u -> this.createRequests(u.getKey()))
           .filter(Objects::nonNull);
     } catch (IOException e) {
       logger.warn("Cannot read users");

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/user/ServiceUserRepository.java

@@ -90,12 +90,12 @@ public Stream<? extends User> stream() throws IOException {
     Instant now = Instant.now();
     if (!now.isAfter(nextFetchTime)
         || !nextFetch.compareAndSet(nextFetchTime, now.plus(FETCH_THRESHOLD))) {
-      logger.info("Providing cached user information...");
+      logger.debug("Providing cached user information...");
       return timedCachedUsers.stream();
     }
 
     logger.info("Requesting user information from webservice");
-    Request request = requestFor("users" + "?source-type=FitBit").build();
+    Request request = requestFor("users?source-type=FitBit").build();
     this.timedCachedUsers =
         this.<Users>makeRequest(request, USER_LIST_READER).getUsers().stream()
             .filter(u -> u.isComplete()

kafka-connect-rest-source/src/main/java/org/radarbase/connect/rest/RestSourceConnectorConfig.java

@@ -214,12 +214,4 @@ public RequestGenerator getRequestGenerator() {
     requestGenerator.initialize(this);
     return requestGenerator;
   }
-
-  public Duration getPollIntervalPerUser() {
-    return Duration.ofMinutes(30);
-  }
-
-  public Duration getTooManyRequestsCooldownInterval() {
-    return Duration.ofHours(1);
-  }
 }

kafka-connect-rest-source/src/main/java/org/radarbase/connect/rest/RestSourceTask.java

@@ -20,10 +20,12 @@
 import static java.time.temporal.ChronoUnit.MILLIS;
 import static org.radarbase.connect.rest.util.ThrowingFunction.tryOrNull;
 
+import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -63,8 +65,8 @@ public void start(Map<String, String> map) {
 
   @Override
   public List<SourceRecord> poll() throws InterruptedException {
-    LongAdder requestsGenerated = new LongAdder();
-    List<SourceRecord> requests;
+    long requestsGenerated = 0;
+    List<SourceRecord> requests = Collections.emptyList();
 
     do {
       long timeout = MILLIS.between(Instant.now(), requestGenerator.getTimeOfNextRequest());
@@ -73,23 +75,30 @@ public List<SourceRecord> poll() throws InterruptedException {
         Thread.sleep(timeout);
       }
 
-      requests = requestGenerator.requests()
-          .sequential()
-          .filter(RestRequest::isStillValid)
-          .peek(r -> {
-            logger.info("Requesting {}", r.getRequest().url());
-            requestsGenerated.increment();
-          })
-          .map(tryOrNull(RestRequest::handleRequest,
-              (r, ex) -> logger.warn("Failed to make request: {}", ex.toString())))
-          .filter(Objects::nonNull)
-          .map(s -> s.collect(Collectors.toList()))
-          .filter(l -> !l.isEmpty())
-          .findFirst()
-          .orElse(Collections.emptyList());
+      Iterator<? extends RestRequest> requestIterator = requestGenerator.requests()
+          .iterator();
+
+
+      while (requests.isEmpty() && requestIterator.hasNext()) {
+        RestRequest request = requestIterator.next();
+
+        if (!request.isStillValid()) {
+          continue;
+        }
+
+        logger.info("Requesting {}", request.getRequest().url());
+        requestsGenerated++;
+
+        try {
+          requests = request.handleRequest()
+              .collect(Collectors.toList());
+        } catch (IOException ex) {
+          logger.warn("Failed to make request: {}", ex.toString());
+        }
+      }
     } while (requests.isEmpty());
 
-    logger.info("Processed {} records from {} URLs", requests.size(), requestsGenerated.sum());
+    logger.info("Processed {} records from {} URLs", requests.size(), requestsGenerated);
 
     return requests;
   }

kafka-connect-rest-source/src/main/java/org/radarbase/connect/rest/request/RestRequest.java

@@ -72,7 +72,7 @@ public boolean isStillValid() {
 
   /**
    * Handle the request using the internal client, using the request route converter.
-   * @return stream of resulting source records, or {@code null} if the response was not successful.
+   * @return stream of resulting source records.
    * @throws IOException if making or parsing the request failed.
    */
   public Stream<SourceRecord> handleRequest() throws IOException {
@@ -83,7 +83,7 @@ public Stream<SourceRecord> handleRequest() throws IOException {
     try (Response response = client.newCall(request).execute()) {
       if (!response.isSuccessful()) {
         route.requestFailed(this, response);
-        return null;
+        return Stream.empty();
       }
 
       Collection<SourceRecord> records = route.converter().convert(this, response);