Code refactoring

1. Moved RecordSender class into package io.aiven.kafka.connect.http.recordsender package
2. Moved HttpSender into io.aiven.kafka.connect.http.sender package
   and split up HttpSender behave for 2 diff auth type with auth and
   without auth
3. Moved backoff policy into HtpSender

src/integration-test/java/io/aiven/kafka/connect/http/AvroIntegrationTest.java

@@ -44,7 +44,6 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 import org.slf4j.Logger;
@@ -149,7 +148,6 @@ final void tearDown() {
 
     @Test
     @Timeout(30)
-    @Disabled
     final void testBasicDelivery() throws ExecutionException, InterruptedException {
         final BodyRecorderHandler bodyRecorderHandler = new BodyRecorderHandler();
         mockServer.addHandler(bodyRecorderHandler);

src/integration-test/java/io/aiven/kafka/connect/http/IntegrationTest.java

@@ -204,7 +204,7 @@ final void testFailingEvery3rdRequest() throws ExecutionException, InterruptedEx
 
         TestUtils.waitForCondition(
             () -> bodyRecorderHandler.recorderBodies().size() >= expectedBodies.size(),
-            10000,
+            10_000,
             "All requests received by HTTP server"
         );
         log.info("Recorded request bodies: {}", bodyRecorderHandler.recorderBodies());

src/main/java/io/aiven/kafka/connect/http/HttpSinkTask.java

@@ -20,12 +20,13 @@
 import java.util.Map;
 import java.util.Objects;
 
-import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.errors.DataException;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.apache.kafka.connect.sink.SinkTask;
 
 import io.aiven.kafka.connect.http.config.HttpSinkConfig;
+import io.aiven.kafka.connect.http.recordsender.RecordSender;
+import io.aiven.kafka.connect.http.sender.HttpSender;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -38,32 +39,21 @@ public final class HttpSinkTask extends SinkTask {
 
     // required by Connect
     public HttpSinkTask() {
+        this(null);
     }
 
-    // for testing
-    HttpSinkTask(final HttpSender httpSender) {
+    protected HttpSinkTask(final HttpSender httpSender) {
         this.httpSender = httpSender;
     }
 
     @Override
     public void start(final Map<String, String> props) {
         Objects.requireNonNull(props);
-
         final var config = new HttpSinkConfig(props);
-
         if (this.httpSender == null) {
-            this.httpSender = new HttpSender(
-                config.httpUrl(),
-                config.headerAuthorization(),
-                config.headerContentType());
-        }
-
-        if (config.batchingEnabled()) {
-            recordSender = new BatchRecordSender(httpSender,
-                config.batchMaxSize(), config.maxRetries(), config.retryBackoffMs());
-        } else {
-            recordSender = new SingleRecordSender(httpSender, config.maxRetries(), config.retryBackoffMs());
+            this.httpSender = HttpSender.createHttpSender(config);
         }
+        recordSender = RecordSender.createRecordSender(httpSender, config);
     }
 
     @Override
@@ -76,11 +66,7 @@ public void put(final Collection<SinkRecord> records) {
                     throw new DataException("Record value must not be null");
                 }
             }
-            try {
-                recordSender.send(records);
-            } catch (final InterruptedException e) {
-                throw new ConnectException(e);
-            }
+            recordSender.send(records);
         }
     }
 
@@ -93,4 +79,5 @@ public void stop() {
     public String version() {
         return Version.VERSION;
     }
+
 }

src/main/java/io/aiven/kafka/connect/http/config/HttpSinkConfig.java

@@ -90,7 +90,7 @@ public String toString() {
         configDef.define(
             HTTP_AUTHORIZATION_TYPE_CONFIG,
             ConfigDef.Type.STRING,
-            ConfigDef.NO_DEFAULT_VALUE,
+            AuthorizationType.NONE.name,
             new ConfigDef.Validator() {
                 @Override
                 public void ensureValid(final String name, final Object value) {

src/main/java/io/aiven/kafka/connect/http/BatchRecordSender.java → src/main/java/io/aiven/kafka/connect/http/recordsender/BatchRecordSender.java

@@ -14,42 +14,41 @@
  * limitations under the License.
  */
 
-package io.aiven.kafka.connect.http;
+package io.aiven.kafka.connect.http.recordsender;
 
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 
 import org.apache.kafka.connect.sink.SinkRecord;
 
+import io.aiven.kafka.connect.http.sender.HttpSender;
+
 final class BatchRecordSender extends RecordSender {
     private static final String BATCH_RECORD_SEPARATOR = "\n";
 
     private final int batchMaxSize;
 
-    BatchRecordSender(final HttpSender httpSender,
-                      final int batchMaxSize,
-                      final int maxRetries, final int retryBackoffMs) {
-        super(httpSender, maxRetries, retryBackoffMs);
+    protected BatchRecordSender(final HttpSender httpSender, final int batchMaxSize) {
+        super(httpSender);
         this.batchMaxSize = batchMaxSize;
     }
 
     @Override
-    void send(final Collection<SinkRecord> records) throws InterruptedException {
+    public void send(final Collection<SinkRecord> records) {
         final List<SinkRecord> batch = new ArrayList<>(batchMaxSize);
         for (final var record : records) {
             batch.add(record);
             if (batch.size() >= batchMaxSize) {
                 final String body = createRequestBody(batch);
                 batch.clear();
-
-                sendWithRetries(body);
+                httpSender.send(body);
             }
         }
 
         if (!batch.isEmpty()) {
             final String body = createRequestBody(batch);
-            sendWithRetries(body);
+            httpSender.send(body);
         }
     }
 

src/main/java/io/aiven/kafka/connect/http/recordsender/RecordSender.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2019 Aiven Oy
+ *
+ * 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.
+ */
+
+package io.aiven.kafka.connect.http.recordsender;
+
+import java.util.Collection;
+
+import org.apache.kafka.connect.sink.SinkRecord;
+
+import io.aiven.kafka.connect.http.config.HttpSinkConfig;
+import io.aiven.kafka.connect.http.converter.RecordValueConverter;
+import io.aiven.kafka.connect.http.sender.HttpSender;
+
+public abstract class RecordSender {
+
+    protected final HttpSender httpSender;
+
+    protected final RecordValueConverter recordValueConverter = new RecordValueConverter();
+
+    protected RecordSender(final HttpSender httpSender) {
+        this.httpSender = httpSender;
+    }
+
+    public abstract void send(final Collection<SinkRecord> records);
+
+    public static RecordSender createRecordSender(final HttpSender httpSender, final HttpSinkConfig config) {
+        if (config.batchingEnabled()) {
+            return new BatchRecordSender(httpSender, config.batchMaxSize());
+        } else {
+            return new SingleRecordSender(httpSender);
+        }
+    }
+
+}