---
version: '2'
services:
  kafka-ui:
    container_name: kafka-ui
    image: provectuslabs/kafka-ui:v0.7.1
    ports:
      - 8080:8080
    depends_on:
      - kafka0
      - schema-registry0
    environment:
        ...
  kafka0:
    image: confluentinc/cp-kafka:7.2.1.arm64
    hostname: kafka0
    container_name: kafka0
    ports:
      - 9092:9092
      - 9997:9997
    environment:
        ...
    volumes:
        ...
  schema-registry0:
    image: confluentinc/cp-schema-registry:7.2.1.arm64
    ports:
      - 8085:8085
    depends_on:
      - kafka0
    environment:
        ...
  kafka-init-topics:
    image: confluentinc/cp-kafka:7.2.1.arm64
    volumes:
       - ./message.json:/data/message.json
    depends_on:
      - kafka0
    command: ...

  kafka-connect0:
    image: my-connector:latest
    ports:
      - 8083:8083
    depends_on:
      - kafka0
    environment:
        ...

FROM confluentinc/cp-kafka-connect:7.2.1
COPY ./plugins/ /usr/share/java
ENV CONNECT_PLUGIN_PATH="/usr/share/java,/usr/share/confluent-hub-components"
USER 1001

.
├── Dockerfile
└── plugins
   └── connector-all.jar

description = "Kafka Connector Module"

plugins {
    id(Plugins.gradle_shadow) version Plugins.Versions.gradle_shadow
}

dependencies {
    implementation(Dependencies.kafka_connect_api)
    implementation(Dependencies.google_api_client)
    implementation(Dependencies.google_sheet_api)
    implementation(Dependencies.google_oauth_client)

    implementation(Dependencies.slf4j)

    compileOnly(Dependencies.lombok)
    annotationProcessor(Dependencies.lombok)
}

tasks.test {
    useJUnitPlatform()
}

java {
    toolchain {
        sourceCompatibility = JavaVersion.VERSION_11
        targetCompatibility = JavaVersion.VERSION_11
    }
}

dependencies {
    implementation(Dependencies.kafka_connect_api)
    implementation(Dependencies.google_api_client)
    implementation(Dependencies.google_sheet_api)
    implementation(Dependencies.google_oauth_client)

    implementation(Dependencies.slf4j)

    compileOnly(Dependencies.lombok)
    annotationProcessor(Dependencies.lombok)
}

//https://github.com/brewagebear/blog-example/blob/main/kafka-example/connector/src/main/java/io/github/brewagebear/SimpleGoogleSheetSourceConfig.java
public class SimpleGoogleSheetSourceConfig extends AbstractConfig implements CustomConnectConfig {

    public static final String TASK_ID = "task.id";
    public static final String TASK_MAX = "task.max";
    public static final String TOPIC_NAME_CONFIG = "topic";
    public static final String POLL_INTERVAL_CONFIG = "poll.interval.ms";
    public static final long POLL_INTERVAL_DEFAULT = 1000;

    public SimpleGoogleSheetSourceConfig(final Map<?, ?> props) {
        super(CONFIG_DEF, props);
        log.info("{}", props);
    }

    public Map<String, String> getGoogleCredential() {
        ...
        return map;
    }

    public int getFetchSize() { return getInt(GOOGLE_SPREAD_SHEET_FETCH_SIZE_CONFIG); }

    public String getTopicName() {
        return getString(TOPIC_NAME_CONFIG);
    }

    public long getPollInterval() {
        return getLong(POLL_INTERVAL_CONFIG);
    }

//https://github.com/brewagebear/blog-example/blob/main/kafka-example/connector/src/main/java/io/github/brewagebear/CustomConnectConfig.java
public interface CustomConnectConfig {
    ConfigDef CONFIG_DEF =
            new ConfigDef()
                    .define(SimpleGoogleSheetSourceConfig.TOPIC_NAME_CONFIG,
                            ConfigDef.Type.STRING,
                            ConfigDef.Importance.HIGH,
                            "Topic name.")
                    .define(SimpleGoogleSheetSourceConfig.POLL_INTERVAL_CONFIG,
                            ConfigDef.Type.LONG,
                            SimpleGoogleSheetSourceConfig.POLL_INTERVAL_DEFAULT,
                            ConfigDef.Importance.HIGH,
                            "Max interval between messages (ms)")
                    .define(GoogleCredentialsConfig.GOOGLE_SERVICE_PROJECT_ID_CONFIG,
                            ConfigDef.Type.STRING,
                            ConfigDef.Importance.HIGH,
                            "Google Cloud Platform Project id")
                 ...

//https://github.com/brewagebear/blog-example/blob/main/kafka-example/connector/src/main/java/io/github/brewagebear/SimpleGoogleSheetConnector.java
public class SimpleGoogleSheetConnector extends SourceConnector {

    private Map<String, String> configProps;

    @Override
    public void start(Map<String, String> props) {
        log.info("Starting Google Sheet Source -> {}", props);
        configProps = props;
    }

    @Override
    public Class<? extends Task> taskClass() {
        return SimpleGoogleSheetTask.class;
    }

    @Override
    public List<Map<String, String>> taskConfigs(int maxTasks) {
        log.info("Setting task configurations for {} workers.", maxTasks);

        final List<Map<String, String>> configs = new ArrayList<>(maxTasks);

        for (int i = 0; i < maxTasks; ++i) {
            final Map<String, String> taskConfig = new HashMap<>(configProps);
            // add task specific values
            taskConfig.put(TASK_ID, String.valueOf(i));
            taskConfig.put(TASK_MAX, String.valueOf(maxTasks));
            configs.add(taskConfig);
        }
        return configs;
    }

    @Override
    public void stop() {
        log.info("Stopping Google Sheet Source");
    }

    @Override
    public ConfigDef config() {
        return SimpleGoogleSheetSourceConfig.CONFIG_DEF;
    }

    @Override
    public String version() {
        return "1.0";
    }
}

//https://github.com/brewagebear/blog-example/blob/main/kafka-example/connector/src/main/java/io/github/brewagebear/SimpleGoogleSheetTask.java
public class SimpleGoogleSheetTask extends SourceTask {
    ...

    public SimpleGoogleSheetTask() {
        this.time = new SystemTime();
        this.lastPollMs = time.milliseconds();
    }

    @Override
    public void start(Map<String, String> props) {
        log.info("Starting Google Sheet task with config: {}", props);
        config = new SimpleGoogleSheetSourceConfig(props);
        googleSheetService = new GoogleSheetService(config.getGoogleCredential());

        taskId = props.get("task.id");
        range = getNextRange(2);
        fetchSize = config.getFetchSize();

        sheets = googleSheetService.connect();
        running.set(true);

        // get offsets for a specific task id
        final Map<String, Object> offset =
                context.offsetStorageReader().offset(Collections.singletonMap(TASK_ID, taskId));

        if (offset != null) {
            final Long currentOffset = (Long) offset.get(POSITION_NAME);
            if (currentOffset != null) {
                lastProcessedOffset = currentOffset;
            } else {
                // no position found
                lastProcessedOffset = 0L;
            }
        } else {
            // first time there is no offset.
            lastProcessedOffset = 0L;
        }
    }

    @Override
    public List<SourceRecord> poll() {
        log.debug("Polling for new data");

        final long timeSinceLastPollMs = time.milliseconds() - lastPollMs;

        if (timeSinceLastPollMs < config.getPollInterval()) {
            log.debug("Sleep, time since last poll = {}", timeSinceLastPollMs);
            time.sleep(DEFAULT_WAIT_MS);
            return null;
        }

        if (!running.get()) {
            return null;
        }

        List<SourceRecord> records = new ArrayList<>();

        try {
            List<List<String>> data = googleSheetService.getResponse(sheets, range);

            if (data != null && !data.isEmpty()) {
                for (List<String> row : data) {
                    records.add(new SourceRecord(Collections.singletonMap(TASK_ID, taskId),
                            Collections.singletonMap(POSITION_NAME, lastProcessedOffset),
                            config.getTopicName(),
                            Schema.STRING_SCHEMA, row.get(0),
                            Schema.STRING_SCHEMA, row.get(1)
                    ));

                    lastProcessedOffset += 1;
                }

                log.info("polling data {}", records);
                range = getNextRange(lastRow);
                log.info("next range {}", range);
            } else {
                stop();
            }

            lastPollMs = time.milliseconds();

        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        return records;
    }

    private String getNextRange(int lastRange) {
        updateLastRow();
        int endRow = lastRange + fetchSize - 1;
        return "A" + (lastRange) + ":B" + (endRow);
    }

    private void updateLastRow() {
        this.lastRow += fetchSize;
    }

    @Override
    public void stop() {
        running.set(false);
    }

    @Override
    public String version() {
        return "1.0";
    }
}

{
  "name": "google-sheet-source",
  "config": {
    "connector.class": "io.github.brewagebear.SimpleGoogleSheetConnector",
    "topic": "sheet-data",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "true",
    "tasks.max": "1",
    "poll.interval.ms": "50000",
    "project_id": "project_id",
    "private_key_id" : "private_key_id",
    "private_key" : "-----BEGIN PRIVATE KEY-----\n-----END PRIVATE KEY-----\n",
    "client_email" : "test.iam.gserviceaccount.com",
    "client_id": "client_id",
    "sheet_id" : "1U_Q4qWeIzrXezL8CERnfxwNNxfNZlyyVwG0mujTKJZY"
  }
}

curl -X PUT -H "Content-Type: application/json" -d @./google-sheet-source.json http://localhost:8083/connector-plugins/SimpleGoogleSheetConnector/config/validate

curl -X POST -H "Content-Type: application/json" -d @./google-sheet-source.json http://localhost:8083/connectors