push

src/main/java/NotifyDemo.java

+import com.alibaba.dts.boot.MysqlRecordPrinter;
+import com.alibaba.dts.formats.avro.Operation;
+import com.alibaba.dts.common.RecordListener;
+import com.alibaba.dts.common.UserRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+
+import static com.alibaba.dts.boot.Boot.boot;
+import static com.alibaba.dts.common.Util.uncompressionObjectName;
+import static com.alibaba.dts.recordgenerator.Names.*;
+
+public class NotifyDemo {
+    private static final Logger log = LoggerFactory.getLogger(NotifyDemo.class);
+
+    public static Map<String, RecordListener> buildRecordListener() {
+        // user can impl their own listener
+        RecordListener mysqlRecordPrintListener = new RecordListener() {
+            @Override
+            public void consume(UserRecord record) {
+
+                Operation operation = record.getRecord().getOperation();
+                if (!(operation == Operation.DELETE ||
+                        operation == Operation.INSERT ||
+                        operation == Operation.UPDATE)) {
+                    record.commit(String.valueOf(record.getRecord().getSourceTimestamp()));
+                    return;
+
+                }
+                // consume record
+                // MysqlRecordPrinter show how to go through record fields and get general attributes
+                String ret = MysqlRecordPrinter.recordToString(record.getRecord());
+
+                String dbName = null;
+                String tableName = null;
+                // here we get db and table name
+                String[] dbPair = uncompressionObjectName(record.getRecord().getObjectName());
+                if (null != dbPair) {
+                    if (dbPair.length == 2) {
+                        dbName = dbPair[0];
+                        tableName = dbPair[1];
+                    } else if (dbPair.length == 3) {
+                        dbName = dbPair[0];
+                        tableName = dbPair[2];
+                    } else if (dbPair.length == 1) {
+                        dbName = dbPair[0];
+                        tableName = "";
+                    } else {
+                        throw new RuntimeException("invalid db and table name pair for record [" + record + "]");
+                    }
+                }
+
+                if (!Objects.equals(tableName, "store")) {
+                    record.commit(String.valueOf(record.getRecord().getSourceTimestamp()));
+                    return;
+                }
+
+                log.error("dbName {} tableName: {}", dbName, tableName);
+
+                log.info(ret);
+                record.commit(String.valueOf(record.getRecord().getSourceTimestamp()));
+            }
+        };
+        return Collections.singletonMap("mysqlRecordPrinter", mysqlRecordPrintListener);
+    }
+
+    /**
+     * This demo use  hard coded config. User can modify variable value for test
+     * The detailed describe for var in resources/demoConfig
+     */
+    public static Properties getConfigs() {
+        Properties properties = new Properties();
+        // user password and sid for auth
+        properties.setProperty(USER_NAME, "dtsscm");
+        properties.setProperty(PASSWORD_NAME, "A5tJKZur8fUCdluC");
+        properties.setProperty(SID_NAME, "dtsuxfc9cw6150cbg9");
+        // kafka consumer group general same with sid
+        properties.setProperty(GROUP_NAME, "dtsuxfc9cw6150cbg9");
+        // topic to consume, partition is 0
+        properties.setProperty(KAFKA_TOPIC, "cn_hangzhou_pc_bp16ydc05ds3uivel_rootdev_version2");
+        // kafka broker url
+        properties.setProperty(KAFKA_BROKER_URL_NAME, "dts-cn-hangzhou.aliyuncs.com:18001");
+        // initial checkpoint for first seek(a timestamp to set, eg 1566180200 if you want (Mon Aug 19 10:03:21 CST 2019))
+
+
+        properties.setProperty(INITIAL_CHECKPOINT_NAME, "1631700109");
+        // if force use config checkpoint when start. for checkpoint reset
+        properties.setProperty(USE_CONFIG_CHECKPOINT_NAME, "false");
+
+        // use consumer assign or subscribe interface
+        // when use subscribe mode, group config is required. kafka consumer group is enabled
+        properties.setProperty(SUBSCRIBE_MODE_NAME, "assign");
+        return properties;
+    }
+
+    public static void main(String[] args) throws InterruptedException {
+
+        try{
+            boot(getConfigs(), buildRecordListener());
+        }catch(Throwable e){
+            log.error("NotifyDemo: failed cause " + e.getMessage(), e);
+            throw e;
+        } finally {
+            System.exit(0);
+        }
+    }
+}

src/main/java/com/alibaba/dts/boot/Boot.java

+package com.alibaba.dts.boot;
+
+import com.alibaba.dts.common.RecordListener;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.log4j.PropertyConfigurator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import com.alibaba.dts.recordgenerator.OffsetCommitCallBack;
+import com.alibaba.dts.recordprocessor.EtlRecordProcessor;
+import com.alibaba.dts.recordgenerator.RecordGenerator;
+import com.alibaba.dts.recordgenerator.ConsumerWrapFactory;
+import sun.misc.Signal;
+import sun.misc.SignalHandler;
+import com.alibaba.dts.common.Checkpoint;
+import com.alibaba.dts.common.Context;
+import com.alibaba.dts.common.WorkThread;
+
+import java.io.BufferedInputStream;
+import java.io.FileInputStream;
+import java.io.InputStream;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static com.alibaba.dts.recordgenerator.Names.*;
+import static com.alibaba.dts.common.Util.*;
+
+public class Boot {
+    private static final Logger log = LoggerFactory.getLogger(Boot.class);
+    private static final AtomicBoolean existed = new AtomicBoolean(false);
+
+
+    public static void boot(String configFile, Map<String, RecordListener> recordListeners) {
+        boot(loadConfig(configFile), recordListeners);
+    }
+
+    public static void boot(Properties properties, Map<String, RecordListener> recordListeners) {
+        // first init log4j
+        initLog4j();
+        require(null != recordListeners && !recordListeners.isEmpty(), "record listener required");
+        Context context = getStreamContext(properties);
+        // check config
+        checkConfig(properties);
+        RecordGenerator recordGenerator = getRecordGenerator(context, properties);
+        EtlRecordProcessor etlRecordProcessor = getEtlRecordProcessor(context, properties, recordGenerator);
+        recordListeners.forEach((k, v) -> {
+            log.info("Boot: register record listener " + k);
+            etlRecordProcessor.registerRecordListener(k, v);
+        });
+        registerSignalHandler(context);
+        List<WorkThread> startStream = startWorker(etlRecordProcessor, recordGenerator);
+
+        while (!existed.get() ) {
+            sleepMS(1000);
+        }
+        log.info("StreamBoot: shutting down...");
+        for (WorkThread workThread : startStream) {
+            workThread.stop();
+        }
+
+    }
+
+    private static List<WorkThread> startWorker(EtlRecordProcessor etlRecordProcessor, RecordGenerator recordGenerator) {
+        List<WorkThread> ret = new LinkedList<>();
+        ret.add(new WorkThread(etlRecordProcessor));
+        ret.add(new WorkThread(recordGenerator));
+        for (WorkThread workThread : ret) {
+            workThread.start();
+        }
+        return ret;
+    }
+
+    private static void registerSignalHandler(Context context) {
+        SignalHandler signalHandler = new SignalHandler() {
+            @Override
+            public void handle(Signal signal) {
+                // SIG_INT
+                if (signal.getNumber() == 2) {
+                    existed.compareAndSet(false, true);
+                }
+            }
+        };
+        Signal.handle(new Signal("INT"), signalHandler);
+    }
+
+    private static Context getStreamContext(Properties properties) {
+        Context ret =  new Context();
+        return ret;
+    }
+
+
+    // offset@timestamp or timestamp
+    private static Checkpoint parseCheckpoint(String checkpoint) {
+        require(null != checkpoint, "checkpoint should not be null");
+        String[] offsetAndTS = checkpoint.split("@");
+        Checkpoint streamCheckpoint = null;
+        if (offsetAndTS.length == 1) {
+            streamCheckpoint =  new Checkpoint(null, Long.valueOf(offsetAndTS[0]), -1, "");
+        } else if (offsetAndTS.length >= 2) {
+            streamCheckpoint =  new Checkpoint(null, Long.valueOf(offsetAndTS[0]), Long.valueOf(offsetAndTS[1]), "");
+        }
+        return streamCheckpoint;
+    }
+
+    private static RecordGenerator getRecordGenerator(Context context, Properties properties) {
+
+        RecordGenerator recordGenerator = new RecordGenerator(properties, context,
+                parseCheckpoint(properties.getProperty(INITIAL_CHECKPOINT_NAME)),
+                new ConsumerWrapFactory.KafkaConsumerWrapFactory());
+        context.setStreamSource(recordGenerator);
+        return recordGenerator;
+    }
+
+    private static EtlRecordProcessor getEtlRecordProcessor(Context context, Properties properties, RecordGenerator recordGenerator) {
+
+        EtlRecordProcessor etlRecordProcessor = new EtlRecordProcessor(new OffsetCommitCallBack() {
+            @Override
+            public void commit(TopicPartition tp, long timestamp, long offset, String metadata) {
+                recordGenerator.setToCommitCheckpoint(new Checkpoint(tp, timestamp, offset, metadata));
+            }
+        }, context);
+        context.setRecordProcessor(etlRecordProcessor);
+        return etlRecordProcessor;
+    }
+
+
+    // may check some fo config value
+    private static void checkConfig(Properties properties) {
+        require(null != properties.getProperty(USER_NAME), "use should supplied");
+        require(null != properties.getProperty(PASSWORD_NAME), "password should supplied");
+        require(null != properties.getProperty(SID_NAME), "sid should supplied");
+        require(null != properties.getProperty(KAFKA_TOPIC), "kafka topic should supplied");
+        require(null != properties.getProperty(KAFKA_BROKER_URL_NAME), "broker url should supplied");
+    }
+
+    private static Properties initLog4j() {
+        Properties properties = new Properties();
+        InputStream log4jInput = null;
+        try {
+            log4jInput = Thread.currentThread().getContextClassLoader().getResourceAsStream("log4j.properties");
+            PropertyConfigurator.configure(log4jInput);
+        } catch (Exception e) {
+        } finally {
+            swallowErrorClose(log4jInput);
+        }
+        return properties;
+    }
+
+
+    private static Properties loadConfig(String filePath) {
+        Properties ret = new Properties();
+        InputStream toLoad = null;
+        try {
+            toLoad = new BufferedInputStream(new FileInputStream(filePath));
+            ret.load(toLoad);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }  finally {
+            swallowErrorClose(toLoad);
+        }
+        return ret;
+    }
+}

src/main/java/com/alibaba/dts/boot/MysqlRecordPrinter.java

+package com.alibaba.dts.boot;
+
+import com.alibaba.dts.formats.avro.Field;
+import com.alibaba.dts.formats.avro.Record;
+import com.alibaba.dts.common.FieldEntryHolder;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import com.alibaba.dts.recordprocessor.FieldConverter;
+
+import java.util.Iterator;
+import java.util.List;
+
+import static com.alibaba.dts.common.Util.uncompressionObjectName;
+
+/**
+ * this class show how to process record attribute and column
+ */
+public class MysqlRecordPrinter {
+    private static final Logger log = LoggerFactory.getLogger(MysqlRecordPrinter.class);
+    private static final FieldConverter FIELD_CONVERTER = FieldConverter.getConverter("mysql", null);
+
+
+    public static String recordToString(Record record) {
+        StringBuilder stringBuilder = new StringBuilder(256);
+        switch (record.getOperation()) {
+            case DDL: {
+                appendRecordGeneralInfo(record, stringBuilder);
+                String ddl = (String)record.getAfterImages();
+                stringBuilder.append("DDL [").append(ddl).append("]");
+                break;
+            }
+            default: {
+                List<Field> fields = (List<Field>) record.getFields();
+                FieldEntryHolder[] fieldArray = getFieldEntryHolder(record);
+                appendRecordGeneralInfo(record, stringBuilder);
+                appendFields(fields, fieldArray[0], fieldArray[1], stringBuilder);
+                break;
+            }
+        }
+
+        return stringBuilder.toString();
+    }
+
+
+    private static FieldEntryHolder[] getFieldEntryHolder(Record record) {
+        // this is a simple impl, may exist unhandled situation
+        FieldEntryHolder[] fieldArray = new FieldEntryHolder[2];
+
+            fieldArray[0] = new FieldEntryHolder((List<Object>) record.getBeforeImages());
+            fieldArray[1] = new FieldEntryHolder((List<Object>) record.getAfterImages());
+
+        return fieldArray;
+    }
+
+    private static void appendFields(List<Field> fields, FieldEntryHolder before, FieldEntryHolder after, StringBuilder stringBuilder) {
+        if (null != fields) {
+            Iterator<Field> fieldIterator = fields.iterator();
+            while (fieldIterator.hasNext() && before.hasNext() && after.hasNext()) {
+                Field field = fieldIterator.next();
+                Object toPrintBefore = before.take();
+                Object toPrintAfter = after.take();
+                appendField(field, toPrintBefore, toPrintAfter, stringBuilder);
+            }
+        }
+    }
+
+
+
+    private static void appendField(Field field, Object beforeImage, Object afterImage, StringBuilder stringBuilder) {
+        stringBuilder.append("Field [").append(field.getName()).append("]");
+        if (null != beforeImage) {
+            stringBuilder.append("Before [").append(FIELD_CONVERTER.convert(field, beforeImage).toString()).append("]");
+        }
+        if (null != afterImage) {
+            stringBuilder.append("After [").append(FIELD_CONVERTER.convert(field, afterImage).toString()).append("]");
+        }
+        stringBuilder.append("\n");
+    }
+
+    private static void appendRecordGeneralInfo(Record record, StringBuilder stringBuilder) {
+        String dbName = null;
+        String tableName = null;
+        // here we get db and table name
+        String[] dbPair = uncompressionObjectName(record.getObjectName());
+        if (null != dbPair) {
+            if (dbPair.length == 2) {
+                dbName = dbPair[0];
+                tableName = dbPair[1];
+            } else if (dbPair.length == 3) {
+                dbName = dbPair[0];
+                tableName = dbPair[2];
+            } else if (dbPair.length == 1) {
+                dbName = dbPair[0];
+                tableName = "";
+            } else {
+                throw new RuntimeException("invalid db and table name pair for record [" + record + "]");
+            }
+        }
+        stringBuilder.
+                // record id can not be used as unique identifier
+                append("recordID [").append(record.getId()).append("]")
+                // source info contains which source this record came from
+                .append("source [").append(record.getSource()).append("]")
+                // db and table name
+                .append("dbTable [").append(dbName).append(".").append(tableName).append("]")
+                // record type
+                .append("recordType [").append(record.getOperation()).append("]")
+                // record generate timestamp in source log
+                .append("recordTimestamp [").append(record.getSourceTimestamp()).append("]")
+                // record extra tag
+                .append("extra tags [").append(StringUtils.join(record.getTags(), ",")).append("]");
+        stringBuilder.append("\n");
+
+    }
+
+}

src/main/java/com/alibaba/dts/common/AtomicFileStore.java

+package com.alibaba.dts.common;
+
+import java.io.*;
+import java.util.LinkedList;
+import java.util.List;
+
+import static com.alibaba.dts.common.Util.*;
+
+public class AtomicFileStore {
+    private final String fileName;
+
+    public AtomicFileStore(String fileName) {
+        this.fileName = fileName;
+    }
+
+    public List<String> getContent() {
+        List<String> ret = new LinkedList<>();
+        if (!checkFileExists(fileName)) {
+            return ret;
+        }
+        FileReader readFile = null;
+        BufferedReader bufferedReader = null;
+        try {
+            readFile = new FileReader(fileName);
+            bufferedReader = new BufferedReader(readFile);
+            String s = null;
+            while ((s = bufferedReader.readLine()) != null) {
+                ret.add(s);
+            }
+        } catch (Exception e) {
+        } finally {
+            swallowErrorClose(readFile);
+            swallowErrorClose(bufferedReader);
+        }
+        return ret;
+
+    }
+
+    public boolean updateContent(List<String> newContent)  {
+        synchronized (this) {
+            String tmpFileName = fileName + ".tmp";
+            if (checkFileExists(tmpFileName)) {
+                deleteFile(tmpFileName);
+            }
+            boolean writeSuccess = true;
+            FileWriter fileWriter = null;
+            BufferedWriter bufferedWriter = null;
+            try {
+                fileWriter = new FileWriter(tmpFileName);
+                bufferedWriter = new BufferedWriter(fileWriter);
+                for (String content : newContent) {
+                    bufferedWriter.write(content);
+                    bufferedWriter.newLine();
+                }
+                bufferedWriter.flush();
+            } catch (Exception e) {
+                writeSuccess = false;
+            } finally {
+                swallowErrorClose(fileWriter);
+                swallowErrorClose(bufferedWriter);
+            }
+            // BugFix: windows can't rename file to existing file, remove old file first
+            remove();
+            return writeSuccess ? (new File(tmpFileName).renameTo(new File(fileName))) : false;
+        }
+    }
+
+    public void remove()  {
+        deleteFile(fileName);
+    }
+
+}

src/main/java/com/alibaba/dts/common/Checkpoint.java

+package com.alibaba.dts.common;
+
+import org.apache.kafka.common.TopicPartition;
+
+public class Checkpoint {
+
+    public static final Checkpoint INVALID_STREAM_CHECKPOINT = new Checkpoint(null, -1, -1, "-1");
+    private final TopicPartition topicPartition;
+    private final long timeStamp;
+    private final long offset;
+    private final String info;
+
+    public Checkpoint(TopicPartition topicPartition, long timeStamp, long offset, String info) {
+        this.topicPartition = topicPartition;
+        this.timeStamp = timeStamp;
+        this.offset = offset;
+        this.info = info;
+    }
+
+    public long getOffset() {
+        return offset;
+    }
+
+    public long getTimeStamp() {
+        return timeStamp;
+    }
+
+    public String getInfo() {
+        return info;
+    }
+
+    public TopicPartition getTopicPartition() {
+        return topicPartition;
+    }
+
+    public String toString() {
+        return "Checkpoint[ topicPartition: " + topicPartition + "timestamp: " + timeStamp + ", offset: " + offset + ", info: "  + info + "]";
+    }
+}

src/main/java/com/alibaba/dts/common/Context.java

+package com.alibaba.dts.common;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import com.alibaba.dts.recordgenerator.RecordGenerator;
+import com.alibaba.dts.recordprocessor.EtlRecordProcessor;
+
+public class Context {
+    private static final Logger log = LoggerFactory.getLogger(Context.class);
+    private RecordGenerator streamSource;
+    private EtlRecordProcessor recordProcessor;
+    public void setStreamSource(RecordGenerator streamSource) {
+        this.streamSource = streamSource;
+    }
+
+    public EtlRecordProcessor getRecordProcessor() {
+        return recordProcessor;
+    }
+
+    public  void setRecordProcessor(EtlRecordProcessor recordProcessor) {
+        this.recordProcessor = recordProcessor;
+    }
+
+}

src/main/java/com/alibaba/dts/common/FieldEntryHolder.java

+package com.alibaba.dts.common;
+
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+
+public class FieldEntryHolder {
+    private final List<Object> originFields;
+    private final Iterator<Object> iterator;
+    private final List<Object> filteredFields;
+    public FieldEntryHolder(List<Object> originFields) {
+        this.originFields = originFields;
+        if (null == originFields) {
+            this.filteredFields = null;
+            this.iterator = null;
+        } else {
+            this.filteredFields = new LinkedList<>();
+            this.iterator = originFields.iterator();
+        }
+    }
+
+    public boolean hasNext() {
+        if (iterator == null) {
+            return true;
+        }
+        return iterator.hasNext();
+    }
+
+    public void skip() {
+        if (null != iterator) {
+            iterator.next();
+        }
+    }
+
+    public Object take() {
+        if (null != iterator) {
+            Object current = iterator.next();
+            filteredFields.add(current);
+            return  current;
+        } else {
+            return null;
+        }
+    }
+}
\ No newline at end of file

src/main/java/com/alibaba/dts/common/RecordListener.java

+package com.alibaba.dts.common;
+
+public  interface RecordListener {
+
+
+    public void consume(UserRecord record);
+
+}

src/main/java/com/alibaba/dts/common/UniqueKeyGenerator.java

+package com.alibaba.dts.common;
+
+import java.util.concurrent.atomic.AtomicLong;
+/**
+ * for compaction enabled topic, empty key field in producer record is not allowed, so we gene random key to avoid compaction
+ */
+public class UniqueKeyGenerator {
+    private AtomicLong counter;
+    private final String startMSStr;
+    public UniqueKeyGenerator() {
+        counter = new AtomicLong(0);
+        startMSStr = String.valueOf(System.currentTimeMillis()) + "-";
+    }
+    public String nextKey() {
+        return startMSStr + counter.getAndIncrement();
+    }
+
+}
\ No newline at end of file

src/main/java/com/alibaba/dts/common/UserCommitCallBack.java

+package com.alibaba.dts.common;
+
+import com.alibaba.dts.formats.avro.Record;
+import org.apache.kafka.common.TopicPartition;
+
+public interface UserCommitCallBack {
+    public void commit(TopicPartition tp, Record record, long offset, String metadata);
+}

src/main/java/com/alibaba/dts/common/UserRecord.java

+package com.alibaba.dts.common;
+
+import com.alibaba.dts.formats.avro.Record;
+import org.apache.kafka.common.TopicPartition;
+
+public class UserRecord {
+    private final TopicPartition topicPartition;
+    private final long offset;
+    private final Record record;
+    private final UserCommitCallBack userCommitCallBack;
+
+    public UserRecord(TopicPartition tp, long offset, Record record, UserCommitCallBack userCommitCallBack) {
+        this.topicPartition = tp;
+        this.offset = offset;
+        this.record = record;
+        this.userCommitCallBack = userCommitCallBack;
+    }
+
+    public long getOffset() {
+        return offset;
+    }
+
+    public Record getRecord() {
+        return record;
+    }
+
+    public TopicPartition getTopicPartition() {
+        return topicPartition;
+    }
+
+    public void commit(String metadata) {
+        userCommitCallBack.commit(topicPartition, record, offset, metadata);
+    }
+}

src/main/java/com/alibaba/dts/common/Util.java

+package com.alibaba.dts.common;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.config.SaslConfigs;
+import com.alibaba.dts.recordgenerator.ClusterSwitchListener;
+
+import java.io.Closeable;
+import java.io.File;
+import java.util.Properties;
+
+import static com.alibaba.dts.recordgenerator.Names.*;
+
+public class Util {
+    public static  void swallowErrorClose(Closeable target) {
+        try {
+            if (null != target) {
+                target.close();
+            }
+        } catch (Exception e) {
+        }
+    }
+
+    public static void sleepMS(long value) {
+        try {
+            Thread.sleep(value);
+        } catch (Exception e) {
+        }
+    }
+
+    public static void mergeSourceKafkaProperties(Properties originProperties, Properties mergeToProperties) {
+        originProperties.forEach((k, v) ->{
+            String key = (String)k;
+            if (key.startsWith("kafka.")) {
+                String toPutKey = key.substring(6);
+                mergeToProperties.setProperty(toPutKey, (String)v);
+            }
+        });
+        mergeToProperties.setProperty(SaslConfigs.SASL_JAAS_CONFIG,
+                buildJaasConfig(originProperties.getProperty(SID_NAME), originProperties.getProperty(USER_NAME), originProperties.getProperty(PASSWORD_NAME)));
+        mergeToProperties.setProperty(SaslConfigs.SASL_MECHANISM, "PLAIN");
+        mergeToProperties.setProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
+        mergeToProperties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, originProperties.getProperty(KAFKA_BROKER_URL_NAME));
+        mergeToProperties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, originProperties.getProperty(GROUP_NAME));
+        // disable auto commit
+        mergeToProperties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+
+        mergeToProperties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
+        mergeToProperties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
+        // to let the consumer feel the switch of cluster and reseek the offset by timestamp
+        mergeToProperties.setProperty(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, ClusterSwitchListener.class.getName());
+
+    }
+
+    public static void require(boolean predict, String errMessage) {
+        if (!predict) {
+            throw new RuntimeException(errMessage);
+        }
+    }
+
+    public static String buildJaasConfig(String sid, String  user, String password) {
+        String jaasTemplate = "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"%s-%s\" password=\"%s\";";
+        return String.format(jaasTemplate, user, sid,  password);
+    }
+
+    public static boolean checkFileExists(String fileName) {
+        File metaFile = new File(fileName);
+        return metaFile.exists();
+    }
+
+    public static void deleteFile(String fileName) {
+        File maybeAbsolutePath = new File(fileName);
+        if (!maybeAbsolutePath.exists()) {
+            return;
+        }
+        File metaFile = null;
+        if (maybeAbsolutePath.isAbsolute()) {
+            metaFile = maybeAbsolutePath;
+        } else {
+            File currentPath = new File(".");
+            metaFile = new File(currentPath.getAbsolutePath() + File.separator + fileName);
+        }
+        boolean deleted = metaFile.delete();
+        if (!deleted) {
+            throw new RuntimeException(metaFile.getAbsolutePath() + " should be cleaned anyway");
+        }
+    }
+
+    private static void setIfAbsent(Properties properties, String key, String valueSetIfAbsent) {
+        if (StringUtils.isEmpty(properties.getProperty(key))) {
+            properties.setProperty(key, valueSetIfAbsent);
+        }
+    }
+
+    public  static String[] uncompressionObjectName(String compressionName){
+        if(null == compressionName || compressionName.isEmpty() ){
+            return  null;
+        }
+
+        String [] names = compressionName.split("\\.");
+
+        int length = names.length;
+
+        for(int i=0;i<length;++i){
+            names[i] = unescapeName(names[i]);
+        }
+        return names;
+    }
+
+    private  static String unescapeName(String name){
+        if (null == name || (name.indexOf("\\u002E") < 0)) {
+            return name;
+        }
+
+        StringBuilder builder = new StringBuilder();
+
+        int length = name.length();
+
+        for(int i=0;i<length;++i){
+            char c = name.charAt(i);
+            if('\\' == c && ( i<length-6 && 'u' == name.charAt(i + 1)
+                    && '0' == name.charAt(i + 2) && '0' == name.charAt(i+3)
+                    && '2' == name.charAt(i + 4) && 'E' == name.charAt(i+5))){
+                builder.append(".");
+                i += 5;
+                continue;
+            }else{
+                builder.append(c);
+            }
+        }
+
+        return builder.toString();
+    }
+
+
+    public static void sleepMs(long ms) {
+        try {
+            Thread.sleep(ms);
+        } catch (Exception e) {
+        }
+    }
+}

src/main/java/com/alibaba/dts/common/WorkThread.java

+package com.alibaba.dts.common;
+
+import java.io.Closeable;
+
+import static com.alibaba.dts.common.Util.swallowErrorClose;
+
+public class WorkThread<T extends Runnable & Closeable> {
+    private final T r;
+    private final Thread worker;
+
+    public WorkThread(T r) {
+        this.r = r;
+        worker = new Thread(r);
+    }
+
+    public void start() {
+        worker.start();
+    }
+
+    public void stop() {
+        swallowErrorClose(r);
+        try {
+            worker.join(10000, 0);
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+}
\ No newline at end of file

src/main/java/com/alibaba/dts/formats/avro/BinaryObject.java

+/**
+ * Autogenerated by Avro
+ *
+ * DO NOT EDIT DIRECTLY
+ */
+package com.alibaba.dts.formats.avro;
+
+import org.apache.avro.specific.SpecificData;
+import org.apache.avro.message.BinaryMessageEncoder;
+import org.apache.avro.message.BinaryMessageDecoder;
+import org.apache.avro.message.SchemaStore;
+
+@SuppressWarnings("all")
+@org.apache.avro.specific.AvroGenerated
+public class BinaryObject extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
+  private static final long serialVersionUID = 9094139956249411247L;
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"BinaryObject\",\"namespace\":\"com.alibaba.dts.formats.avro\",\"fields\":[{\"name\":\"type\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}},{\"name\":\"value\",\"type\":\"bytes\"}]}");
+  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
+
+  private static SpecificData MODEL$ = new SpecificData();
+
+  private static final BinaryMessageEncoder<BinaryObject> ENCODER =
+      new BinaryMessageEncoder<BinaryObject>(MODEL$, SCHEMA$);
+
+  private static final BinaryMessageDecoder<BinaryObject> DECODER =
+      new BinaryMessageDecoder<BinaryObject>(MODEL$, SCHEMA$);
+
+  /**
+   * Return the BinaryMessageDecoder instance used by this class.
+   */
+  public static BinaryMessageDecoder<BinaryObject> getDecoder() {
+    return DECODER;
+  }
+
+  /**
+   * Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}.
+   * @param resolver a {@link SchemaStore} used to find schemas by fingerprint
+   */
+  public static BinaryMessageDecoder<BinaryObject> createDecoder(SchemaStore resolver) {
+    return new BinaryMessageDecoder<BinaryObject>(MODEL$, SCHEMA$, resolver);
+  }
+
+  /** Serializes this BinaryObject to a ByteBuffer. */
+  public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException {
+    return ENCODER.encode(this);
+  }
+
+  /** Deserializes a BinaryObject from a ByteBuffer. */
+  public static BinaryObject fromByteBuffer(
+      java.nio.ByteBuffer b) throws java.io.IOException {
+    return DECODER.decode(b);
+  }
+
+  @Deprecated public java.lang.String type;
+  @Deprecated public java.nio.ByteBuffer value;
+
+  /**
+   * Default constructor.  Note that this does not initialize fields
+   * to their default values from the schema.  If that is desired then
+   * one should use <code>newBuilder()</code>.
+   */
+  public BinaryObject() {}
+
+  /**
+   * All-args constructor.
+   * @param type The new value for type
+   * @param value The new value for value
+   */
+  public BinaryObject(java.lang.String type, java.nio.ByteBuffer value) {
+    this.type = type;
+    this.value = value;
+  }
+
+  public org.apache.avro.Schema getSchema() { return SCHEMA$; }
+  // Used by DatumWriter.  Applications should not call.
+  public java.lang.Object get(int field$) {
+    switch (field$) {
+    case 0: return type;
+    case 1: return value;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  // Used by DatumReader.  Applications should not call.
+  @SuppressWarnings(value="unchecked")
+  public void put(int field$, java.lang.Object value$) {
+    switch (field$) {
+    case 0: type = (java.lang.String)value$; break;
+    case 1: value = (java.nio.ByteBuffer)value$; break;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  /**
+   * Gets the value of the 'type' field.
+   * @return The value of the 'type' field.
+   */
+  public java.lang.String getType() {
+    return type;
+  }
+
+  /**
+   * Sets the value of the 'type' field.
+   * @param value the value to set.
+   */
+  public void setType(java.lang.String value) {
+    this.type = value;
+  }
+
+  /**
+   * Gets the value of the 'value' field.
+   * @return The value of the 'value' field.
+   */
+  public java.nio.ByteBuffer getValue() {
+    return value;
+  }
+
+  /**
+   * Sets the value of the 'value' field.
+   * @param value the value to set.
+   */
+  public void setValue(java.nio.ByteBuffer value) {
+    this.value = value;
+  }
+
+  /**
+   * Creates a new BinaryObject RecordBuilder.
+   * @return A new BinaryObject RecordBuilder
+   */
+  public static com.alibaba.dts.formats.avro.BinaryObject.Builder newBuilder() {
+    return new com.alibaba.dts.formats.avro.BinaryObject.Builder();
+  }
+
+  /**
+   * Creates a new BinaryObject RecordBuilder by copying an existing Builder.
+   * @param other The existing builder to copy.
+   * @return A new BinaryObject RecordBuilder
+   */
+  public static com.alibaba.dts.formats.avro.BinaryObject.Builder newBuilder(com.alibaba.dts.formats.avro.BinaryObject.Builder other) {
+    return new com.alibaba.dts.formats.avro.BinaryObject.Builder(other);
+  }
+
+  /**
+   * Creates a new BinaryObject RecordBuilder by copying an existing BinaryObject instance.
+   * @param other The existing instance to copy.
+   * @return A new BinaryObject RecordBuilder
+   */
+  public static com.alibaba.dts.formats.avro.BinaryObject.Builder newBuilder(com.alibaba.dts.formats.avro.BinaryObject other) {
+    return new com.alibaba.dts.formats.avro.BinaryObject.Builder(other);
+  }
+
+  /**
+   * RecordBuilder for BinaryObject instances.
+   */
+  public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<BinaryObject>
+    implements org.apache.avro.data.RecordBuilder<BinaryObject> {
+
+    private java.lang.String type;
+    private java.nio.ByteBuffer value;
+
+    /** Creates a new Builder */
+    private Builder() {
+      super(SCHEMA$);
+    }
+
+    /**
+     * Creates a Builder by copying an existing Builder.
+     * @param other The existing Builder to copy.
+     */
+    private Builder(com.alibaba.dts.formats.avro.BinaryObject.Builder other) {
+      super(other);
+      if (isValidValue(fields()[0], other.type)) {
+        this.type = data().deepCopy(fields()[0].schema(), other.type);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.value)) {
+        this.value = data().deepCopy(fields()[1].schema(), other.value);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+     * Creates a Builder by copying an existing BinaryObject instance
+     * @param other The existing instance to copy.
+     */
+    private Builder(com.alibaba.dts.formats.avro.BinaryObject other) {
+            super(SCHEMA$);
+      if (isValidValue(fields()[0], other.type)) {
+        this.type = data().deepCopy(fields()[0].schema(), other.type);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.value)) {
+        this.value = data().deepCopy(fields()[1].schema(), other.value);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+      * Gets the value of the 'type' field.
+      * @return The value.
+      */
+    public java.lang.String getType() {
+      return type;
+    }
+
+    /**
+      * Sets the value of the 'type' field.
+      * @param value The value of 'type'.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.BinaryObject.Builder setType(java.lang.String value) {
+      validate(fields()[0], value);
+      this.type = value;
+      fieldSetFlags()[0] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'type' field has been set.
+      * @return True if the 'type' field has been set, false otherwise.
+      */
+    public boolean hasType() {
+      return fieldSetFlags()[0];
+    }
+
+
+    /**
+      * Clears the value of the 'type' field.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.BinaryObject.Builder clearType() {
+      type = null;
+      fieldSetFlags()[0] = false;
+      return this;
+    }
+
+    /**
+      * Gets the value of the 'value' field.
+      * @return The value.
+      */
+    public java.nio.ByteBuffer getValue() {
+      return value;
+    }
+
+    /**
+      * Sets the value of the 'value' field.
+      * @param value The value of 'value'.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.BinaryObject.Builder setValue(java.nio.ByteBuffer value) {
+      validate(fields()[1], value);
+      this.value = value;
+      fieldSetFlags()[1] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'value' field has been set.
+      * @return True if the 'value' field has been set, false otherwise.
+      */
+    public boolean hasValue() {
+      return fieldSetFlags()[1];
+    }
+
+
+    /**
+      * Clears the value of the 'value' field.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.BinaryObject.Builder clearValue() {
+      value = null;
+      fieldSetFlags()[1] = false;
+      return this;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public BinaryObject build() {
+      try {
+        BinaryObject record = new BinaryObject();
+        record.type = fieldSetFlags()[0] ? this.type : (java.lang.String) defaultValue(fields()[0]);
+        record.value = fieldSetFlags()[1] ? this.value : (java.nio.ByteBuffer) defaultValue(fields()[1]);
+        return record;
+      } catch (java.lang.Exception e) {
+        throw new org.apache.avro.AvroRuntimeException(e);
+      }
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumWriter<BinaryObject>
+    WRITER$ = (org.apache.avro.io.DatumWriter<BinaryObject>)MODEL$.createDatumWriter(SCHEMA$);
+
+  @Override public void writeExternal(java.io.ObjectOutput out)
+    throws java.io.IOException {
+    WRITER$.write(this, SpecificData.getEncoder(out));
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumReader<BinaryObject>
+    READER$ = (org.apache.avro.io.DatumReader<BinaryObject>)MODEL$.createDatumReader(SCHEMA$);
+
+  @Override public void readExternal(java.io.ObjectInput in)
+    throws java.io.IOException {
+    READER$.read(this, SpecificData.getDecoder(in));
+  }
+
+}

src/main/java/com/alibaba/dts/formats/avro/Character.java

+/**
+ * Autogenerated by Avro
+ *
+ * DO NOT EDIT DIRECTLY
+ */
+package com.alibaba.dts.formats.avro;
+
+import org.apache.avro.specific.SpecificData;
+import org.apache.avro.message.BinaryMessageEncoder;
+import org.apache.avro.message.BinaryMessageDecoder;
+import org.apache.avro.message.SchemaStore;
+
+@SuppressWarnings("all")
+@org.apache.avro.specific.AvroGenerated
+public class Character extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
+  private static final long serialVersionUID = -1775377444003683903L;
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Character\",\"namespace\":\"com.alibaba.dts.formats.avro\",\"fields\":[{\"name\":\"charset\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}},{\"name\":\"value\",\"type\":\"bytes\"}]}");
+  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
+
+  private static SpecificData MODEL$ = new SpecificData();
+
+  private static final BinaryMessageEncoder<Character> ENCODER =
+      new BinaryMessageEncoder<Character>(MODEL$, SCHEMA$);
+
+  private static final BinaryMessageDecoder<Character> DECODER =
+      new BinaryMessageDecoder<Character>(MODEL$, SCHEMA$);
+
+  /**
+   * Return the BinaryMessageDecoder instance used by this class.
+   */
+  public static BinaryMessageDecoder<Character> getDecoder() {
+    return DECODER;
+  }
+
+  /**
+   * Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}.
+   * @param resolver a {@link SchemaStore} used to find schemas by fingerprint
+   */
+  public static BinaryMessageDecoder<Character> createDecoder(SchemaStore resolver) {
+    return new BinaryMessageDecoder<Character>(MODEL$, SCHEMA$, resolver);
+  }
+
+  /** Serializes this Character to a ByteBuffer. */
+  public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException {
+    return ENCODER.encode(this);
+  }
+
+  /** Deserializes a Character from a ByteBuffer. */
+  public static Character fromByteBuffer(
+      java.nio.ByteBuffer b) throws java.io.IOException {
+    return DECODER.decode(b);
+  }
+
+  @Deprecated public java.lang.String charset;
+  @Deprecated public java.nio.ByteBuffer value;
+
+  /**
+   * Default constructor.  Note that this does not initialize fields
+   * to their default values from the schema.  If that is desired then
+   * one should use <code>newBuilder()</code>.
+   */
+  public Character() {}
+
+  /**
+   * All-args constructor.
+   * @param charset The new value for charset
+   * @param value The new value for value
+   */
+  public Character(java.lang.String charset, java.nio.ByteBuffer value) {
+    this.charset = charset;
+    this.value = value;
+  }
+
+  public org.apache.avro.Schema getSchema() { return SCHEMA$; }
+  // Used by DatumWriter.  Applications should not call.
+  public java.lang.Object get(int field$) {
+    switch (field$) {
+    case 0: return charset;
+    case 1: return value;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  // Used by DatumReader.  Applications should not call.
+  @SuppressWarnings(value="unchecked")
+  public void put(int field$, java.lang.Object value$) {
+    switch (field$) {
+    case 0: charset = (java.lang.String)value$; break;
+    case 1: value = (java.nio.ByteBuffer)value$; break;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  /**
+   * Gets the value of the 'charset' field.
+   * @return The value of the 'charset' field.
+   */
+  public java.lang.String getCharset() {
+    return charset;
+  }
+
+  /**
+   * Sets the value of the 'charset' field.
+   * @param value the value to set.
+   */
+  public void setCharset(java.lang.String value) {
+    this.charset = value;
+  }
+
+  /**
+   * Gets the value of the 'value' field.
+   * @return The value of the 'value' field.
+   */
+  public java.nio.ByteBuffer getValue() {
+    return value;
+  }
+
+  /**
+   * Sets the value of the 'value' field.
+   * @param value the value to set.
+   */
+  public void setValue(java.nio.ByteBuffer value) {
+    this.value = value;
+  }
+
+  /**
+   * Creates a new Character RecordBuilder.
+   * @return A new Character RecordBuilder
+   */
+  public static com.alibaba.dts.formats.avro.Character.Builder newBuilder() {
+    return new com.alibaba.dts.formats.avro.Character.Builder();
+  }
+
+  /**
+   * Creates a new Character RecordBuilder by copying an existing Builder.
+   * @param other The existing builder to copy.
+   * @return A new Character RecordBuilder
+   */
+  public static com.alibaba.dts.formats.avro.Character.Builder newBuilder(com.alibaba.dts.formats.avro.Character.Builder other) {
+    return new com.alibaba.dts.formats.avro.Character.Builder(other);
+  }
+
+  /**
+   * Creates a new Character RecordBuilder by copying an existing Character instance.
+   * @param other The existing instance to copy.
+   * @return A new Character RecordBuilder
+   */
+  public static com.alibaba.dts.formats.avro.Character.Builder newBuilder(com.alibaba.dts.formats.avro.Character other) {
+    return new com.alibaba.dts.formats.avro.Character.Builder(other);
+  }
+
+  /**
+   * RecordBuilder for Character instances.
+   */
+  public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<Character>
+    implements org.apache.avro.data.RecordBuilder<Character> {
+
+    private java.lang.String charset;
+    private java.nio.ByteBuffer value;
+
+    /** Creates a new Builder */
+    private Builder() {
+      super(SCHEMA$);
+    }
+
+    /**
+     * Creates a Builder by copying an existing Builder.
+     * @param other The existing Builder to copy.
+     */
+    private Builder(com.alibaba.dts.formats.avro.Character.Builder other) {
+      super(other);
+      if (isValidValue(fields()[0], other.charset)) {
+        this.charset = data().deepCopy(fields()[0].schema(), other.charset);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.value)) {
+        this.value = data().deepCopy(fields()[1].schema(), other.value);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+     * Creates a Builder by copying an existing Character instance
+     * @param other The existing instance to copy.
+     */
+    private Builder(com.alibaba.dts.formats.avro.Character other) {
+            super(SCHEMA$);
+      if (isValidValue(fields()[0], other.charset)) {
+        this.charset = data().deepCopy(fields()[0].schema(), other.charset);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.value)) {
+        this.value = data().deepCopy(fields()[1].schema(), other.value);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+      * Gets the value of the 'charset' field.
+      * @return The value.
+      */
+    public java.lang.String getCharset() {
+      return charset;
+    }
+
+    /**
+      * Sets the value of the 'charset' field.
+      * @param value The value of 'charset'.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.Character.Builder setCharset(java.lang.String value) {
+      validate(fields()[0], value);
+      this.charset = value;
+      fieldSetFlags()[0] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'charset' field has been set.
+      * @return True if the 'charset' field has been set, false otherwise.
+      */
+    public boolean hasCharset() {
+      return fieldSetFlags()[0];
+    }
+
+
+    /**
+      * Clears the value of the 'charset' field.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.Character.Builder clearCharset() {
+      charset = null;
+      fieldSetFlags()[0] = false;
+      return this;
+    }
+
+    /**
+      * Gets the value of the 'value' field.
+      * @return The value.
+      */
+    public java.nio.ByteBuffer getValue() {
+      return value;
+    }
+
+    /**
+      * Sets the value of the 'value' field.
+      * @param value The value of 'value'.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.Character.Builder setValue(java.nio.ByteBuffer value) {
+      validate(fields()[1], value);
+      this.value = value;
+      fieldSetFlags()[1] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'value' field has been set.
+      * @return True if the 'value' field has been set, false otherwise.
+      */
+    public boolean hasValue() {
+      return fieldSetFlags()[1];
+    }
+
+
+    /**
+      * Clears the value of the 'value' field.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.Character.Builder clearValue() {
+      value = null;
+      fieldSetFlags()[1] = false;
+      return this;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public Character build() {
+      try {
+        Character record = new Character();
+        record.charset = fieldSetFlags()[0] ? this.charset : (java.lang.String) defaultValue(fields()[0]);
+        record.value = fieldSetFlags()[1] ? this.value : (java.nio.ByteBuffer) defaultValue(fields()[1]);
+        return record;
+      } catch (java.lang.Exception e) {
+        throw new org.apache.avro.AvroRuntimeException(e);
+      }
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumWriter<Character>
+    WRITER$ = (org.apache.avro.io.DatumWriter<Character>)MODEL$.createDatumWriter(SCHEMA$);
+
+  @Override public void writeExternal(java.io.ObjectOutput out)
+    throws java.io.IOException {
+    WRITER$.write(this, SpecificData.getEncoder(out));
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumReader<Character>
+    READER$ = (org.apache.avro.io.DatumReader<Character>)MODEL$.createDatumReader(SCHEMA$);
+
+  @Override public void readExternal(java.io.ObjectInput in)
+    throws java.io.IOException {
+    READER$.read(this, SpecificData.getDecoder(in));
+  }
+
+}

src/main/java/com/alibaba/dts/formats/avro/EmptyObject.java

+/**
+ * Autogenerated by Avro
+ *
+ * DO NOT EDIT DIRECTLY
+ */
+package com.alibaba.dts.formats.avro;
+@SuppressWarnings("all")
+@org.apache.avro.specific.AvroGenerated
+public enum EmptyObject {
+  NULL, NONE  ;
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"enum\",\"name\":\"EmptyObject\",\"namespace\":\"com.alibaba.dts.formats.avro\",\"symbols\":[\"NULL\",\"NONE\"]}");
+  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
+}

src/main/java/com/alibaba/dts/formats/avro/Field.java

+/**
+ * Autogenerated by Avro
+ *
+ * DO NOT EDIT DIRECTLY
+ */
+package com.alibaba.dts.formats.avro;
+
+import org.apache.avro.specific.SpecificData;
+import org.apache.avro.message.BinaryMessageEncoder;
+import org.apache.avro.message.BinaryMessageDecoder;
+import org.apache.avro.message.SchemaStore;
+
+@SuppressWarnings("all")
+@org.apache.avro.specific.AvroGenerated
+public class Field extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
+  private static final long serialVersionUID = 49929941316960932L;
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Field\",\"namespace\":\"com.alibaba.dts.formats.avro\",\"fields\":[{\"name\":\"name\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}},{\"name\":\"dataTypeNumber\",\"type\":\"int\"}]}");
+  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
+
+  private static SpecificData MODEL$ = new SpecificData();
+
+  private static final BinaryMessageEncoder<Field> ENCODER =
+      new BinaryMessageEncoder<Field>(MODEL$, SCHEMA$);
+
+  private static final BinaryMessageDecoder<Field> DECODER =
+      new BinaryMessageDecoder<Field>(MODEL$, SCHEMA$);
+
+  /**
+   * Return the BinaryMessageDecoder instance used by this class.
+   */
+  public static BinaryMessageDecoder<Field> getDecoder() {
+    return DECODER;
+  }
+
+  /**
+   * Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}.
+   * @param resolver a {@link SchemaStore} used to find schemas by fingerprint
+   */
+  public static BinaryMessageDecoder<Field> createDecoder(SchemaStore resolver) {
+    return new BinaryMessageDecoder<Field>(MODEL$, SCHEMA$, resolver);
+  }
+
+  /** Serializes this Field to a ByteBuffer. */
+  public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException {
+    return ENCODER.encode(this);
+  }
+
+  /** Deserializes a Field from a ByteBuffer. */
+  public static Field fromByteBuffer(
+      java.nio.ByteBuffer b) throws java.io.IOException {
+    return DECODER.decode(b);
+  }
+
+  @Deprecated public java.lang.String name;
+  @Deprecated public int dataTypeNumber;
+
+  /**
+   * Default constructor.  Note that this does not initialize fields
+   * to their default values from the schema.  If that is desired then
+   * one should use <code>newBuilder()</code>.
+   */
+  public Field() {}
+
+  /**
+   * All-args constructor.
+   * @param name The new value for name
+   * @param dataTypeNumber The new value for dataTypeNumber
+   */
+  public Field(java.lang.String name, java.lang.Integer dataTypeNumber) {
+    this.name = name;
+    this.dataTypeNumber = dataTypeNumber;
+  }
+
+  public org.apache.avro.Schema getSchema() { return SCHEMA$; }
+  // Used by DatumWriter.  Applications should not call.
+  public java.lang.Object get(int field$) {
+    switch (field$) {
+    case 0: return name;
+    case 1: return dataTypeNumber;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  // Used by DatumReader.  Applications should not call.
+  @SuppressWarnings(value="unchecked")
+  public void put(int field$, java.lang.Object value$) {
+    switch (field$) {
+    case 0: name = (java.lang.String)value$; break;
+    case 1: dataTypeNumber = (java.lang.Integer)value$; break;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  /**
+   * Gets the value of the 'name' field.
+   * @return The value of the 'name' field.
+   */
+  public java.lang.String getName() {
+    return name;
+  }
+
+  /**
+   * Sets the value of the 'name' field.
+   * @param value the value to set.
+   */
+  public void setName(java.lang.String value) {
+    this.name = value;
+  }
+
+  /**
+   * Gets the value of the 'dataTypeNumber' field.
+   * @return The value of the 'dataTypeNumber' field.
+   */
+  public java.lang.Integer getDataTypeNumber() {
+    return dataTypeNumber;
+  }
+
+  /**
+   * Sets the value of the 'dataTypeNumber' field.
+   * @param value the value to set.
+   */
+  public void setDataTypeNumber(java.lang.Integer value) {
+    this.dataTypeNumber = value;
+  }
+
+  /**
+   * Creates a new Field RecordBuilder.
+   * @return A new Field RecordBuilder
+   */
+  public static com.alibaba.dts.formats.avro.Field.Builder newBuilder() {
+    return new com.alibaba.dts.formats.avro.Field.Builder();
+  }
+
+  /**
+   * Creates a new Field RecordBuilder by copying an existing Builder.
+   * @param other The existing builder to copy.
+   * @return A new Field RecordBuilder
+   */
+  public static com.alibaba.dts.formats.avro.Field.Builder newBuilder(com.alibaba.dts.formats.avro.Field.Builder other) {
+    return new com.alibaba.dts.formats.avro.Field.Builder(other);
+  }
+
+  /**
+   * Creates a new Field RecordBuilder by copying an existing Field instance.
+   * @param other The existing instance to copy.
+   * @return A new Field RecordBuilder
+   */
+  public static com.alibaba.dts.formats.avro.Field.Builder newBuilder(com.alibaba.dts.formats.avro.Field other) {
+    return new com.alibaba.dts.formats.avro.Field.Builder(other);
+  }
+
+  /**
+   * RecordBuilder for Field instances.
+   */
+  public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<Field>
+    implements org.apache.avro.data.RecordBuilder<Field> {
+
+    private java.lang.String name;
+    private int dataTypeNumber;
+
+    /** Creates a new Builder */
+    private Builder() {
+      super(SCHEMA$);
+    }
+
+    /**
+     * Creates a Builder by copying an existing Builder.
+     * @param other The existing Builder to copy.
+     */
+    private Builder(com.alibaba.dts.formats.avro.Field.Builder other) {
+      super(other);
+      if (isValidValue(fields()[0], other.name)) {
+        this.name = data().deepCopy(fields()[0].schema(), other.name);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.dataTypeNumber)) {
+        this.dataTypeNumber = data().deepCopy(fields()[1].schema(), other.dataTypeNumber);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+     * Creates a Builder by copying an existing Field instance
+     * @param other The existing instance to copy.
+     */
+    private Builder(com.alibaba.dts.formats.avro.Field other) {
+            super(SCHEMA$);
+      if (isValidValue(fields()[0], other.name)) {
+        this.name = data().deepCopy(fields()[0].schema(), other.name);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.dataTypeNumber)) {
+        this.dataTypeNumber = data().deepCopy(fields()[1].schema(), other.dataTypeNumber);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+      * Gets the value of the 'name' field.
+      * @return The value.
+      */
+    public java.lang.String getName() {
+      return name;
+    }
+
+    /**
+      * Sets the value of the 'name' field.
+      * @param value The value of 'name'.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.Field.Builder setName(java.lang.String value) {
+      validate(fields()[0], value);
+      this.name = value;
+      fieldSetFlags()[0] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'name' field has been set.
+      * @return True if the 'name' field has been set, false otherwise.
+      */
+    public boolean hasName() {
+      return fieldSetFlags()[0];
+    }
+
+
+    /**
+      * Clears the value of the 'name' field.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.Field.Builder clearName() {
+      name = null;
+      fieldSetFlags()[0] = false;
+      return this;
+    }
+
+    /**
+      * Gets the value of the 'dataTypeNumber' field.
+      * @return The value.
+      */
+    public java.lang.Integer getDataTypeNumber() {
+      return dataTypeNumber;
+    }
+
+    /**
+      * Sets the value of the 'dataTypeNumber' field.
+      * @param value The value of 'dataTypeNumber'.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.Field.Builder setDataTypeNumber(int value) {
+      validate(fields()[1], value);
+      this.dataTypeNumber = value;
+      fieldSetFlags()[1] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'dataTypeNumber' field has been set.
+      * @return True if the 'dataTypeNumber' field has been set, false otherwise.
+      */
+    public boolean hasDataTypeNumber() {
+      return fieldSetFlags()[1];
+    }
+
+
+    /**
+      * Clears the value of the 'dataTypeNumber' field.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.Field.Builder clearDataTypeNumber() {
+      fieldSetFlags()[1] = false;
+      return this;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public Field build() {
+      try {
+        Field record = new Field();
+        record.name = fieldSetFlags()[0] ? this.name : (java.lang.String) defaultValue(fields()[0]);
+        record.dataTypeNumber = fieldSetFlags()[1] ? this.dataTypeNumber : (java.lang.Integer) defaultValue(fields()[1]);
+        return record;
+      } catch (java.lang.Exception e) {
+        throw new org.apache.avro.AvroRuntimeException(e);
+      }
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumWriter<Field>
+    WRITER$ = (org.apache.avro.io.DatumWriter<Field>)MODEL$.createDatumWriter(SCHEMA$);
+
+  @Override public void writeExternal(java.io.ObjectOutput out)
+    throws java.io.IOException {
+    WRITER$.write(this, SpecificData.getEncoder(out));
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumReader<Field>
+    READER$ = (org.apache.avro.io.DatumReader<Field>)MODEL$.createDatumReader(SCHEMA$);
+
+  @Override public void readExternal(java.io.ObjectInput in)
+    throws java.io.IOException {
+    READER$.read(this, SpecificData.getDecoder(in));
+  }
+
+}

src/main/java/com/alibaba/dts/formats/avro/Integer.java

+/**
+ * Autogenerated by Avro
+ *
+ * DO NOT EDIT DIRECTLY
+ */
+package com.alibaba.dts.formats.avro;
+
+import org.apache.avro.specific.SpecificData;
+import org.apache.avro.message.BinaryMessageEncoder;
+import org.apache.avro.message.BinaryMessageDecoder;
+import org.apache.avro.message.SchemaStore;
+
+@SuppressWarnings("all")
+@org.apache.avro.specific.AvroGenerated
+public class Integer extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
+  private static final long serialVersionUID = 8676388679064639164L;
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Integer\",\"namespace\":\"com.alibaba.dts.formats.avro\",\"fields\":[{\"name\":\"precision\",\"type\":\"int\"},{\"name\":\"value\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}}]}");
+  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
+
+  private static SpecificData MODEL$ = new SpecificData();
+
+  private static final BinaryMessageEncoder<Integer> ENCODER =
+      new BinaryMessageEncoder<Integer>(MODEL$, SCHEMA$);
+
+  private static final BinaryMessageDecoder<Integer> DECODER =
+      new BinaryMessageDecoder<Integer>(MODEL$, SCHEMA$);
+
+  /**
+   * Return the BinaryMessageDecoder instance used by this class.
+   */
+  public static BinaryMessageDecoder<Integer> getDecoder() {
+    return DECODER;
+  }
+
+  /**
+   * Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}.
+   * @param resolver a {@link SchemaStore} used to find schemas by fingerprint
+   */
+  public static BinaryMessageDecoder<Integer> createDecoder(SchemaStore resolver) {
+    return new BinaryMessageDecoder<Integer>(MODEL$, SCHEMA$, resolver);
+  }
+
+  /** Serializes this Integer to a ByteBuffer. */
+  public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException {
+    return ENCODER.encode(this);
+  }
+
+  /** Deserializes a Integer from a ByteBuffer. */
+  public static Integer fromByteBuffer(
+      java.nio.ByteBuffer b) throws java.io.IOException {
+    return DECODER.decode(b);
+  }
+
+  @Deprecated public int precision;
+  @Deprecated public java.lang.String value;
+
+  /**
+   * Default constructor.  Note that this does not initialize fields
+   * to their default values from the schema.  If that is desired then
+   * one should use <code>newBuilder()</code>.
+   */
+  public Integer() {}
+
+  /**
+   * All-args constructor.
+   * @param precision The new value for precision
+   * @param value The new value for value
+   */
+  public Integer(java.lang.Integer precision, java.lang.String value) {
+    this.precision = precision;
+    this.value = value;
+  }
+
+  public org.apache.avro.Schema getSchema() { return SCHEMA$; }
+  // Used by DatumWriter.  Applications should not call.
+  public java.lang.Object get(int field$) {
+    switch (field$) {
+    case 0: return precision;
+    case 1: return value;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  // Used by DatumReader.  Applications should not call.
+  @SuppressWarnings(value="unchecked")
+  public void put(int field$, java.lang.Object value$) {
+    switch (field$) {
+    case 0: precision = (java.lang.Integer)value$; break;
+    case 1: value = (java.lang.String)value$; break;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  /**
+   * Gets the value of the 'precision' field.
+   * @return The value of the 'precision' field.
+   */
+  public java.lang.Integer getPrecision() {
+    return precision;
+  }
+
+  /**
+   * Sets the value of the 'precision' field.
+   * @param value the value to set.
+   */
+  public void setPrecision(java.lang.Integer value) {
+    this.precision = value;
+  }
+
+  /**
+   * Gets the value of the 'value' field.
+   * @return The value of the 'value' field.
+   */
+  public java.lang.String getValue() {
+    return value;
+  }
+
+  /**
+   * Sets the value of the 'value' field.
+   * @param value the value to set.
+   */
+  public void setValue(java.lang.String value) {
+    this.value = value;
+  }
+
+  /**
+   * Creates a new Integer RecordBuilder.
+   * @return A new Integer RecordBuilder
+   */
+  public static com.alibaba.dts.formats.avro.Integer.Builder newBuilder() {
+    return new com.alibaba.dts.formats.avro.Integer.Builder();
+  }
+
+  /**
+   * Creates a new Integer RecordBuilder by copying an existing Builder.
+   * @param other The existing builder to copy.
+   * @return A new Integer RecordBuilder
+   */
+  public static com.alibaba.dts.formats.avro.Integer.Builder newBuilder(com.alibaba.dts.formats.avro.Integer.Builder other) {
+    return new com.alibaba.dts.formats.avro.Integer.Builder(other);
+  }
+
+  /**
+   * Creates a new Integer RecordBuilder by copying an existing Integer instance.
+   * @param other The existing instance to copy.
+   * @return A new Integer RecordBuilder
+   */
+  public static com.alibaba.dts.formats.avro.Integer.Builder newBuilder(com.alibaba.dts.formats.avro.Integer other) {
+    return new com.alibaba.dts.formats.avro.Integer.Builder(other);
+  }
+
+  /**
+   * RecordBuilder for Integer instances.
+   */
+  public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<Integer>
+    implements org.apache.avro.data.RecordBuilder<Integer> {
+
+    private int precision;
+    private java.lang.String value;
+
+    /** Creates a new Builder */
+    private Builder() {
+      super(SCHEMA$);
+    }
+
+    /**
+     * Creates a Builder by copying an existing Builder.
+     * @param other The existing Builder to copy.
+     */
+    private Builder(com.alibaba.dts.formats.avro.Integer.Builder other) {
+      super(other);
+      if (isValidValue(fields()[0], other.precision)) {
+        this.precision = data().deepCopy(fields()[0].schema(), other.precision);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.value)) {
+        this.value = data().deepCopy(fields()[1].schema(), other.value);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+     * Creates a Builder by copying an existing Integer instance
+     * @param other The existing instance to copy.
+     */
+    private Builder(com.alibaba.dts.formats.avro.Integer other) {
+            super(SCHEMA$);
+      if (isValidValue(fields()[0], other.precision)) {
+        this.precision = data().deepCopy(fields()[0].schema(), other.precision);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.value)) {
+        this.value = data().deepCopy(fields()[1].schema(), other.value);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+      * Gets the value of the 'precision' field.
+      * @return The value.
+      */
+    public java.lang.Integer getPrecision() {
+      return precision;
+    }
+
+    /**
+      * Sets the value of the 'precision' field.
+      * @param value The value of 'precision'.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.Integer.Builder setPrecision(int value) {
+      validate(fields()[0], value);
+      this.precision = value;
+      fieldSetFlags()[0] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'precision' field has been set.
+      * @return True if the 'precision' field has been set, false otherwise.
+      */
+    public boolean hasPrecision() {
+      return fieldSetFlags()[0];
+    }
+
+
+    /**
+      * Clears the value of the 'precision' field.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.Integer.Builder clearPrecision() {
+      fieldSetFlags()[0] = false;
+      return this;
+    }
+
+    /**
+      * Gets the value of the 'value' field.
+      * @return The value.
+      */
+    public java.lang.String getValue() {
+      return value;
+    }
+
+    /**
+      * Sets the value of the 'value' field.
+      * @param value The value of 'value'.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.Integer.Builder setValue(java.lang.String value) {
+      validate(fields()[1], value);
+      this.value = value;
+      fieldSetFlags()[1] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'value' field has been set.
+      * @return True if the 'value' field has been set, false otherwise.
+      */
+    public boolean hasValue() {
+      return fieldSetFlags()[1];
+    }
+
+
+    /**
+      * Clears the value of the 'value' field.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.Integer.Builder clearValue() {
+      value = null;
+      fieldSetFlags()[1] = false;
+      return this;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public Integer build() {
+      try {
+        Integer record = new Integer();
+        record.precision = fieldSetFlags()[0] ? this.precision : (java.lang.Integer) defaultValue(fields()[0]);
+        record.value = fieldSetFlags()[1] ? this.value : (java.lang.String) defaultValue(fields()[1]);
+        return record;
+      } catch (java.lang.Exception e) {
+        throw new org.apache.avro.AvroRuntimeException(e);
+      }
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumWriter<Integer>
+    WRITER$ = (org.apache.avro.io.DatumWriter<Integer>)MODEL$.createDatumWriter(SCHEMA$);
+
+  @Override public void writeExternal(java.io.ObjectOutput out)
+    throws java.io.IOException {
+    WRITER$.write(this, SpecificData.getEncoder(out));
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumReader<Integer>
+    READER$ = (org.apache.avro.io.DatumReader<Integer>)MODEL$.createDatumReader(SCHEMA$);
+
+  @Override public void readExternal(java.io.ObjectInput in)
+    throws java.io.IOException {
+    READER$.read(this, SpecificData.getDecoder(in));
+  }
+
+}

src/main/java/com/alibaba/dts/formats/avro/Operation.java

+/**
+ * Autogenerated by Avro
+ *
+ * DO NOT EDIT DIRECTLY
+ */
+package com.alibaba.dts.formats.avro;
+@SuppressWarnings("all")
+@org.apache.avro.specific.AvroGenerated
+public enum Operation {
+  INSERT, UPDATE, DELETE, DDL, BEGIN, COMMIT, ROLLBACK, ABORT, HEARTBEAT, CHECKPOINT, COMMAND, FILL, FINISH, CONTROL, RDB, NOOP, INIT  ;
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"enum\",\"name\":\"Operation\",\"namespace\":\"com.alibaba.dts.formats.avro\",\"symbols\":[\"INSERT\",\"UPDATE\",\"DELETE\",\"DDL\",\"BEGIN\",\"COMMIT\",\"ROLLBACK\",\"ABORT\",\"HEARTBEAT\",\"CHECKPOINT\",\"COMMAND\",\"FILL\",\"FINISH\",\"CONTROL\",\"RDB\",\"NOOP\",\"INIT\"]}");
+  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
+}

src/main/java/com/alibaba/dts/formats/avro/SourceType.java

+/**
+ * Autogenerated by Avro
+ *
+ * DO NOT EDIT DIRECTLY
+ */
+package com.alibaba.dts.formats.avro;
+@SuppressWarnings("all")
+@org.apache.avro.specific.AvroGenerated
+public enum SourceType {
+  MySQL, Oracle, SQLServer, PostgreSQL, MongoDB, Redis, DB2, PPAS, DRDS, HBASE, HDFS, FILE, OTHER  ;
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"enum\",\"name\":\"SourceType\",\"namespace\":\"com.alibaba.dts.formats.avro\",\"symbols\":[\"MySQL\",\"Oracle\",\"SQLServer\",\"PostgreSQL\",\"MongoDB\",\"Redis\",\"DB2\",\"PPAS\",\"DRDS\",\"HBASE\",\"HDFS\",\"FILE\",\"OTHER\"]}");
+  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
+}

src/main/java/com/alibaba/dts/formats/avro/TextGeometry.java

+/**
+ * Autogenerated by Avro
+ *
+ * DO NOT EDIT DIRECTLY
+ */
+package com.alibaba.dts.formats.avro;
+
+import org.apache.avro.specific.SpecificData;
+import org.apache.avro.message.BinaryMessageEncoder;
+import org.apache.avro.message.BinaryMessageDecoder;
+import org.apache.avro.message.SchemaStore;
+
+@SuppressWarnings("all")
+@org.apache.avro.specific.AvroGenerated
+public class TextGeometry extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
+  private static final long serialVersionUID = 8264944508099388832L;
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"TextGeometry\",\"namespace\":\"com.alibaba.dts.formats.avro\",\"fields\":[{\"name\":\"type\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}},{\"name\":\"value\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}}]}");
+  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
+
+  private static SpecificData MODEL$ = new SpecificData();
+
+  private static final BinaryMessageEncoder<TextGeometry> ENCODER =
+      new BinaryMessageEncoder<TextGeometry>(MODEL$, SCHEMA$);
+
+  private static final BinaryMessageDecoder<TextGeometry> DECODER =
+      new BinaryMessageDecoder<TextGeometry>(MODEL$, SCHEMA$);
+
+  /**
+   * Return the BinaryMessageDecoder instance used by this class.
+   */
+  public static BinaryMessageDecoder<TextGeometry> getDecoder() {
+    return DECODER;
+  }
+
+  /**
+   * Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}.
+   * @param resolver a {@link SchemaStore} used to find schemas by fingerprint
+   */
+  public static BinaryMessageDecoder<TextGeometry> createDecoder(SchemaStore resolver) {
+    return new BinaryMessageDecoder<TextGeometry>(MODEL$, SCHEMA$, resolver);
+  }
+
+  /** Serializes this TextGeometry to a ByteBuffer. */
+  public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException {
+    return ENCODER.encode(this);
+  }
+
+  /** Deserializes a TextGeometry from a ByteBuffer. */
+  public static TextGeometry fromByteBuffer(
+      java.nio.ByteBuffer b) throws java.io.IOException {
+    return DECODER.decode(b);
+  }
+
+  @Deprecated public java.lang.String type;
+  @Deprecated public java.lang.String value;
+
+  /**
+   * Default constructor.  Note that this does not initialize fields
+   * to their default values from the schema.  If that is desired then
+   * one should use <code>newBuilder()</code>.
+   */
+  public TextGeometry() {}
+
+  /**
+   * All-args constructor.
+   * @param type The new value for type
+   * @param value The new value for value
+   */
+  public TextGeometry(java.lang.String type, java.lang.String value) {
+    this.type = type;
+    this.value = value;
+  }
+
+  public org.apache.avro.Schema getSchema() { return SCHEMA$; }
+  // Used by DatumWriter.  Applications should not call.
+  public java.lang.Object get(int field$) {
+    switch (field$) {
+    case 0: return type;
+    case 1: return value;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  // Used by DatumReader.  Applications should not call.
+  @SuppressWarnings(value="unchecked")
+  public void put(int field$, java.lang.Object value$) {
+    switch (field$) {
+    case 0: type = (java.lang.String)value$; break;
+    case 1: value = (java.lang.String)value$; break;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  /**
+   * Gets the value of the 'type' field.
+   * @return The value of the 'type' field.
+   */
+  public java.lang.String getType() {
+    return type;
+  }
+
+  /**
+   * Sets the value of the 'type' field.
+   * @param value the value to set.
+   */
+  public void setType(java.lang.String value) {
+    this.type = value;
+  }
+
+  /**
+   * Gets the value of the 'value' field.
+   * @return The value of the 'value' field.
+   */
+  public java.lang.String getValue() {
+    return value;
+  }
+
+  /**
+   * Sets the value of the 'value' field.
+   * @param value the value to set.
+   */
+  public void setValue(java.lang.String value) {
+    this.value = value;
+  }
+
+  /**
+   * Creates a new TextGeometry RecordBuilder.
+   * @return A new TextGeometry RecordBuilder
+   */
+  public static com.alibaba.dts.formats.avro.TextGeometry.Builder newBuilder() {
+    return new com.alibaba.dts.formats.avro.TextGeometry.Builder();
+  }
+
+  /**
+   * Creates a new TextGeometry RecordBuilder by copying an existing Builder.
+   * @param other The existing builder to copy.
+   * @return A new TextGeometry RecordBuilder
+   */
+  public static com.alibaba.dts.formats.avro.TextGeometry.Builder newBuilder(com.alibaba.dts.formats.avro.TextGeometry.Builder other) {
+    return new com.alibaba.dts.formats.avro.TextGeometry.Builder(other);
+  }
+
+  /**
+   * Creates a new TextGeometry RecordBuilder by copying an existing TextGeometry instance.
+   * @param other The existing instance to copy.
+   * @return A new TextGeometry RecordBuilder
+   */
+  public static com.alibaba.dts.formats.avro.TextGeometry.Builder newBuilder(com.alibaba.dts.formats.avro.TextGeometry other) {
+    return new com.alibaba.dts.formats.avro.TextGeometry.Builder(other);
+  }
+
+  /**
+   * RecordBuilder for TextGeometry instances.
+   */
+  public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<TextGeometry>
+    implements org.apache.avro.data.RecordBuilder<TextGeometry> {
+
+    private java.lang.String type;
+    private java.lang.String value;
+
+    /** Creates a new Builder */
+    private Builder() {
+      super(SCHEMA$);
+    }
+
+    /**
+     * Creates a Builder by copying an existing Builder.
+     * @param other The existing Builder to copy.
+     */
+    private Builder(com.alibaba.dts.formats.avro.TextGeometry.Builder other) {
+      super(other);
+      if (isValidValue(fields()[0], other.type)) {
+        this.type = data().deepCopy(fields()[0].schema(), other.type);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.value)) {
+        this.value = data().deepCopy(fields()[1].schema(), other.value);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+     * Creates a Builder by copying an existing TextGeometry instance
+     * @param other The existing instance to copy.
+     */
+    private Builder(com.alibaba.dts.formats.avro.TextGeometry other) {
+            super(SCHEMA$);
+      if (isValidValue(fields()[0], other.type)) {
+        this.type = data().deepCopy(fields()[0].schema(), other.type);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.value)) {
+        this.value = data().deepCopy(fields()[1].schema(), other.value);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+      * Gets the value of the 'type' field.
+      * @return The value.
+      */
+    public java.lang.String getType() {
+      return type;
+    }
+
+    /**
+      * Sets the value of the 'type' field.
+      * @param value The value of 'type'.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.TextGeometry.Builder setType(java.lang.String value) {
+      validate(fields()[0], value);
+      this.type = value;
+      fieldSetFlags()[0] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'type' field has been set.
+      * @return True if the 'type' field has been set, false otherwise.
+      */
+    public boolean hasType() {
+      return fieldSetFlags()[0];
+    }
+
+
+    /**
+      * Clears the value of the 'type' field.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.TextGeometry.Builder clearType() {
+      type = null;
+      fieldSetFlags()[0] = false;
+      return this;
+    }
+
+    /**
+      * Gets the value of the 'value' field.
+      * @return The value.
+      */
+    public java.lang.String getValue() {
+      return value;
+    }
+
+    /**
+      * Sets the value of the 'value' field.
+      * @param value The value of 'value'.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.TextGeometry.Builder setValue(java.lang.String value) {
+      validate(fields()[1], value);
+      this.value = value;
+      fieldSetFlags()[1] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'value' field has been set.
+      * @return True if the 'value' field has been set, false otherwise.
+      */
+    public boolean hasValue() {
+      return fieldSetFlags()[1];
+    }
+
+
+    /**
+      * Clears the value of the 'value' field.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.TextGeometry.Builder clearValue() {
+      value = null;
+      fieldSetFlags()[1] = false;
+      return this;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public TextGeometry build() {
+      try {
+        TextGeometry record = new TextGeometry();
+        record.type = fieldSetFlags()[0] ? this.type : (java.lang.String) defaultValue(fields()[0]);
+        record.value = fieldSetFlags()[1] ? this.value : (java.lang.String) defaultValue(fields()[1]);
+        return record;
+      } catch (java.lang.Exception e) {
+        throw new org.apache.avro.AvroRuntimeException(e);
+      }
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumWriter<TextGeometry>
+    WRITER$ = (org.apache.avro.io.DatumWriter<TextGeometry>)MODEL$.createDatumWriter(SCHEMA$);
+
+  @Override public void writeExternal(java.io.ObjectOutput out)
+    throws java.io.IOException {
+    WRITER$.write(this, SpecificData.getEncoder(out));
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumReader<TextGeometry>
+    READER$ = (org.apache.avro.io.DatumReader<TextGeometry>)MODEL$.createDatumReader(SCHEMA$);
+
+  @Override public void readExternal(java.io.ObjectInput in)
+    throws java.io.IOException {
+    READER$.read(this, SpecificData.getDecoder(in));
+  }
+
+}

src/main/java/com/alibaba/dts/formats/avro/TextObject.java

+/**
+ * Autogenerated by Avro
+ *
+ * DO NOT EDIT DIRECTLY
+ */
+package com.alibaba.dts.formats.avro;
+
+import org.apache.avro.specific.SpecificData;
+import org.apache.avro.message.BinaryMessageEncoder;
+import org.apache.avro.message.BinaryMessageDecoder;
+import org.apache.avro.message.SchemaStore;
+
+@SuppressWarnings("all")
+@org.apache.avro.specific.AvroGenerated
+public class TextObject extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
+  private static final long serialVersionUID = -4704733630955722876L;
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"TextObject\",\"namespace\":\"com.alibaba.dts.formats.avro\",\"fields\":[{\"name\":\"type\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}},{\"name\":\"value\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}}]}");
+  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
+
+  private static SpecificData MODEL$ = new SpecificData();
+
+  private static final BinaryMessageEncoder<TextObject> ENCODER =
+      new BinaryMessageEncoder<TextObject>(MODEL$, SCHEMA$);
+
+  private static final BinaryMessageDecoder<TextObject> DECODER =
+      new BinaryMessageDecoder<TextObject>(MODEL$, SCHEMA$);
+
+  /**
+   * Return the BinaryMessageDecoder instance used by this class.
+   */
+  public static BinaryMessageDecoder<TextObject> getDecoder() {
+    return DECODER;
+  }
+
+  /**
+   * Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}.
+   * @param resolver a {@link SchemaStore} used to find schemas by fingerprint
+   */
+  public static BinaryMessageDecoder<TextObject> createDecoder(SchemaStore resolver) {
+    return new BinaryMessageDecoder<TextObject>(MODEL$, SCHEMA$, resolver);
+  }
+
+  /** Serializes this TextObject to a ByteBuffer. */
+  public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException {
+    return ENCODER.encode(this);
+  }
+
+  /** Deserializes a TextObject from a ByteBuffer. */
+  public static TextObject fromByteBuffer(
+      java.nio.ByteBuffer b) throws java.io.IOException {
+    return DECODER.decode(b);
+  }
+
+  @Deprecated public java.lang.String type;
+  @Deprecated public java.lang.String value;
+
+  /**
+   * Default constructor.  Note that this does not initialize fields
+   * to their default values from the schema.  If that is desired then
+   * one should use <code>newBuilder()</code>.
+   */
+  public TextObject() {}
+
+  /**
+   * All-args constructor.
+   * @param type The new value for type
+   * @param value The new value for value
+   */
+  public TextObject(java.lang.String type, java.lang.String value) {
+    this.type = type;
+    this.value = value;
+  }
+
+  public org.apache.avro.Schema getSchema() { return SCHEMA$; }
+  // Used by DatumWriter.  Applications should not call.
+  public java.lang.Object get(int field$) {
+    switch (field$) {
+    case 0: return type;
+    case 1: return value;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  // Used by DatumReader.  Applications should not call.
+  @SuppressWarnings(value="unchecked")
+  public void put(int field$, java.lang.Object value$) {
+    switch (field$) {
+    case 0: type = (java.lang.String)value$; break;
+    case 1: value = (java.lang.String)value$; break;
+    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
+    }
+  }
+
+  /**
+   * Gets the value of the 'type' field.
+   * @return The value of the 'type' field.
+   */
+  public java.lang.String getType() {
+    return type;
+  }
+
+  /**
+   * Sets the value of the 'type' field.
+   * @param value the value to set.
+   */
+  public void setType(java.lang.String value) {
+    this.type = value;
+  }
+
+  /**
+   * Gets the value of the 'value' field.
+   * @return The value of the 'value' field.
+   */
+  public java.lang.String getValue() {
+    return value;
+  }
+
+  /**
+   * Sets the value of the 'value' field.
+   * @param value the value to set.
+   */
+  public void setValue(java.lang.String value) {
+    this.value = value;
+  }
+
+  /**
+   * Creates a new TextObject RecordBuilder.
+   * @return A new TextObject RecordBuilder
+   */
+  public static com.alibaba.dts.formats.avro.TextObject.Builder newBuilder() {
+    return new com.alibaba.dts.formats.avro.TextObject.Builder();
+  }
+
+  /**
+   * Creates a new TextObject RecordBuilder by copying an existing Builder.
+   * @param other The existing builder to copy.
+   * @return A new TextObject RecordBuilder
+   */
+  public static com.alibaba.dts.formats.avro.TextObject.Builder newBuilder(com.alibaba.dts.formats.avro.TextObject.Builder other) {
+    return new com.alibaba.dts.formats.avro.TextObject.Builder(other);
+  }
+
+  /**
+   * Creates a new TextObject RecordBuilder by copying an existing TextObject instance.
+   * @param other The existing instance to copy.
+   * @return A new TextObject RecordBuilder
+   */
+  public static com.alibaba.dts.formats.avro.TextObject.Builder newBuilder(com.alibaba.dts.formats.avro.TextObject other) {
+    return new com.alibaba.dts.formats.avro.TextObject.Builder(other);
+  }
+
+  /**
+   * RecordBuilder for TextObject instances.
+   */
+  public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<TextObject>
+    implements org.apache.avro.data.RecordBuilder<TextObject> {
+
+    private java.lang.String type;
+    private java.lang.String value;
+
+    /** Creates a new Builder */
+    private Builder() {
+      super(SCHEMA$);
+    }
+
+    /**
+     * Creates a Builder by copying an existing Builder.
+     * @param other The existing Builder to copy.
+     */
+    private Builder(com.alibaba.dts.formats.avro.TextObject.Builder other) {
+      super(other);
+      if (isValidValue(fields()[0], other.type)) {
+        this.type = data().deepCopy(fields()[0].schema(), other.type);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.value)) {
+        this.value = data().deepCopy(fields()[1].schema(), other.value);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+     * Creates a Builder by copying an existing TextObject instance
+     * @param other The existing instance to copy.
+     */
+    private Builder(com.alibaba.dts.formats.avro.TextObject other) {
+            super(SCHEMA$);
+      if (isValidValue(fields()[0], other.type)) {
+        this.type = data().deepCopy(fields()[0].schema(), other.type);
+        fieldSetFlags()[0] = true;
+      }
+      if (isValidValue(fields()[1], other.value)) {
+        this.value = data().deepCopy(fields()[1].schema(), other.value);
+        fieldSetFlags()[1] = true;
+      }
+    }
+
+    /**
+      * Gets the value of the 'type' field.
+      * @return The value.
+      */
+    public java.lang.String getType() {
+      return type;
+    }
+
+    /**
+      * Sets the value of the 'type' field.
+      * @param value The value of 'type'.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.TextObject.Builder setType(java.lang.String value) {
+      validate(fields()[0], value);
+      this.type = value;
+      fieldSetFlags()[0] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'type' field has been set.
+      * @return True if the 'type' field has been set, false otherwise.
+      */
+    public boolean hasType() {
+      return fieldSetFlags()[0];
+    }
+
+
+    /**
+      * Clears the value of the 'type' field.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.TextObject.Builder clearType() {
+      type = null;
+      fieldSetFlags()[0] = false;
+      return this;
+    }
+
+    /**
+      * Gets the value of the 'value' field.
+      * @return The value.
+      */
+    public java.lang.String getValue() {
+      return value;
+    }
+
+    /**
+      * Sets the value of the 'value' field.
+      * @param value The value of 'value'.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.TextObject.Builder setValue(java.lang.String value) {
+      validate(fields()[1], value);
+      this.value = value;
+      fieldSetFlags()[1] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'value' field has been set.
+      * @return True if the 'value' field has been set, false otherwise.
+      */
+    public boolean hasValue() {
+      return fieldSetFlags()[1];
+    }
+
+
+    /**
+      * Clears the value of the 'value' field.
+      * @return This builder.
+      */
+    public com.alibaba.dts.formats.avro.TextObject.Builder clearValue() {
+      value = null;
+      fieldSetFlags()[1] = false;
+      return this;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public TextObject build() {
+      try {
+        TextObject record = new TextObject();
+        record.type = fieldSetFlags()[0] ? this.type : (java.lang.String) defaultValue(fields()[0]);
+        record.value = fieldSetFlags()[1] ? this.value : (java.lang.String) defaultValue(fields()[1]);
+        return record;
+      } catch (java.lang.Exception e) {
+        throw new org.apache.avro.AvroRuntimeException(e);
+      }
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumWriter<TextObject>
+    WRITER$ = (org.apache.avro.io.DatumWriter<TextObject>)MODEL$.createDatumWriter(SCHEMA$);
+
+  @Override public void writeExternal(java.io.ObjectOutput out)
+    throws java.io.IOException {
+    WRITER$.write(this, SpecificData.getEncoder(out));
+  }
+
+  @SuppressWarnings("unchecked")
+  private static final org.apache.avro.io.DatumReader<TextObject>
+    READER$ = (org.apache.avro.io.DatumReader<TextObject>)MODEL$.createDatumReader(SCHEMA$);
+
+  @Override public void readExternal(java.io.ObjectInput in)
+    throws java.io.IOException {
+    READER$.read(this, SpecificData.getDecoder(in));
+  }
+
+}

src/main/java/com/alibaba/dts/metastore/KafkaMetaStore.java

+package com.alibaba.dts.metastore;
+
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.consumer.OffsetCommitCallback;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import com.alibaba.dts.common.Checkpoint;
+
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.Future;
+
+public class KafkaMetaStore implements MetaStore<Checkpoint> {
+    private static final Logger log = LoggerFactory.getLogger(KafkaMetaStore.class);
+
+    private volatile KafkaConsumer kafkaConsumer;
+
+    public KafkaMetaStore(KafkaConsumer kafkaConsumer) {
+        this.kafkaConsumer = kafkaConsumer;
+    }
+
+    public void resetKafkaConsumer(KafkaConsumer newConsumer) {
+        this.kafkaConsumer = newConsumer;
+    }
+
+    @Override
+    public Future<Checkpoint> serializeTo(TopicPartition topicPartition, String group, Checkpoint value) {
+        KafkaFutureImpl ret = new KafkaFutureImpl();
+        if (null != kafkaConsumer) {
+            OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(value.getOffset(), String.valueOf(value.getTimeStamp()));
+            // Notice: commitAsync is only put commit offset request to sending queue, the future  result will be driven by KafkaConsumer.poll() function
+            // So if you only call this method but not poll, you may not wait offset commit call back
+            kafkaConsumer.commitAsync(Collections.singletonMap(topicPartition, offsetAndMetadata), new OffsetCommitCallback() {
+                @Override
+                public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
+                    if (null != exception) {
+                        log.warn("KafkaMetaStore: Commit offset for group[" + group + "] topicPartition[" + topicPartition.toString() + "] " +
+                                value.toString() + " failed cause " + exception.getMessage(), exception);
+                        ret.completeExceptionally(exception);
+                    } else {
+                        log.debug("KafkaMetaStore:Commit offset success for group[{}] topicPartition [{}] {}", group, topicPartition, value);
+                        ret.complete(value);
+                    }
+                }
+            });
+        } else {
+            log.warn("KafkaMetaStore: kafka consumer not set, ignore report");
+            ret.complete(value);
+        }
+        return ret;
+
+    }
+
+    @Override
+    public Checkpoint deserializeFrom(TopicPartition topicPartition, String group) {
+        if (null != kafkaConsumer) {
+            OffsetAndMetadata offsetAndMetadata = kafkaConsumer.committed(topicPartition);
+            if (null != offsetAndMetadata) {
+                return new Checkpoint(topicPartition, Long.valueOf(offsetAndMetadata.metadata()), offsetAndMetadata.offset(), offsetAndMetadata.metadata());
+            } else {
+                return null;
+            }
+        } else {
+            log.warn("KafkaMetaStore: kafka consumer not set, ignore fetch offset");
+            throw new KafkaException("KafkaMetaStore: kafka consumer not set, ignore fetch offset for group[" + group + "] and tp [" + topicPartition + "]");
+        }
+    }
+}

src/main/java/com/alibaba/dts/metastore/LocalFileMetaStore.java

+package com.alibaba.dts.metastore;
+
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+import com.alibaba.dts.common.Checkpoint;
+import com.alibaba.dts.common.AtomicFileStore;
+
+import java.util.*;
+import java.util.concurrent.Future;
+
+public class LocalFileMetaStore implements MetaStore<Checkpoint> {
+    private static final String GROUP_ID_NAME = "groupID";
+    private static final String STREAM_CHECKPOINT_NAME = "streamCheckpoint";
+    private static final String TOPIC_NAME = "topic";
+    private static final String PARTITION_NAME = "partition";
+    private static final String OFFSET_NAME = "offset";
+    private static final String TIMESTAMP_NAME = "timestamp";
+    private static final String INFO_NAME = "info";
+
+    private static class StoreElement {
+        final String groupName;
+        final Map<TopicPartition, Checkpoint> streamCheckpoint;
+
+        private StoreElement(String groupName, Map<TopicPartition, Checkpoint> streamCheckpoint) {
+            this.groupName = groupName;
+            this.streamCheckpoint = streamCheckpoint;
+        }
+    }
+
+    private final AtomicFileStore fileStore;
+    private final Map<String, Map<TopicPartition, Checkpoint>> inMemStore = new HashMap<>();
+    public LocalFileMetaStore(String fileName) {
+        this.fileStore = new AtomicFileStore(fileName);
+    }
+
+    private String toJson(StoreElement storeElement) {
+        JSONObject jsonObject = new JSONObject();
+        jsonObject.put(GROUP_ID_NAME, storeElement.groupName);
+        JSONArray jsonArray = new JSONArray();
+        storeElement.streamCheckpoint.forEach((tp, checkpoint) -> {
+            JSONObject streamCheckpointJsonObject = new JSONObject();
+            streamCheckpointJsonObject.put(TOPIC_NAME, tp.topic());
+            streamCheckpointJsonObject.put(PARTITION_NAME, tp.partition());
+            streamCheckpointJsonObject.put(OFFSET_NAME, checkpoint.getOffset());
+            streamCheckpointJsonObject.put(TIMESTAMP_NAME, checkpoint.getTimeStamp());
+            streamCheckpointJsonObject.put(INFO_NAME, checkpoint.getInfo());
+            jsonArray.add(streamCheckpointJsonObject);
+        });
+
+        jsonObject.put(STREAM_CHECKPOINT_NAME, jsonArray);
+        return jsonObject.toJSONString();
+    }
+
+    private StoreElement fromString(String jsonString) {
+        JSONObject jsonObject = JSONObject.parseObject(jsonString);
+        String groupName = jsonObject.getString(GROUP_ID_NAME);
+        JSONArray streamCheckpointJsonObject = jsonObject.getJSONArray(STREAM_CHECKPOINT_NAME);
+        Map<TopicPartition, Checkpoint> checkpointInfo = new HashMap<>();
+        for (Object o : streamCheckpointJsonObject) {
+            JSONObject tpAndCheckpoint = (JSONObject) o;
+            String topic = tpAndCheckpoint.getString(TOPIC_NAME);
+            int partition = tpAndCheckpoint.getInteger(PARTITION_NAME);
+            long offset = tpAndCheckpoint.getLong(OFFSET_NAME);
+            long timestamp = tpAndCheckpoint.getLong(TIMESTAMP_NAME);
+            String info = tpAndCheckpoint.getString(INFO_NAME);
+            checkpointInfo.put(new TopicPartition(topic, partition), new Checkpoint(new TopicPartition(topic, partition), timestamp, offset, info));
+        }
+
+        return new StoreElement(groupName, checkpointInfo);
+    }
+
+    @Override
+    public Future<Checkpoint> serializeTo(TopicPartition topicPartition, String groupID, Checkpoint value) {
+        Map<TopicPartition, Checkpoint> topicPartitionCheckpoint = inMemStore.get(groupID);
+        if (null == topicPartitionCheckpoint) {
+            topicPartitionCheckpoint = new HashMap<>();
+        }
+        topicPartitionCheckpoint.put(topicPartition, value);
+        inMemStore.put(groupID, topicPartitionCheckpoint);
+
+        List<String> toSerialize = new LinkedList<>();
+        inMemStore.forEach((k, v) ->{
+            toSerialize.add(toJson(new StoreElement(k, v)));
+        });
+        fileStore.updateContent(toSerialize);
+        KafkaFutureImpl ret =  new KafkaFutureImpl<>();
+        ret.complete(value);
+        return ret;
+    }
+
+    @Override
+    public Checkpoint deserializeFrom(TopicPartition topicPartition, String groupID) {
+        Map<TopicPartition, Checkpoint> tpAndCheckpointMap = inMemStore.get(groupID);
+        if (null != tpAndCheckpointMap) {
+            Checkpoint ret = tpAndCheckpointMap.get(topicPartition);
+            if (null != ret) {
+                return ret;
+
+            }
+        }
+        List<String> storedCheckpoint = fileStore.getContent();
+        for (String checkpoint : storedCheckpoint) {
+            StoreElement storeElement = fromString(checkpoint);
+            // add to cache
+            if (!inMemStore.containsKey(storeElement.groupName)) {
+                inMemStore.put(storeElement.groupName, storeElement.streamCheckpoint);
+            }
+            if (StringUtils.equals(storeElement.groupName, groupID)) {
+                return storeElement.streamCheckpoint.get(topicPartition);
+            }
+        }
+        return null;
+    }
+
+}

src/main/java/com/alibaba/dts/metastore/MetaStore.java

+package com.alibaba.dts.metastore;
+
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.concurrent.Future;
+
+public interface MetaStore<V> {
+    Future<V> serializeTo(TopicPartition topicPartition, String group, V value);
+    V deserializeFrom(TopicPartition topicPartition, String group);
+}

src/main/java/com/alibaba/dts/metastore/MetaStoreCenter.java

+package com.alibaba.dts.metastore;
+
+import org.apache.kafka.common.TopicPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import com.alibaba.dts.common.Checkpoint;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class MetaStoreCenter {
+    private static final Logger log = LoggerFactory.getLogger(MetaStoreCenter.class);
+    private final Map<String, MetaStore<Checkpoint>> registeredStore = new HashMap<>();
+    public MetaStoreCenter() {
+
+    }
+
+    public void registerStore(String name, MetaStore metaStore) {
+        log.info("MetaStoreCenter: register metaStore {}", name);
+        registeredStore.put(name, metaStore);
+    }
+
+    public void store(TopicPartition topicPartition, String group, Checkpoint value) {
+        registeredStore.values().forEach(v -> {
+            v.serializeTo(topicPartition, group, value);
+        });
+    }
+
+    public Checkpoint seek(String storeName, TopicPartition tp, String group) {
+        MetaStore<Checkpoint> metaStore = registeredStore.get(storeName);
+        if (null != metaStore) {
+            return metaStore.deserializeFrom(tp, group);
+        } else {
+            return null;
+        }
+    }
+}

src/main/java/com/alibaba/dts/recordgenerator/ConsumerWrapFactory.java

+package com.alibaba.dts.recordgenerator;
+
+import java.util.Properties;
+
+public interface ConsumerWrapFactory {
+    public ConsumerWrap getConsumerWrap(Properties properties);
+
+    public static class KafkaConsumerWrapFactory implements ConsumerWrapFactory {
+        @Override
+        public ConsumerWrap getConsumerWrap(Properties properties) {
+            return new ConsumerWrap.DefaultConsumerWrap(properties);
+        }
+    }
+}

src/main/java/com/alibaba/dts/recordgenerator/OffsetCommitCallBack.java

+package com.alibaba.dts.recordgenerator;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.TopicPartition;
+
+public interface OffsetCommitCallBack {
+    void commit(TopicPartition tp, long timestamp, long offset, String metadata);
+}