rework the way that writes to files are managed. Remove this from the…

… RecordGrouper, and more a something that tells the caller the group, not manages the group.

Allow full files to be written in the background
Allow files to be written when we reach a timeout (i.e. a max delay for a record)
Update the kafka commits when we have written files, rather than waiting for commit to be forced by timeout
Avoid OOM issues by having back pressure, so that we can flush or cause earlier writes if we have too many buffered records
Provide different writer models so that we can write data before the flush (removing the memory pressure potentially, depending on the writer)

commons/src/main/java/io/aiven/kafka/connect/common/grouper/RecordStreamer.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2020 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.common.grouper;
+
+import org.apache.kafka.connect.sink.SinkRecord;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * The interface for classes that associates {@link SinkRecord}s with files by some criteria.
+ */
+public interface RecordStreamer{
+    /**
+     * determine the logical grouping of the record
+     *
+     * @param record
+     *            - record to group
+     */
+    String getStream(SinkRecord record);
+
+    /**
+     * determine the actual filename of the record
+     *
+     * @param record
+     *            - record to drive the filename
+     */
+    String getFilename(SinkRecord record);
+}

gcs-sink-connector/src/main/java/io/aiven/kafka/connect/gcs/AsyncGcsSinkConfig.java

@@ -0,0 +1,43 @@
+package io.aiven.kafka.connect.gcs;
+
+import io.aiven.kafka.connect.common.config.validators.ClassValidator;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.connect.connector.Task;
+
+import java.util.Map;
+
+public class AsyncGcsSinkConfig extends GcsSinkConfig{
+
+    private static final String GROUP_ASYNC = "Async";
+    private static final String ASYNC_MAX_RECORD_AGE = "async.file.max.record.age.ms";
+    private static final String ASYNC_MAX_OPEN_FILES = "async.file.max.open.files";
+    private static final String ASYNC_TASK_CLASS = "async.task.class";
+
+    public AsyncGcsSinkConfig(Map<String, String> properties) {
+        super(configDef(), properties);
+    }
+
+    public static ConfigDef configDef() {
+        final ConfigDef configDef = GcsSinkConfig.configDef();
+        addAsyncConfig(configDef);
+        return configDef;
+    }
+    private static void addAsyncConfig(final ConfigDef configDef) {
+        int groupCounter = 0;
+        configDef.define(ASYNC_MAX_RECORD_AGE, ConfigDef.Type.INT, 60000, ConfigDef.Importance.LOW,
+                "write files asynchronously", GROUP_ASYNC, groupCounter++, ConfigDef.Width.NONE, ASYNC_MAX_RECORD_AGE);
+        configDef.define(ASYNC_MAX_OPEN_FILES, ConfigDef.Type.INT, 100, ConfigDef.Importance.LOW,
+                "write files asynchronously", GROUP_ASYNC, groupCounter++, ConfigDef.Width.NONE, ASYNC_MAX_OPEN_FILES);
+        configDef.define(ASYNC_TASK_CLASS, ConfigDef.Type.CLASS,  AsyncGcsSinkTask.class, new ClassValidator(AsyncGcsSinkTask.class),
+                ConfigDef.Importance.LOW,"the task class", GROUP_ASYNC, groupCounter++, ConfigDef.Width.NONE, ASYNC_TASK_CLASS);
+    }
+
+
+    public int getAsyncMaxRecordAgeMs() {
+        return getInt(ASYNC_MAX_RECORD_AGE);
+    }
+
+    public Class<? extends Task> getTaskClass() {
+        return getClass(ASYNC_MAX_RECORD_AGE).asSubclass(AsyncGcsSinkTask.class);
+    }
+}

gcs-sink-connector/src/main/java/io/aiven/kafka/connect/gcs/AsyncGcsSinkConnector.java

@@ -0,0 +1,29 @@
+package io.aiven.kafka.connect.gcs;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.connect.connector.Task;
+
+import java.util.Map;
+
+public class AsyncGcsSinkConnector extends GcsSinkConnector {
+    private AsyncGcsSinkConfig config;
+    @Override
+    public Class<? extends Task> taskClass() {
+        return config.getTaskClass();
+    }
+
+    protected void setConfig(AsyncGcsSinkConfig config) {
+        this.config = config ;
+        super.setConfig(config);
+    }
+
+    @Override
+    protected void parseConfig(Map<String, String> props) {
+        setConfig(new AsyncGcsSinkConfig(props));
+    }
+
+    @Override
+    public ConfigDef config() {
+        return AsyncGcsSinkConfig.configDef();
+    }
+}