fixes

Author: Iulius Hutuleac

.vscode/settings.json

@@ -0,0 +1,3 @@
+{
+    "java.configuration.updateBuildConfiguration": "automatic"
+}

kafka-connect-http-infra/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <groupId>com.github.castorm</groupId>
         <artifactId>kafka-connect-http-parent</artifactId>
-        <version>0.8.14</version>
+        <version>0.8.15</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

kafka-connect-http-test/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>kafka-connect-http-parent</artifactId>
         <groupId>com.github.castorm</groupId>
-        <version>0.8.14</version>
+        <version>0.8.15</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

kafka-connect-http/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>kafka-connect-http-parent</artifactId>
         <groupId>com.github.castorm</groupId>
-        <version>0.8.14</version>
+        <version>0.8.15</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

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

@@ -130,6 +130,7 @@ public List<SourceRecord> poll() throws InterruptedException {
         long autoOffset = 0;
         long autoDateIncrement = 0;
         long autoDateBackoff = 0;
+        boolean pagingEnabled = nextPageOffsetField != null && !nextPageOffsetField.isEmpty() && hasNextPageField != null && !hasNextPageField.isEmpty();
 
         if( autoDateInitialOffset != null && !autoDateInitialOffset.isEmpty()) {
             try {
@@ -148,11 +149,15 @@ public List<SourceRecord> poll() throws InterruptedException {
 
             offset.setValue(AUTOTIMESTAMP, autoOffset);
         }
-        offset.setValue(nextPageOffsetField, "");
-        offset.setValue(hasNextPageField, "");
+
         String hasNextPageFlag = "true";
         String nextPageValue = "";
 
+        if(pagingEnabled){
+            offset.setValue(nextPageOffsetField, "");
+            offset.setValue(hasNextPageField, "");
+        }
+
         while(hasNextPageFlag.matches("true")) {
             HttpRequest request = requestFactory.createRequest(offset);
 
@@ -166,11 +171,9 @@ public List<SourceRecord> poll() throws InterruptedException {
 
             if(!records.isEmpty()) {
                 allRecords.addAll(records);
-                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);
+                if(pagingEnabled){
+                    nextPageValue = (String) records.get(0).sourceOffset().get(nextPageOffsetField);
+                    hasNextPageFlag = (String) records.get(0).sourceOffset().get(hasNextPageField);                
                     offset.setValue(nextPageOffsetField, nextPageValue);
                     offset.setValue(hasNextPageField, hasNextPageFlag);
                 } else {

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

@@ -89,6 +89,8 @@ public void configure(Map<String, ?> configs) {
                 .connectTimeout(config.getConnectionTimeoutMillis(), MILLISECONDS)
                 .readTimeout(config.getReadTimeoutMillis(), MILLISECONDS)
                 .retryOnConnectionFailure(true)
+                .followRedirects(false)
+                .followSslRedirects(false)
                 .addInterceptor(createLoggingInterceptor())
                 .addInterceptor(chain -> chain.proceed(authorize(chain.request())))
                 .authenticator((route, response) -> authorize(response.request()))

pom.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.github.castorm</groupId>
     <artifactId>kafka-connect-http-parent</artifactId>
-    <version>0.8.14</version>
+    <version>0.8.15</version>
     <packaging>pom</packaging>
 
     <name>Kafka Connect HTTP Parent</name>