Make forbidden backoff time configurable

Author: yatharthranjan

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

@@ -17,8 +17,6 @@
 
 package org.radarbase.connect.rest.fitbit;
 
-import static org.apache.kafka.common.config.ConfigDef.NO_DEFAULT_VALUE;
-
 import java.lang.reflect.InvocationTargetException;
 import java.net.MalformedURLException;
 import java.net.URL;
@@ -31,11 +29,9 @@
 import java.util.List;
 import java.util.Map;
 
-import okhttp3.Headers;
-import okhttp3.HttpUrl;
-
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigDef.Importance;
+import static org.apache.kafka.common.config.ConfigDef.NO_DEFAULT_VALUE;
 import org.apache.kafka.common.config.ConfigDef.NonEmptyString;
 import org.apache.kafka.common.config.ConfigDef.Type;
 import org.apache.kafka.common.config.ConfigDef.Validator;
@@ -47,6 +43,9 @@
 import org.radarbase.connect.rest.fitbit.user.UserRepository;
 import org.radarbase.connect.rest.fitbit.user.YamlUserRepository;
 
+import okhttp3.Headers;
+import okhttp3.HttpUrl;
+
 public class FitbitRestSourceConnectorConfig extends RestSourceConnectorConfig {
   public static final String FITBIT_USERS_CONFIG = "fitbit.users";
   private static final String FITBIT_USERS_DOC =
@@ -250,6 +249,11 @@ public class FitbitRestSourceConnectorConfig extends RestSourceConnectorConfig {
   private static final String FITBIT_MAX_FORBIDDEN_DISPLAY = "Max forbidden responses";
   private static final int FITBIT_MAX_FORBIDDEN_DEFAULT = 3;
 
+  private static final String FITBIT_FORBIDDEN_BACKOFF_CONFIG = "fitbit.request.forbidden.backoff.s";
+  private static final String FITBIT_FORBIDDEN_BACKOFF_DOC = "Backoff time in seconds between forbidden requests.";
+  private static final String FITBIT_FORBIDDEN_BACKOFF_DISPLAY = "Forbidden backoff time (s)";
+  private static final int FITBIT_FORBIDDEN_BACKOFF_DEFAULT = 86400; // 24 hours
+
 
   private UserRepository userRepository;
   private final Headers clientCredentials;
@@ -699,6 +703,18 @@ public String toString() {
             ++orderInGroup,
             Width.SHORT,
             FITBIT_MAX_FORBIDDEN_DISPLAY)
+        
+        .define(FITBIT_FORBIDDEN_BACKOFF_CONFIG,
+            Type.INT,
+            FITBIT_FORBIDDEN_BACKOFF_DEFAULT,
+            Importance.MEDIUM,
+            FITBIT_FORBIDDEN_BACKOFF_DOC,
+            group,
+            ++orderInGroup,
+            Width.SHORT,
+            FITBIT_FORBIDDEN_BACKOFF_DISPLAY)
+        
+        
         ;
   }
 
@@ -902,4 +918,8 @@ public URL getFitbitUserRepositoryTokenUrl() {
   public int getMaxForbidden() {
     return getInt(FITBIT_MAX_FORBIDDEN_CONFIG);
   }
+
+  public int getForbiddenBackoff() {
+    return getInt(FITBIT_FORBIDDEN_BACKOFF_CONFIG);
+  }
 }

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

@@ -17,21 +17,16 @@
 
 package org.radarbase.connect.rest.fitbit.route;
 
-import static java.time.ZoneOffset.UTC;
-import static java.time.temporal.ChronoUnit.DAYS;
-import static java.time.temporal.ChronoUnit.MINUTES;
-import static java.time.temporal.ChronoUnit.NANOS;
-import static java.time.temporal.ChronoUnit.SECONDS;
-import static org.radarbase.connect.rest.converter.PayloadToSourceRecordConverter.MIN_INSTANT;
-import static org.radarbase.connect.rest.converter.PayloadToSourceRecordConverter.TIMESTAMP_OFFSET_KEY;
-import static org.radarbase.connect.rest.converter.PayloadToSourceRecordConverter.nearFuture;
-import static org.radarbase.connect.rest.request.PollingRequestRoute.max;
-
 import java.io.IOException;
 import java.time.Duration;
 import java.time.Instant;
+import static java.time.ZoneOffset.UTC;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
+import static java.time.temporal.ChronoUnit.DAYS;
+import static java.time.temporal.ChronoUnit.MINUTES;
+import static java.time.temporal.ChronoUnit.NANOS;
+import static java.time.temporal.ChronoUnit.SECONDS;
 import java.time.temporal.TemporalAmount;
 import java.util.AbstractMap;
 import java.util.Comparator;
@@ -42,11 +37,13 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
-import okhttp3.Request;
-import okhttp3.Response;
+
 import org.apache.kafka.connect.source.SourceRecord;
 import org.apache.kafka.connect.storage.OffsetStorageReader;
 import org.radarbase.connect.rest.RestSourceConnectorConfig;
+import static org.radarbase.connect.rest.converter.PayloadToSourceRecordConverter.MIN_INSTANT;
+import static org.radarbase.connect.rest.converter.PayloadToSourceRecordConverter.TIMESTAMP_OFFSET_KEY;
+import static org.radarbase.connect.rest.converter.PayloadToSourceRecordConverter.nearFuture;
 import org.radarbase.connect.rest.fitbit.FitbitRestSourceConnectorConfig;
 import org.radarbase.connect.rest.fitbit.request.FitbitRequestGenerator;
 import org.radarbase.connect.rest.fitbit.request.FitbitRestRequest;
@@ -55,10 +52,14 @@
 import org.radarbase.connect.rest.fitbit.user.UserRepository;
 import org.radarbase.connect.rest.fitbit.util.DateRange;
 import org.radarbase.connect.rest.request.PollingRequestRoute;
+import static org.radarbase.connect.rest.request.PollingRequestRoute.max;
 import org.radarbase.connect.rest.request.RestRequest;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import okhttp3.Request;
+import okhttp3.Response;
+
 /**
  * Route for regular polling.
  *
@@ -118,6 +119,7 @@ public abstract class FitbitPollingRoute implements PollingRequestRoute {
   private Duration tooManyRequestsCooldown;
   private final Map<String, Integer> forbidden403Counter;
   private int maxForbiddenResponses;
+  private Duration forbidden403Cooldown;
 
   public FitbitPollingRoute(
       FitbitRequestGenerator generator,
@@ -144,6 +146,7 @@ public void initialize(RestSourceConnectorConfig config) {
         .minus(getPollIntervalPerUser());
     this.maxForbiddenResponses = fitbitConfig.getMaxForbidden();
     this.converter().initialize(fitbitConfig);
+    this.forbidden403Cooldown = Duration.ofSeconds(fitbitConfig.getForbiddenBackoff());
   }
 
   @Override
@@ -190,7 +193,7 @@ public void requestFailed(RestRequest request, Response response) {
       String userId = user.getId();
       int count = forbidden403Counter.compute(userId, (k, v) -> v == null ? 1 : v + 1);
       if (count >= maxForbiddenResponses) {
-        Instant backOff = lastPoll.plus(ONE_DAY);
+        Instant backOff = lastPoll.plus(this.forbidden403Cooldown);
         lastPollPerUser.put(userId, backOff);
         forbidden403Counter.remove(userId);
         logger.warn("User {} reached max 403 responses for route {}. Backing off until {}",