Merge pull request #50 from RADAR-base/reset-user-id

Add version property to user so it can be reset

Author: yatharthranjan

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

@@ -98,7 +98,7 @@ private List<Map<String, String>> configureTasks(int maxTasks) {
     try {
 
       List<Map<String, String>> userTasks = fitbitConfig.getUserRepository().stream()
-          .map(User::getId)
+          .map(User::getVersionedId)
           // group users based on their hashCode
           // in principle this allows for more efficient reconfigurations for a fixed number of tasks,
           // since that allows existing tasks to only handle small modifications users to handle.

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

@@ -105,7 +105,7 @@ public OkHttpClient getClient(User user) {
   public Map<String, Map<String, Object>> getPartitions(String route) {
     try {
       return userRepository.stream()
-          .collect(Collectors.toMap(User::getId, u -> getPartition(route, u)));
+          .collect(Collectors.toMap(User::getVersionedId, u -> getPartition(route, u)));
     } catch (IOException e) {
       logger.warn("Failed to initialize user partitions for route {}: {}", route, e.toString());
       return Collections.emptyMap();
@@ -114,7 +114,7 @@ public Map<String, Map<String, Object>> getPartitions(String route) {
 
   public Map<String, Object> getPartition(String route, User user) {
     Map<String, Object> partition = new HashMap<>(4);
-    partition.put("user", user.getId());
+    partition.put("user", user.getVersionedId());
     partition.put("route", route);
     return partition;
   }

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

@@ -143,7 +143,7 @@ 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();
+    String userKey = ((FitbitRestRequest) request).getUser().getVersionedId();
     Instant offset = Instant.ofEpochMilli((Long) record.sourceOffset().get(TIMESTAMP_OFFSET_KEY));
     offsets.put(userKey, offset);
   }
@@ -154,7 +154,7 @@ public void requestEmpty(RestRequest request) {
     FitbitRestRequest fitbitRequest = (FitbitRestRequest) request;
     Instant endOffset = fitbitRequest.getDateRange().end().toInstant();
     if (DAYS.between(endOffset, lastPoll) >= HISTORICAL_TIME_DAYS) {
-      String key = fitbitRequest.getUser().getId();
+      String key = fitbitRequest.getUser().getVersionedId();
       offsets.put(key, endOffset);
     }
   }
@@ -197,7 +197,7 @@ public Stream<FitbitRestRequest> requests() {
       return userRepository.stream()
           .map(u -> new AbstractMap.SimpleImmutableEntry<>(u, nextPoll(u)))
           .filter(u -> lastPoll.isAfter(u.getValue()))
-          .sorted(Comparator.comparing(Map.Entry::getValue))
+          .sorted(Map.Entry.comparingByValue())
           .flatMap(u -> this.createRequests(u.getKey()))
           .filter(Objects::nonNull);
     } catch (IOException e) {
@@ -216,7 +216,7 @@ public Instant getTimeOfNextRequest() {
   }
 
   private Map<String, Object> getPartition(User user) {
-    return partitions.computeIfAbsent(user.getId(),
+    return partitions.computeIfAbsent(user.getVersionedId(),
         k -> generator.getPartition(routeName, user));
   }
 
@@ -279,7 +279,7 @@ public Instant getLastPoll() {
   }
 
   protected Instant getOffset(User user) {
-    return offsets.getOrDefault(user.getId(), user.getStartDate().minus(ONE_NANO));
+    return offsets.getOrDefault(user.getVersionedId(), user.getStartDate().minus(ONE_NANO));
   }
 
   /**

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

@@ -34,6 +34,7 @@
 public class LocalUser implements User {
   private static final Pattern ILLEGAL_CHARACTERS_PATTERN = Pattern.compile("[^a-zA-Z0-9_-]");
   private String id;
+  private String version;
   private String externalUserId;
   private String projectId;
   private String userId;
@@ -94,9 +95,19 @@ public void setFitbitUserId(String id) {
     this.externalUserId = id;
   }
 
+  @Override
+  public String getVersion() {
+    return version;
+  }
+
+  public void setVersion(String version) {
+    this.version = version;
+  }
+
   public LocalUser copy() {
     LocalUser copy = new LocalUser();
     copy.id = id;
+    copy.version = version;
     copy.externalUserId = externalUserId;
     copy.projectId = projectId;
     copy.userId = userId;
@@ -116,7 +127,8 @@ public synchronized SchemaAndValue getObservationKey(AvroData avroData) {
 
   @Override
   public String toString() {
-    return "LocalUser{" + "id='" + id + '\''
+    return "LocalUser{id='" + id + '\''
+        + ", version='" + version + '\''
         + ", externalUserId='" + externalUserId + '\''
         + ", projectId='" + projectId + '\''
         + ", userId='" + userId + '\''
@@ -134,19 +146,17 @@ public boolean equals(Object o) {
     }
     LocalUser localUser = (LocalUser) o;
     return Objects.equals(id, localUser.id)
+        && Objects.equals(version, localUser.version)
         && Objects.equals(externalUserId, localUser.externalUserId)
         && Objects.equals(projectId, localUser.projectId)
         && Objects.equals(userId, localUser.userId)
         && Objects.equals(sourceId, localUser.sourceId)
         && Objects.equals(startDate, localUser.startDate)
-        && Objects.equals(endDate, localUser.endDate)
-        && Objects.equals(observationKey, localUser.observationKey);
+        && Objects.equals(endDate, localUser.endDate);
   }
 
   @Override
   public int hashCode() {
-
-    return Objects
-        .hash(id, externalUserId, projectId, userId, sourceId, startDate, endDate, observationKey);
+    return Objects.hash(id, version);
   }
 }

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

@@ -99,7 +99,7 @@ public Stream<? extends User> stream() throws IOException {
     this.timedCachedUsers =
         this.<Users>makeRequest(request, USER_LIST_READER).getUsers().stream()
             .filter(u -> u.isComplete()
-                && (containedUsers.isEmpty() || containedUsers.contains(u.getId())))
+                && (containedUsers.isEmpty() || containedUsers.contains(u.getVersionedId())))
             .collect(Collectors.toSet());
 
     return this.timedCachedUsers.stream();

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

@@ -31,6 +31,8 @@ static SchemaAndValue computeObservationKey(AvroData avroData, User user) {
 
   String getId();
 
+  String getVersion();
+
   String getExternalUserId();
 
   String getProjectId();
@@ -43,6 +45,15 @@ static SchemaAndValue computeObservationKey(AvroData avroData, User user) {
 
   String getSourceId();
 
+  default String getVersionedId() {
+    String version = getVersion();
+    if (version == null) {
+      return getId();
+    } else {
+      return getId() + "#" + version;
+    }
+  }
+
   SchemaAndValue getObservationKey(AvroData avroData);
 
   default Boolean isComplete() {

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

@@ -134,7 +134,7 @@ public Stream<LocalUser> stream() {
     Stream<LockedUser> users = this.users.values().stream()
         .filter(lockedTest(u -> u.getOAuth2Credentials().hasRefreshToken()));
     if (!configuredUsers.isEmpty()) {
-      users = users.filter(lockedTest(u -> configuredUsers.contains(u.getId())));
+      users = users.filter(lockedTest(u -> configuredUsers.contains(u.getVersionedId())));
     }
     return users.map(lockedApply(LocalUser::copy));
   }
@@ -297,7 +297,7 @@ private void store(Path path, LocalUser user) {
    * Local user that is protected by a multi-threading lock to avoid simultaneous IO
    * and modifications.
    */
-  private final class LockedUser {
+  private static final class LockedUser {
     final Lock lock = new ReentrantLock();
     final LocalUser user;
     final Path path;