Merge pull request #71 from RADAR-base/release-0.3.2

Release 0.3.2

Author: yatharthranjan

build.gradle

@@ -11,7 +11,7 @@ subprojects {
     apply plugin: 'java-library'
 
     group = 'org.radarbase'
-    version = '0.3.2-SNAPSHOT'
+    version = '0.3.2'
 
     sourceCompatibility = 1.8
     targetCompatibility = 1.8

docker-compose.yml

@@ -9,7 +9,7 @@ services:
   # Zookeeper Cluster                                                         #
   #---------------------------------------------------------------------------#
   zookeeper-1:
-    image: confluentinc/cp-zookeeper:5.1.0
+    image: confluentinc/cp-zookeeper:5.5.1
     environment:
       ZOOKEEPER_SERVER_ID: 1
       ZOOKEEPER_CLIENT_PORT: 2181
@@ -19,7 +19,7 @@ services:
       ZOOKEEPER_SERVERS: zookeeper-1:2888:3888;zookeeper-2:2888:3888;zookeeper-3:2888:3888
 
   zookeeper-2:
-    image: confluentinc/cp-zookeeper:5.1.0
+    image: confluentinc/cp-zookeeper:5.5.1
     environment:
       ZOOKEEPER_SERVER_ID: 2
       ZOOKEEPER_CLIENT_PORT: 2181
@@ -29,7 +29,7 @@ services:
       ZOOKEEPER_SERVERS: zookeeper-1:2888:3888;zookeeper-2:2888:3888;zookeeper-3:2888:3888
 
   zookeeper-3:
-    image: confluentinc/cp-zookeeper:5.1.0
+    image: confluentinc/cp-zookeeper:5.5.1
     environment:
       ZOOKEEPER_SERVER_ID: 3
       ZOOKEEPER_CLIENT_PORT: 2181
@@ -42,7 +42,7 @@ services:
   # Kafka Cluster                                                             #
   #---------------------------------------------------------------------------#
   kafka-1:
-    image: confluentinc/cp-kafka:5.1.0
+    image: confluentinc/cp-kafka:5.5.1
     depends_on:
       - zookeeper-1
       - zookeeper-2
@@ -61,7 +61,7 @@ services:
       KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE: "false"
 
   kafka-2:
-    image: confluentinc/cp-kafka:5.1.0
+    image: confluentinc/cp-kafka:5.5.1
     depends_on:
       - zookeeper-1
       - zookeeper-2
@@ -80,7 +80,7 @@ services:
       KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE: "false"
 
   kafka-3:
-    image: confluentinc/cp-kafka:5.1.0
+    image: confluentinc/cp-kafka:5.5.1
     depends_on:
       - zookeeper-1
       - zookeeper-2
@@ -102,7 +102,7 @@ services:
   # Schema Registry                                                           #
   #---------------------------------------------------------------------------#
   schema-registry-1:
-    image: confluentinc/cp-schema-registry:5.1.0
+    image: confluentinc/cp-schema-registry:5.5.1
     depends_on:
       - zookeeper-1
       - zookeeper-2
@@ -124,7 +124,7 @@ services:
   # REST proxy                                                                #
   #---------------------------------------------------------------------------#
   rest-proxy-1:
-    image: confluentinc/cp-kafka-rest:5.1.0
+    image: confluentinc/cp-kafka-rest:5.5.1
     depends_on:
       - zookeeper-1
       - zookeeper-2

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

@@ -27,18 +27,14 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.Iterator;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Optional;
-import java.util.OptionalDouble;
-import java.util.OptionalLong;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
-import okhttp3.Response;
-import okhttp3.ResponseBody;
+import okhttp3.Headers;
+import org.apache.avro.Schema.Field;
 import org.apache.avro.generic.IndexedRecord;
 import org.apache.kafka.connect.data.SchemaAndValue;
 import org.apache.kafka.connect.source.SourceRecord;
@@ -74,19 +70,18 @@ public FitbitAvroConverter(AvroData avroData) {
 
   @Override
   public Collection<SourceRecord> convert(
-      RestRequest restRequest, Response response) throws IOException {
-    ResponseBody body = response.body();
-    if (body == null) {
+      RestRequest restRequest, Headers headers, byte[] data) throws IOException {
+    if (data == null) {
       throw new IOException("Failed to read body");
     }
-    JsonNode activities = JSON_READER.readTree(body.charStream());
+    JsonNode activities = JSON_READER.readTree(data);
 
     User user = ((FitbitRestRequest) restRequest).getUser();
     final SchemaAndValue key = user.getObservationKey(avroData);
     double timeReceived = System.currentTimeMillis() / 1000d;
 
     return processRecords((FitbitRestRequest)restRequest, activities, timeReceived)
-        .filter(Objects::nonNull)
+        .filter(t -> validateRecord((FitbitRestRequest)restRequest, t))
         .map(t -> {
           SchemaAndValue avro = avroData.toConnectData(t.value.getSchema(), t.value);
           Map<String, ?> offset = Collections.singletonMap(
@@ -98,6 +93,22 @@ public Collection<SourceRecord> convert(
         .collect(Collectors.toList());
   }
 
+  private boolean validateRecord(FitbitRestRequest request, TopicData record) {
+    if (record == null) {
+      return false;
+    }
+    Instant endDate = request.getUser().getEndDate();
+    if (endDate == null) {
+      return true;
+    }
+    Field timeField = record.value.getSchema().getField("time");
+    if (timeField != null) {
+      long time = (long) (((Double)record.value.get(timeField.pos()) * 1000.0));
+      return Instant.ofEpochMilli(time).isBefore(endDate);
+    }
+    return true;
+  }
+
   /** Process the JSON records generated by given request. */
   protected abstract Stream<TopicData> processRecords(
       FitbitRestRequest request,

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

@@ -312,7 +312,7 @@ protected Duration getLookbackTime() {
    */
   protected Instant nextPoll(User user) {
     Instant offset = getOffset(user);
-    if (offset.isAfter(user.getEndDate())) {
+    if (offset.isAfter(user.getEndDate().minus(getEndDateThreshold()))) {
       return nearFuture();
     } else {
       Instant nextPoll = lastPollPerUser.getOrDefault(user.getId(), MIN_INSTANT)
@@ -321,6 +321,10 @@ protected Instant nextPoll(User user) {
     }
   }
 
+  private TemporalAmount getEndDateThreshold() {
+    return Duration.ofHours(1);
+  }
+
   /**
    * Generate one date per day, using UTC time zone. The first date will have the time from the
    * given startDate. Following time stamps will start at 00:00. This will not up to the date of

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

@@ -45,6 +45,9 @@ public class LocalUser implements User {
   @JsonProperty("oauth2")
   private OAuth2UserCredentials oauth2Credentials = new OAuth2UserCredentials();
 
+  @JsonProperty("authorized")
+  private Boolean isAuthorized;
+
   @JsonIgnore
   private SchemaAndValue observationKey;
 
@@ -95,6 +98,15 @@ public void setFitbitUserId(String id) {
     this.externalUserId = id;
   }
 
+  @Override
+  public boolean isAuthorized() {
+    if(isAuthorized == null) {
+      return !oauth2Credentials.isAccessTokenExpired()
+          || oauth2Credentials.hasRefreshToken();
+    }
+    return isAuthorized;
+  }
+
   @Override
   public String getVersion() {
     return version;
@@ -115,6 +127,7 @@ public LocalUser copy() {
     copy.endDate = endDate;
     copy.sourceId = sourceId;
     copy.oauth2Credentials = oauth2Credentials;
+    copy.isAuthorized = isAuthorized;
     return copy;
   }
 
@@ -133,6 +146,7 @@ public String toString() {
         + ", projectId='" + projectId + '\''
         + ", userId='" + userId + '\''
         + ", sourceId='" + sourceId + '\''
+        + ", isAuthorized='" + isAuthorized() + '\''
         + '}';
   }
 
@@ -152,7 +166,8 @@ public boolean equals(Object o) {
         && Objects.equals(userId, localUser.userId)
         && Objects.equals(sourceId, localUser.sourceId)
         && Objects.equals(startDate, localUser.startDate)
-        && Objects.equals(endDate, localUser.endDate);
+        && Objects.equals(endDate, localUser.endDate)
+        && Objects.equals(isAuthorized(), localUser.isAuthorized());
   }
 
   @Override

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

@@ -45,6 +45,8 @@ static SchemaAndValue computeObservationKey(AvroData avroData, User user) {
 
   String getSourceId();
 
+  boolean isAuthorized();
+
   default String getVersionedId() {
     String version = getVersion();
     if (version == null) {
@@ -60,6 +62,7 @@ default Boolean isComplete() {
     return getEndDate() != null
         && getStartDate() != null
         && getProjectId() != null
-        && getUserId() != null;
+        && getUserId() != null
+        && isAuthorized();
   }
 }

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

@@ -132,6 +132,9 @@ private void forceUpdateUsers() {
 
   @Override
   public Stream<LocalUser> stream() {
+     if (nextFetch.get().equals(MIN_INSTANT)) {
+      applyPendingUpdates();
+    }
     Stream<LockedUser> users = this.users.values().stream()
         .filter(lockedTest(u -> u.getOAuth2Credentials().hasRefreshToken()));
     if (!configuredUsers.isEmpty()) {

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

@@ -70,6 +70,12 @@ public String getSourceId() {
     return fitbitAuthDetails.getSourceId();
   }
 
+  @Override
+  public boolean isAuthorized() {
+    return !fitbitAuthDetails.getOauth2Credentials().isAccessTokenExpired()
+        || fitbitAuthDetails.getOauth2Credentials().hasRefreshToken();
+  }
+
   public FirebaseUserDetails getFirebaseUserDetails() {
     return firebaseUserDetails;
   }

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

@@ -23,6 +23,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Map;
+import okhttp3.Headers;
 import okhttp3.Response;
 import okhttp3.ResponseBody;
 import org.apache.kafka.connect.data.Schema;
@@ -36,15 +37,13 @@ public class BytesPayloadConverter implements PayloadToSourceRecordConverter {
 
   // Just bytes for incoming messages
   @Override
-  public Collection<SourceRecord> convert(RestRequest request, Response response) throws IOException {
+  public Collection<SourceRecord> convert(RestRequest request, Headers headers, byte[] data) {
     Map<String, Long> sourceOffset = Collections.singletonMap(
         TIMESTAMP_OFFSET_KEY, currentTimeMillis());
-    ResponseBody body = response.body();
-    byte[] result = body != null ? body.bytes() : null;
-    String topic = topicSelector.getTopic(request, result);
+    String topic = topicSelector.getTopic(request, data);
     return Collections.singleton(
         new SourceRecord(request.getPartition(), sourceOffset,
-            topic, Schema.BYTES_SCHEMA, result));
+            topic, Schema.BYTES_SCHEMA, data));
   }
 
   @Override

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

@@ -22,7 +22,7 @@
 import java.time.Instant;
 import java.time.temporal.TemporalAmount;
 import java.util.Collection;
-import okhttp3.Response;
+import okhttp3.Headers;
 import org.apache.kafka.connect.source.SourceRecord;
 import org.radarbase.connect.rest.config.RestSourceTool;
 import org.radarbase.connect.rest.request.RestRequest;
@@ -33,7 +33,7 @@ public interface PayloadToSourceRecordConverter extends RestSourceTool {
   TemporalAmount NEAR_FUTURE = Duration.ofDays(31L);
 
   Collection<SourceRecord> convert(
-      RestRequest request, Response response) throws IOException;
+      RestRequest request, Headers headers, byte[] data) throws IOException;
 
   static Instant nearFuture() {
     return Instant.now().plus(NEAR_FUTURE);