package com.jasongj.kafka.connect;

import java.util.Collection;
import java.util.Map;

import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.connect.sink.SinkRecord;
import org.apache.kafka.connect.sink.SinkTask;

public class ConsoleSourceTask extends SinkTask {

    @Override
    public String version() {
        return new ConsoleSourceConnect().version();
    }

    @Override
    public void flush(Map<TopicPartition, OffsetAndMetadata> arg0) {
        
    }

    @Override
    public void put(Collection<SinkRecord> arg0) {
        
    }

    @Override
    public void start(Map<String, String> arg0) {
        
    }

    @Override
    public void stop() {
        
    }

}

@@ -45,107 +45,190 @@
 ava.
-util.Collection;%0Aimport java.util.Map;%0A%0Aimport org.apache.kafka.clients.consumer.OffsetAndMetadata;
+io.BufferedReader;%0Aimport java.io.IOException;%0Aimport java.io.InputStream;%0Aimport java.io.InputStreamReader;%0Aimport java.util.ArrayList;%0Aimport java.util.List;%0Aimport java.util.Map;%0A
 %0Aimp
@@ -254,27 +254,25 @@
 a.co
-mmon.TopicPartition
+nnect.data.Schema
 ;%0Aim
@@ -302,24 +302,28 @@
 onnect.s
-ink.Sink
+ource.Source
 Record;%0A
@@ -359,16 +359,20 @@
 ct.s
-ink.Sink
+ource.Source
 Task
@@ -414,19 +414,21 @@
 xtends S
-ink
+ource
 Task %7B%0A%0A
@@ -435,189 +435,435 @@
     
-@Override%0A    public String version() %7B%0A        return new ConsoleSourceConnect().version();%0A    %7D%0A%0A    @Override%0A    public void flush(Map%3CTopicPartition, OffsetAndMetadata%3E arg0) 
+private InputStream inputStream;%0A    private String topic;%0A    %0A    @Override%0A    public String version() %7B%0A        return null;%0A    %7D%0A%0A    @Override%0A    public List%3CSourceRecord%3E poll() throws InterruptedException %7B%0A        try%7B%0A            BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));%0A            List%3CSourceRecord%3E records = new ArrayList%3CSourceRecord%3E();%0A            while(reader.ready())
 %7B%0A  
@@ -872,89 +872,264 @@
     
-%0A
     
-%7D%0A%0A
     
-@Override%0A    public void put(Collection%3CSinkRecord%3E arg0) %7B%0A        
+String line = reader.readLine();%0A//                SourceRecord record = new SourceRecord(null, null, topic, Schema.STRING_SCHEMA, line);%0A            %7D%0A            return records;%0A        %7D catch(IOException ex) %7B%0A            return null;%0A        %7D
 %0A   
@@ -1192,12 +1192,12 @@
 ng%3E 
-arg0
+conf
 ) %7B%0A
@@ -1196,32 +1196,91 @@
 conf) %7B%0A        
+inputStream = System.in;%0A        topic = conf.get(%22topic%22);
 %0A    %7D%0A%0A    @Ove
@@ -1326,10 +1326,11 @@
 %0A    %7D%0A%0A
+%0A
 %7D%0A

Implement ConsoleSourceTask