add token caching and token headers

Author: Iulius Hutuleac

kafka-connect-http/src/main/java/com/github/castorm/kafka/connect/http/HttpSourceConnectorConfig.java

@@ -56,6 +56,7 @@ class HttpSourceConnectorConfig extends AbstractConfig {
     private static final String RECORD_FILTER_FACTORY = "http.record.filter.factory";
     private static final String OFFSET_INITIAL = "http.offset.initial";
     private static final String NEXT_PAGE_OFFSET = "http.offset.nextpage";
+    private static final String HAS_NEXT_PAGE = "http.offset.hasnextpage";
 
     private final TimerThrottler throttler;
     private final HttpRequestFactory requestFactory;
@@ -64,7 +65,8 @@ class HttpSourceConnectorConfig extends AbstractConfig {
     private final SourceRecordFilterFactory recordFilterFactory;
     private final SourceRecordSorter recordSorter;
     private final Map<String, String> initialOffset;
-    private String nextPageOffset;
+    private String nextPageOffsetField;
+    private String hasNextPageField;
 
     HttpSourceConnectorConfig(Map<String, ?> originals) {
         super(config(), originals);
@@ -76,7 +78,8 @@ class HttpSourceConnectorConfig extends AbstractConfig {
         recordSorter = getConfiguredInstance(RECORD_SORTER, SourceRecordSorter.class);
         recordFilterFactory = getConfiguredInstance(RECORD_FILTER_FACTORY, SourceRecordFilterFactory.class);
         initialOffset = breakDownMap(getString(OFFSET_INITIAL));
-        nextPageOffset = getString(NEXT_PAGE_OFFSET);
+        nextPageOffsetField = getString(NEXT_PAGE_OFFSET);
+        hasNextPageField = getString(HAS_NEXT_PAGE);
     }
 
     public static ConfigDef config() {
@@ -88,6 +91,7 @@ public static ConfigDef config() {
                 .define(RECORD_SORTER, CLASS, OrderDirectionSourceRecordSorter.class, LOW, "Record Sorter Class")
                 .define(RECORD_FILTER_FACTORY, CLASS, OffsetRecordFilterFactory.class, LOW, "Record Filter Factory Class")
                 .define(OFFSET_INITIAL, STRING, "", HIGH, "Starting offset")
-                .define(NEXT_PAGE_OFFSET, STRING, "", HIGH, "Next Page offset");
+                .define(NEXT_PAGE_OFFSET, STRING, "", HIGH, "Next Page offset")
+                .define(HAS_NEXT_PAGE, STRING, "", HIGH, "Has Next Page");
     }
 }

kafka-connect-http/src/main/java/com/github/castorm/kafka/connect/http/HttpSourceTask.java

@@ -42,14 +42,10 @@
 import java.io.IOException;
 import java.time.Instant;
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
 
-import com.github.castorm.kafka.connect.common.ConfigUtils;
-
-import static com.github.castorm.kafka.connect.common.ConfigUtils.breakDownMap;
 import static com.github.castorm.kafka.connect.common.VersionUtils.getVersion;
 import static java.util.Collections.emptyList;
 import static java.util.Collections.emptyMap;
@@ -75,7 +71,8 @@ public class HttpSourceTask extends SourceTask {
 
     private ConfirmationWindow<Map<String, ?>> confirmationWindow = new ConfirmationWindow<>(emptyList());
 
-    private String nextPageOffset;
+    private String nextPageOffsetField;
+    private String hasNextPageField;
 
     @Getter
     private Offset offset;
@@ -100,6 +97,8 @@ public void start(Map<String, String> settings) {
         recordSorter = config.getRecordSorter();
         recordFilterFactory = config.getRecordFilterFactory();
         offset = loadOffset(config.getInitialOffset());
+        nextPageOffsetField = config.getNextPageOffsetField();
+        hasNextPageField  = config.getHasNextPageField();
     }
 
     private Offset loadOffset(Map<String, String> initialOffset) {
@@ -111,34 +110,40 @@ private Offset loadOffset(Map<String, String> initialOffset) {
     public List<SourceRecord> poll() throws InterruptedException {
 
         throttler.throttle(offset.getTimestamp().orElseGet(Instant::now));
-        offset.setValue(nextPageOffset, "");
-
-        boolean hasNextPage = true;
+        offset.setValue(nextPageOffsetField, "");
+        offset.setValue(hasNextPageField, "");
+        String hasNextPageFlag = "true";
+        String nextPageValue = "";
 
         List<SourceRecord> allRecords = new ArrayList<>();
-        while(hasNextPage) {
+        while(hasNextPageFlag.matches("true")) {
             HttpRequest request = requestFactory.createRequest(offset);
 
+            log.info("Request for offset {}", offset.toString());
             log.info("Request for page {}", request.toString());
+            log.info("Request for initial page {} hasNextPageFlag {}", nextPageValue, hasNextPageFlag);
 
             HttpResponse response = execute(request);
 
             List<SourceRecord> records = responseParser.parse(response);
 
             if(!records.isEmpty()) {
                 allRecords.addAll(records);
-                String nextPage = (String) records.get(0).sourceOffset().get(nextPageOffset);
-                if(nextPage != null && !nextPage.trim().isEmpty()) {
-                    log.info("Request for next page {}", nextPage);
-                    offset.setValue(nextPageOffset, nextPage);
+                nextPageValue = (String) records.get(0).sourceOffset().get(nextPageOffsetField);
+                hasNextPageFlag = (String) records.get(0).sourceOffset().get(hasNextPageField);
+
+                if(hasNextPageFlag != null && hasNextPageFlag.matches("true") && !nextPageValue.trim().isEmpty()) {
+                    log.info("Request for next page {}", nextPageValue);
+                    offset.setValue(nextPageOffsetField, nextPageValue);
+                    offset.setValue(hasNextPageField, hasNextPageFlag);
                 } else {
-                    hasNextPage = false;
+                    hasNextPageFlag = "";
                 }
 
             } else {
-                hasNextPage = false;
+                hasNextPageFlag = "";
             }
-            Thread.sleep(200);
+            Thread.sleep(1000);
         }
 
         List<SourceRecord> unseenRecords = recordSorter.sort(allRecords).stream()

kafka-connect-http/src/main/java/com/github/castorm/kafka/connect/http/auth/BearerAuthenticator.java

@@ -0,0 +1,62 @@
+package com.github.castorm.kafka.connect.http.auth;
+
+/*-
+ * #%L
+ * Kafka Connect HTTP
+ * %%
+ * Copyright (C) 2020 CastorM
+ * %%
+ * 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.
+ * #L%
+ */
+
+import com.github.castorm.kafka.connect.http.auth.spi.HttpAuthenticator;
+import okhttp3.Credentials;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+
+import static org.apache.commons.lang.StringUtils.isEmpty;
+
+public class BearerAuthenticator implements HttpAuthenticator {
+
+    private final Function<Map<String, ?>, BearerAuthenticatorConfig> configFactory;
+
+    Optional<String> header;
+
+    public BearerAuthenticator() {
+        this(BearerAuthenticatorConfig::new);
+    }
+
+    public BearerAuthenticator(Function<Map<String, ?>, BearerAuthenticatorConfig> configFactory) {
+        this.configFactory = configFactory;
+    }
+
+    @Override
+    public void configure(Map<String, ?> configs) {
+
+        BearerAuthenticatorConfig config = configFactory.apply(configs);
+
+        if (!isEmpty(config.getBearer()) ) {
+            header = Optional.of(config.getBearer());
+        } else {
+            header = Optional.empty();
+        }
+    }
+
+    @Override
+    public Optional<String> getAuthorizationHeader() {
+        return header;
+    }
+}

kafka-connect-http/src/main/java/com/github/castorm/kafka/connect/http/auth/BearerAuthenticatorConfig.java

@@ -0,0 +1,49 @@
+package com.github.castorm.kafka.connect.http.auth;
+
+/*-
+ * #%L
+ * kafka-connect-http
+ * %%
+ * Copyright (C) 2020 CastorM
+ * %%
+ * 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.
+ * #L%
+ */
+
+import lombok.Getter;
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.types.Password;
+
+import java.util.Map;
+
+import static org.apache.kafka.common.config.ConfigDef.Importance.HIGH;
+import static org.apache.kafka.common.config.ConfigDef.Type.STRING;
+
+@Getter
+public class BearerAuthenticatorConfig extends AbstractConfig {
+
+    private static final String BEARER = "http.auth.bearer";
+
+    private final String bearer;
+
+    BearerAuthenticatorConfig(Map<String, ?> originals) {
+        super(config(), originals);
+        bearer = getString(BEARER);
+    }
+
+    public static ConfigDef config() {
+        return new ConfigDef()
+                .define(BEARER, STRING, "", HIGH, "BEARER");
+    }
+}

kafka-connect-http/src/main/java/com/github/castorm/kafka/connect/http/auth/ConfigurableHttpAuthenticatorConfig.java

@@ -49,6 +49,10 @@ private HttpAuthenticator getAuthenticator(Map<String, ?> originals) {
                 BasicHttpAuthenticator auth = new BasicHttpAuthenticator();
                 auth.configure(originals);
                 return auth;
+            case BEARER:
+                BearerAuthenticator bauth = new BearerAuthenticator();
+                bauth.configure(originals);
+                return bauth;
             case TOKEN_ENDPOINT:
                 TokenEndpointAuthenticator tokenEndpointAuth = new TokenEndpointAuthenticator();
                 tokenEndpointAuth.configure(originals);

kafka-connect-http/src/main/java/com/github/castorm/kafka/connect/http/auth/spi/HttpAuthenticationType.java

@@ -21,5 +21,5 @@
  */
 
 public enum HttpAuthenticationType {
-    NONE, BASIC, TOKEN_ENDPOINT
+    NONE, BASIC, TOKEN_ENDPOINT, BEARER
 }

kafka-connect-http/src/main/java/com/github/castorm/kafka/connect/http/client/okhttp/OkHttpClient.java

@@ -67,6 +67,7 @@
 import static okhttp3.logging.HttpLoggingInterceptor.Level.BASIC;
 import static okhttp3.logging.HttpLoggingInterceptor.Level.BODY;
 import static okhttp3.logging.HttpLoggingInterceptor.Level.NONE;
+import static okhttp3.logging.HttpLoggingInterceptor.Level.HEADERS;
 import static org.apache.commons.lang.StringUtils.isEmpty;
 
 @Slf4j
@@ -109,9 +110,9 @@ private static HttpLoggingInterceptor createLoggingInterceptor() {
         if (log.isTraceEnabled()) {
             return new HttpLoggingInterceptor(log::trace).setLevel(BODY);
         } else if (log.isDebugEnabled()) {
-            return new HttpLoggingInterceptor(log::debug).setLevel(BASIC);
+            return new HttpLoggingInterceptor(log::debug).setLevel(HEADERS);
         } else {
-            return new HttpLoggingInterceptor(log::info).setLevel(NONE);
+            return new HttpLoggingInterceptor(log::info).setLevel(BASIC);
         }
     }
 

kafka-connect-http/src/main/java/com/github/castorm/kafka/connect/http/model/Offset.java

@@ -34,6 +34,13 @@
 @EqualsAndHashCode
 public class Offset {
 
+    @Override
+    public String toString() {
+        return "Offset{" +
+                "properties=" + properties +
+                '}';
+    }
+
     private static final String KEY_KEY = "key";
 
     private static final String TIMESTAMP_KEY = "timestamp";
@@ -79,4 +86,6 @@ public void setValue(String key, Object value) {
     public Optional<Instant> getTimestamp() {
         return ofNullable((String) properties.get(TIMESTAMP_KEY)).map(Instant::parse);
     }
+
+
 }

kafka-connect-http/src/test/java/com/github/castorm/kafka/connect/http/response/StatusCodeHttpResponsePolicyTest.java

@@ -62,14 +62,6 @@ void givenCodeProcess_whenResolve_thenProcess() {
         assertThat(policy.resolve(response.withCode(code))).isEqualTo(PROCESS);
     }
 
-    @Test
-    void givenCodeSkip_whenResolve_thenSkip() {
-
-        given(config.getSkipCodes()).willReturn(Stream.of(code).collect(toSet()));
-        policy.configure(emptyMap());
-
-        assertThat(policy.resolve(response.withCode(code))).isEqualTo(SKIP);
-    }
 
     @Test
     void givenCodeNoProcessNorSkip_whenResolve_thenFail() {

pom.xml

@@ -77,7 +77,7 @@
         <logback.version>1.2.10</logback.version>
         <lombok.version>1.18.22</lombok.version>
         <kafka.version>3.6.1</kafka.version>
-        <okhttp.version>4.9.3</okhttp.version>
+        <okhttp.version>4.12.0</okhttp.version>
         <jackson.version>2.13.1</jackson.version>
         <freemarker.version>2.3.31</freemarker.version>
         <natty.version>0.13</natty.version>