Rework rest request to convert records out of requests

Author: blootsvoets

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

@@ -33,8 +33,7 @@
 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;
@@ -71,12 +70,11 @@ 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);

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);

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

@@ -19,12 +19,11 @@
 
 import static java.lang.System.currentTimeMillis;
 
-import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Map;
-import okhttp3.Response;
-import okhttp3.ResponseBody;
+import okhttp3.Headers;
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.source.SourceRecord;
 import org.radarbase.connect.rest.RestSourceConnectorConfig;
@@ -35,10 +34,9 @@ public class StringPayloadConverter implements PayloadToSourceRecordConverter {
   private TopicSelector topicSelector;
 
   @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();
-    String result = body == null ? null : body.string();
+    String result = data == null ? null : new String(data, StandardCharsets.UTF_8);
     String topic = topicSelector.getTopic(request, result);
     return Collections.singleton(
         new SourceRecord(request.getPartition(), sourceOffset, topic,

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

@@ -22,9 +22,11 @@
 import java.util.Map;
 import java.util.function.Predicate;
 import java.util.stream.Stream;
+import okhttp3.Headers;
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
 import okhttp3.Response;
+import okhttp3.ResponseBody;
 import org.apache.kafka.connect.source.SourceRecord;
 
 /**
@@ -82,18 +84,24 @@ public Stream<SourceRecord> handleRequest() throws IOException {
 
     Collection<SourceRecord> records;
 
+    byte[] data;
+    Headers headers;
     try (Response response = client.newCall(request).execute()) {
       if (!response.isSuccessful()) {
         route.requestFailed(this, response);
         return Stream.empty();
       }
 
-      records = route.converter().convert(this, response);
+      headers = response.headers();
+      ResponseBody body = response.body();
+      data = body != null ? body.bytes() : null;
     } catch (IOException ex) {
       route.requestFailed(this, null);
       throw ex;
     }
 
+    records = route.converter().convert(this, headers, data);
+
     if (records.isEmpty()) {
       route.requestEmpty(this);
     } else {