Bump protobuf-java from 3.2.0 to 3.16.1 in /hbase11xsqlwriter

Bumps [protobuf-java](https://github.com/protocolbuffers/protobuf) from 3.2.0 to 3.16.1.
- [Release notes](https://github.com/protocolbuffers/protobuf/releases)
- [Changelog](https://github.com/protocolbuffers/protobuf/blob/master/generate_changelog.py)
- [Commits](https://github.com/protocolbuffers/protobuf/compare/v3.2.0...v3.16.1)

---
updated-dependencies:
- dependency-name: com.google.protobuf:protobuf-java
  dependency-type: direct:production
...

Signed-off-by: dependabot[bot] <support@github.com>

README.md

@@ -1,11 +1,14 @@
 ![Datax-logo](https://github.com/alibaba/DataX/blob/master/images/DataX-logo.jpg)
 
 
-
 # DataX
 
-DataX 是阿里巴巴集团内被广泛使用的离线数据同步工具/平台，实现包括 MySQL、Oracle、SqlServer、Postgre、HDFS、Hive、ADS、HBase、TableStore(OTS)、MaxCompute(ODPS)、DRDS 等各种异构数据源之间高效的数据同步功能。
+DataX 是阿里云 [DataWorks数据集成](https://www.aliyun.com/product/bigdata/ide) 的开源版本，在阿里巴巴集团内被广泛使用的离线数据同步工具/平台。DataX 实现了包括 MySQL、Oracle、OceanBase、SqlServer、Postgre、HDFS、Hive、ADS、HBase、TableStore(OTS)、MaxCompute(ODPS)、Hologres、DRDS 等各种异构数据源之间高效的数据同步功能。
 
+# DataX 商业版本
+阿里云DataWorks数据集成是DataX团队在阿里云上的商业化产品，致力于提供复杂网络环境下、丰富的异构数据源之间高速稳定的数据移动能力，以及繁杂业务背景下的数据同步解决方案。目前已经支持云上近3000家客户，单日同步数据超过3万亿条。DataWorks数据集成目前支持离线50+种数据源，可以进行整库迁移、批量上云、增量同步、分库分表等各类同步解决方案。2020年更新实时同步能力，2020年更新实时同步能力，支持10+种数据源的读写任意组合。提供MySQL，Oracle等多种数据源到阿里云MaxCompute，Hologres等大数据引擎的一键全增量同步解决方案。
+
+商业版本参见：  https://www.aliyun.com/product/bigdata/ide
 
 
 # Features
@@ -36,6 +39,7 @@ DataX目前已经有了比较全面的插件体系，主流的RDBMS数据库、N
 | ------------ | ---------- | :-------: | :-------: |:-------: |
 | RDBMS 关系型数据库 | MySQL      |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/mysqlreader/doc/mysqlreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/mysqlwriter/doc/mysqlwriter.md)|
 |              | Oracle     |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/oraclereader/doc/oraclereader.md) 、[写](https://github.com/alibaba/DataX/blob/master/oraclewriter/doc/oraclewriter.md)|
+|              | OceanBase  |     √     |     √     |[读](https://open.oceanbase.com/docs/community/oceanbase-database/V3.1.0/use-datax-to-full-migration-data-to-oceanbase) 、[写](https://open.oceanbase.com/docs/community/oceanbase-database/V3.1.0/use-datax-to-full-migration-data-to-oceanbase)|
 |              | SQLServer  |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/sqlserverreader/doc/sqlserverreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/sqlserverwriter/doc/sqlserverwriter.md)|
 |              | PostgreSQL |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/postgresqlreader/doc/postgresqlreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/postgresqlwriter/doc/postgresqlwriter.md)|
 |              | DRDS |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/drdsreader/doc/drdsreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/drdswriter/doc/drdswriter.md)|
@@ -43,13 +47,13 @@ DataX目前已经有了比较全面的插件体系，主流的RDBMS数据库、N
 | 阿里云数仓数据存储    | ODPS       |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/odpsreader/doc/odpsreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/odpswriter/doc/odpswriter.md)|
 |              | ADS        |           |     √     |[写](https://github.com/alibaba/DataX/blob/master/adswriter/doc/adswriter.md)|
 |              | OSS        |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/ossreader/doc/ossreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/osswriter/doc/osswriter.md)|
-|              | OCS        |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/ocsreader/doc/ocsreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/ocswriter/doc/ocswriter.md)|
+|              | OCS        |           |     √     |[写](https://github.com/alibaba/DataX/blob/master/ocswriter/doc/ocswriter.md)|
 | NoSQL数据存储    | OTS        |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/otsreader/doc/otsreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/otswriter/doc/otswriter.md)|
 |              | Hbase0.94  |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/hbase094xreader/doc/hbase094xreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/hbase094xwriter/doc/hbase094xwriter.md)|
 |              | Hbase1.1   |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/hbase11xreader/doc/hbase11xreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/hbase11xwriter/doc/hbase11xwriter.md)|
 |              | Phoenix4.x   |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/hbase11xsqlreader/doc/hbase11xsqlreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/hbase11xsqlwriter/doc/hbase11xsqlwriter.md)|
 |              | Phoenix5.x   |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/hbase20xsqlreader/doc/hbase20xsqlreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/hbase20xsqlwriter/doc/hbase20xsqlwriter.md)|
-|              | MongoDB    |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/mongoreader/doc/mongoreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/mongowriter/doc/mongowriter.md)|
+|              | MongoDB    |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/mongodbreader/doc/mongodbreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/mongodbwriter/doc/mongodbwriter.md)|
 |              | Hive       |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/hdfsreader/doc/hdfsreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/hdfswriter/doc/hdfswriter.md)|
 |              | Cassandra       |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/cassandrareader/doc/cassandrareader.md) 、[写](https://github.com/alibaba/DataX/blob/master/cassandrawriter/doc/cassandrawriter.md)|
 | 无结构化数据存储     | TxtFile    |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/txtfilereader/doc/txtfilereader.md) 、[写](https://github.com/alibaba/DataX/blob/master/txtfilewriter/doc/txtfilewriter.md)|
@@ -59,9 +63,33 @@ DataX目前已经有了比较全面的插件体系，主流的RDBMS数据库、N
 | 时间序列数据库 | OpenTSDB | √ |  |[读](https://github.com/alibaba/DataX/blob/master/opentsdbreader/doc/opentsdbreader.md)|
 |  | TSDB | √ | √ |[读](https://github.com/alibaba/DataX/blob/master/tsdbreader/doc/tsdbreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/tsdbwriter/doc/tsdbhttpwriter.md)|
 
+# 阿里云DataWorks数据集成
+
+目前DataX的已有能力已经全部融和进阿里云的数据集成，并且比DataX更加高效、安全，同时数据集成具备DataX不具备的其它高级特性和功能。可以理解为数据集成是DataX的全面升级的商业化用版本，为企业可以提供稳定、可靠、安全的数据传输服务。与DataX相比，数据集成主要有以下几大突出特点：
+
+支持实时同步：
+
+- 功能简介：https://help.aliyun.com/document_detail/181912.html
+- 支持的数据源：https://help.aliyun.com/document_detail/146778.html
+- 支持数据处理：https://help.aliyun.com/document_detail/146777.html
+
+离线同步数据源种类大幅度扩充：
+
+- 新增比如：DB2、Kafka、Hologres、MetaQ、SAPHANA、达梦等等，持续扩充中
+- 离线同步支持的数据源：https://help.aliyun.com/document_detail/137670.html
+- 具备同步解决方案：
+    - 解决方案系统：https://help.aliyun.com/document_detail/171765.html
+    - 一键全增量：https://help.aliyun.com/document_detail/175676.html
+    - 整库迁移：https://help.aliyun.com/document_detail/137809.html
+    - 批量上云：https://help.aliyun.com/document_detail/146671.html
+    - 更新更多能力请访问：https://help.aliyun.com/document_detail/137663.html
+
+
 # 我要开发新的插件
+
 请点击：[DataX插件开发宝典](https://github.com/alibaba/DataX/blob/master/dataxPluginDev.md)
 
+
 # 项目成员
 
 核心Contributions: 言柏 、枕水、秋奇、青砾、一斅、云时
@@ -108,7 +136,23 @@ This software is free to use under the Apache License [Apache license](https://g
     8. 对高并发、高稳定可用性、高性能、大数据处理有过实际项目及产品经验者优先考虑；
     9. 有大数据产品、云产品、中间件技术解决方案者优先考虑。
 ````
-钉钉用户群：23169395
+钉钉用户群：
 
+- DataX开源用户交流群
+    - <img src="https://github.com/alibaba/DataX/blob/master/images/DataX%E5%BC%80%E6%BA%90%E7%94%A8%E6%88%B7%E4%BA%A4%E6%B5%81%E7%BE%A4.jpg" width="20%" height="20%">
 
+- DataX开源用户交流群2
+    - <img src="https://github.com/alibaba/DataX/blob/master/images/DataX%E5%BC%80%E6%BA%90%E7%94%A8%E6%88%B7%E4%BA%A4%E6%B5%81%E7%BE%A42.jpg" width="20%" height="20%">
+
+- DataX开源用户交流群3
+    - <img src="https://github.com/alibaba/DataX/blob/master/images/DataX%E5%BC%80%E6%BA%90%E7%94%A8%E6%88%B7%E4%BA%A4%E6%B5%81%E7%BE%A43.jpg" width="20%" height="20%">
+
+- DataX开源用户交流群4
+    - <img src="https://github.com/alibaba/DataX/blob/master/images/DataX%E5%BC%80%E6%BA%90%E7%94%A8%E6%88%B7%E4%BA%A4%E6%B5%81%E7%BE%A44.jpg" width="20%" height="20%">
+
+- DataX开源用户交流群5
+    - <img src="https://github.com/alibaba/DataX/blob/master/images/DataX%E5%BC%80%E6%BA%90%E7%94%A8%E6%88%B7%E4%BA%A4%E6%B5%81%E7%BE%A45.jpg" width="20%" height="20%">
+
+- DataX开源用户交流群6 
+    - <img src="https://user-images.githubusercontent.com/1905000/124073771-139cbd00-da75-11eb-9a3f-598cba145a76.png" width="20%" height="20%">
 

clickhousewriter/pom.xml

@@ -29,12 +29,6 @@
             <artifactId>datax-common</artifactId>
             <version>${datax-project-version}</version>
         </dependency>
-        <dependency>
-            <groupId>com.alibaba.datax</groupId>
-            <artifactId>simulator</artifactId>
-            <version>${datax-project-version}</version>
-            <scope>test</scope>
-        </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>

clickhousewriter/src/main/java/com/alibaba/datax/plugin/writer/clickhousewriter/ClickhouseWriter.java

@@ -12,7 +12,6 @@ import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
 import com.alibaba.datax.plugin.rdbms.writer.CommonRdbmsWriter;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONArray;
-import ru.yandex.clickhouse.ClickHouseTuple;
 
 import java.sql.Array;
 import java.sql.Connection;

core/pom.xml

@@ -41,12 +41,12 @@
         <dependency>
             <groupId>org.apache.httpcomponents</groupId>
             <artifactId>httpclient</artifactId>
-            <version>4.4</version>
+            <version>4.5</version>
         </dependency>
         <dependency>
             <groupId>org.apache.httpcomponents</groupId>
             <artifactId>fluent-hc</artifactId>
-            <version>4.4</version>
+            <version>4.5</version>
         </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>

core/src/main/bin/datax.py

@@ -1,23 +1,26 @@
 #!/usr/bin/env python
 # -*- coding:utf-8 -*-
 
-import sys
-import os
-import signal
-import subprocess
-import time
-import re
-import socket
-import json
-from optparse import OptionParser
-from optparse import OptionGroup
-from string import Template
 import codecs
+import json
+import os
 import platform
+import re
+import signal
+import socket
+import subprocess
+import sys
+import time
+from optparse import OptionGroup
+from optparse import OptionParser
+from string import Template
+
+ispy2 = sys.version_info.major == 2
 
 def isWindows():
     return platform.system() == 'Windows'
 
+
 DATAX_HOME = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 
 DATAX_VERSION = 'DATAX-OPENSOURCE-3.0'
@@ -52,13 +55,19 @@ def getLocalIp():
 
 def suicide(signum, e):
     global child_process
-    print >> sys.stderr, "[Error] DataX receive unexpected signal %d, starts to suicide." % (signum)
+    if ispy2:
+        print >> sys.stderr, "[Error] DataX receive unexpected signal %d, starts to suicide." % (signum)
+    else:
+        print("[Error] DataX receive unexpected signal %d, starts to suicide." % (signum), sys.stderr)
 
     if child_process:
         child_process.send_signal(signal.SIGQUIT)
         time.sleep(1)
         child_process.kill()
-    print >> sys.stderr, "DataX Process was killed ! you did ?"
+    if ispy2:
+        print >> sys.stderr, "DataX Process was killed ! you did ?"
+    else:
+        print("DataX Process was killed ! you did ?", sys.stderr)
     sys.exit(RET_STATE["KILL"])
 
 
@@ -92,10 +101,10 @@ def getOptionParser():
                                        'if you have mutiple parameters: -p"-DtableName=your-table-name -DcolumnName=your-column-name".'
                                        'Note: you should config in you job tableName with ${tableName}.')
     prodEnvOptionGroup.add_option("-r", "--reader", metavar="<parameter used in view job config[reader] template>",
-                                  action="store", dest="reader",type="string",
+                                  action="store", dest="reader", type="string",
                                   help='View job config[reader] template, eg: mysqlreader,streamreader')
     prodEnvOptionGroup.add_option("-w", "--writer", metavar="<parameter used in view job config[writer] template>",
-                                  action="store", dest="writer",type="string",
+                                  action="store", dest="writer", type="string",
                                   help='View job config[writer] template, eg: mysqlwriter,streamwriter')
     parser.add_option_group(prodEnvOptionGroup)
 
@@ -108,45 +117,50 @@ def getOptionParser():
     parser.add_option_group(devEnvOptionGroup)
     return parser
 
+
 def generateJobConfigTemplate(reader, writer):
-    readerRef = "Please refer to the %s document:\n     https://github.com/alibaba/DataX/blob/master/%s/doc/%s.md \n" % (reader,reader,reader)
-    writerRef = "Please refer to the %s document:\n     https://github.com/alibaba/DataX/blob/master/%s/doc/%s.md \n " % (writer,writer,writer)
-    print readerRef
-    print writerRef
+    readerRef = "Please refer to the %s document:\n     https://github.com/alibaba/DataX/blob/master/%s/doc/%s.md \n" % (
+        reader, reader, reader)
+    writerRef = "Please refer to the %s document:\n     https://github.com/alibaba/DataX/blob/master/%s/doc/%s.md \n " % (
+        writer, writer, writer)
+    print(readerRef)
+    print(writerRef)
     jobGuid = 'Please save the following configuration as a json file and  use\n     python {DATAX_HOME}/bin/datax.py {JSON_FILE_NAME}.json \nto run the job.\n'
-    print jobGuid
-    jobTemplate={
-      "job": {
-        "setting": {
-          "speed": {
-            "channel": ""
-          }
-        },
-        "content": [
-          {
-            "reader": {},
-            "writer": {}
-          }
-        ]
-      }
+    print(jobGuid)
+    jobTemplate = {
+        "job": {
+            "setting": {
+                "speed": {
+                    "channel": ""
+                }
+            },
+            "content": [
+                {
+                    "reader": {},
+                    "writer": {}
+                }
+            ]
+        }
     }
-    readerTemplatePath = "%s/plugin/reader/%s/plugin_job_template.json" % (DATAX_HOME,reader)
-    writerTemplatePath = "%s/plugin/writer/%s/plugin_job_template.json" % (DATAX_HOME,writer)
+    readerTemplatePath = "%s/plugin/reader/%s/plugin_job_template.json" % (DATAX_HOME, reader)
+    writerTemplatePath = "%s/plugin/writer/%s/plugin_job_template.json" % (DATAX_HOME, writer)
     try:
-      readerPar = readPluginTemplate(readerTemplatePath);
-    except Exception, e:
-       print "Read reader[%s] template error: can\'t find file %s" % (reader,readerTemplatePath)
+        readerPar = readPluginTemplate(readerTemplatePath)
+    except:
+        print("Read reader[%s] template error: can\'t find file %s" % (reader, readerTemplatePath))
     try:
-      writerPar = readPluginTemplate(writerTemplatePath);
-    except Exception, e:
-      print "Read writer[%s] template error: : can\'t find file %s" % (writer,writerTemplatePath)
-    jobTemplate['job']['content'][0]['reader'] = readerPar;
-    jobTemplate['job']['content'][0]['writer'] = writerPar;
-    print json.dumps(jobTemplate, indent=4, sort_keys=True)
+        writerPar = readPluginTemplate(writerTemplatePath)
+    except:
+        print("Read writer[%s] template error: : can\'t find file %s" % (writer, writerTemplatePath))
+    jobTemplate['job']['content'][0]['reader'] = readerPar
+    jobTemplate['job']['content'][0]['writer'] = writerPar
+    print(json.dumps(jobTemplate, indent=4, sort_keys=True))
+
 
 def readPluginTemplate(plugin):
     with open(plugin, 'r') as f:
-            return json.load(f)
+        return json.load(f)
+
 
 def isUrl(path):
     if not path:
@@ -168,7 +182,7 @@ def buildStartCommand(options, args):
 
     if options.remoteDebug:
         tempJVMCommand = tempJVMCommand + " " + REMOTE_DEBUG_CONFIG
-        print 'local ip: ', getLocalIp()
+        print('local ip: ', getLocalIp())
 
     if options.loglevel:
         tempJVMCommand = tempJVMCommand + " " + ("-Dloglevel=%s" % (options.loglevel))
@@ -198,11 +212,11 @@ def buildStartCommand(options, args):
 
 
 def printCopyright():
-    print '''
+    print('''
 DataX (%s), From Alibaba !
 Copyright (C) 2010-2017, Alibaba Group. All Rights Reserved.
 
-''' % DATAX_VERSION
+''' % DATAX_VERSION)
     sys.stdout.flush()
 
 
@@ -211,7 +225,7 @@ if __name__ == "__main__":
     parser = getOptionParser()
     options, args = parser.parse_args(sys.argv[1:])
     if options.reader is not None and options.writer is not None:
-        generateJobConfigTemplate(options.reader,options.writer)
+        generateJobConfigTemplate(options.reader, options.writer)
         sys.exit(RET_STATE['OK'])
     if len(args) != 1:
         parser.print_help()

core/src/main/bin/perftrace.py

@@ -174,6 +174,9 @@ def parsePluginName(jdbcUrl, pluginType):
     db2Regex = re.compile('jdbc:(db2)://.*')
     if (db2Regex.match(jdbcUrl)):
         name = 'db2'
+    kingbaseesRegex = re.compile('jdbc:(kingbase8)://.*')
+    if (kingbaseesRegex.match(jdbcUrl)):
+        name = 'kingbasees'
     return "%s%s" % (name, pluginType)
 
 def renderDataXJson(paramsDict, readerOrWriter = 'reader', channel = 1):

core/src/main/java/com/alibaba/datax/core/job/JobContainer.java

@@ -427,7 +427,7 @@ public class JobContainer extends AbstractContainer {
             Long channelLimitedByteSpeed = this.configuration
                     .getLong(CoreConstant.DATAX_CORE_TRANSPORT_CHANNEL_SPEED_BYTE);
             if (channelLimitedByteSpeed == null || channelLimitedByteSpeed <= 0) {
-                DataXException.asDataXException(
+                throw DataXException.asDataXException(
                         FrameworkErrorCode.CONFIG_ERROR,
                         "在有总bps限速条件下，单个channel的bps值不能为空，也不能为非正数");
             }
@@ -448,7 +448,7 @@ public class JobContainer extends AbstractContainer {
             Long channelLimitedRecordSpeed = this.configuration.getLong(
                     CoreConstant.DATAX_CORE_TRANSPORT_CHANNEL_SPEED_RECORD);
             if (channelLimitedRecordSpeed == null || channelLimitedRecordSpeed <= 0) {
-                DataXException.asDataXException(FrameworkErrorCode.CONFIG_ERROR,
+                throw DataXException.asDataXException(FrameworkErrorCode.CONFIG_ERROR,
                         "在有总tps限速条件下，单个channel的tps值不能为空，也不能为非正数");
             }
 

dataxPluginDev.md

@@ -111,7 +111,7 @@ public class SomeReader extends Reader {
 ```
 
 `Job`接口功能如下：
-- `init`: Job对象初始化工作，测试可以通过`super.getPluginJobConf()`获取与本插件相关的配置。读插件获得配置中`reader`部分，写插件获得`writer`部分。
+- `init`: Job对象初始化工作，此时可以通过`super.getPluginJobConf()`获取与本插件相关的配置。读插件获得配置中`reader`部分，写插件获得`writer`部分。
 - `prepare`: 全局准备工作，比如odpswriter清空目标表。
 - `split`: 拆分`Task`。参数`adviceNumber`框架建议的拆分数，一般是运行时所配置的并发度。值返回的是`Task`的配置列表。
 - `post`: 全局的后置工作，比如mysqlwriter同步完影子表后的rename操作。
@@ -155,7 +155,7 @@ public class SomeReader extends Reader {
 ```
 
 - `name`: 插件名称，大小写敏感。框架根据用户在配置文件中指定的名称来搜寻插件。 **十分重要** 。
-- `class`: 入口类的全限定名称，框架通过反射穿件入口类的实例。**十分重要** 。
+- `class`: 入口类的全限定名称，框架通过反射插件入口类的实例。**十分重要** 。
 - `description`: 描述信息。
 - `developer`: 开发人员。
 
@@ -435,7 +435,7 @@ DataX的内部类型在实现上会选用不同的java类型：
 
 #### 如何处理脏数据
 
-在`Reader.Task`和`Writer.Task`中，功过`AbstractTaskPlugin.getPluginCollector()`可以拿到一个`TaskPluginCollector`，它提供了一系列`collectDirtyRecord`的方法。当脏数据出现时，只需要调用合适的`collectDirtyRecord`方法，把被认为是脏数据的`Record`传入即可。
+在`Reader.Task`和`Writer.Task`中，通过`AbstractTaskPlugin.getTaskPluginCollector()`可以拿到一个`TaskPluginCollector`，它提供了一系列`collectDirtyRecord`的方法。当脏数据出现时，只需要调用合适的`collectDirtyRecord`方法，把被认为是脏数据的`Record`传入即可。
 
 用户可以在任务的配置中指定脏数据限制条数或者百分比限制，当脏数据超出限制时，框架会结束同步任务，退出。插件需要保证脏数据都被收集到，其他工作交给框架就好。
 
@@ -468,4 +468,4 @@ DataX的内部类型在实现上会选用不同的java类型：
 	- 测试参数集（多组），系统参数（比如并发数），插件参数（比如batchSize）
 	- 不同参数下同步速度（Rec/s, MB/s），机器负载（load, cpu）等，对数据源压力（load, cpu, mem等）。
 6. **约束限制**：是否存在其他的使用限制条件。
-7. **FQA**：用户经常会遇到的问题。
+7. **FAQ**：用户经常会遇到的问题。

drdsreader/doc/drdsreader.md

@@ -50,7 +50,7 @@ DRDS的插件目前DataX只适配了Mysql引擎的场景，DRDS对于DataX而言
                         // 数据库连接密码
                         "password": "root",
                         "column": [
-                            "id"，"name"
+                            "id","name"
                         ],
                         "connection": [
                             {

drdsreader/pom.xml

@@ -42,7 +42,7 @@
         <dependency>
 		    <groupId>mysql</groupId>
 		    <artifactId>mysql-connector-java</artifactId>
-		    <version>5.1.34</version>
+		    <version>${mysql.driver.version}</version>
 		</dependency>
 
 

drdswriter/pom.xml

@@ -44,7 +44,7 @@
         <dependency>
             <groupId>mysql</groupId>
             <artifactId>mysql-connector-java</artifactId>
-            <version>5.1.34</version>
+            <version>${mysql.driver.version}</version>
         </dependency>
     </dependencies>
 

elasticsearchwriter/pom.xml

@@ -50,7 +50,7 @@
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
-            <version>4.11</version>
+            <version>4.13.1</version>
             <scope>test</scope>
         </dependency>
     </dependencies>

ftpwriter/doc/ftpwriter.md

@@ -63,6 +63,7 @@ FtpWriter实现了从DataX协议转为FTP文件功能，FTP文件本身是无结
                         "nullFormat": "null",
                         "dateFormat": "yyyy-MM-dd",
                         "fileFormat": "csv",
+			"suffix": ".csv",
                         "header": []
                     }
                 }
@@ -200,6 +201,14 @@ FtpWriter实现了从DataX协议转为FTP文件功能，FTP文件本身是无结
  	* 必选：否 <br />
  
  	* 默认值：text <br />
+	
+* **suffix**
+
+	* 描述：最后输出文件的后缀，当前支持 ".text"以及".csv"
+ 
+ 	* 必选：否 <br />
+ 
+ 	* 默认值："" <br />
 
 * **header**
 

gdbwriter/doc/gdbwriter.md

@@ -63,32 +63,32 @@ GDBWriter通过DataX框架获取Reader生成的协议数据，使用`g.addV/E(GD
                         "password": "***",
                         "writeMode": "INSERT",
                         "labelType": "VERTEX",
-                        "label": "${1}",
+                        "label": "#{1}",
                         "idTransRule": "none",
                         "session": true,
                         "maxRecordsInBatch": 64,
                         "column": [
                             {
                                 "name": "id",
-                                "value": "${0}",
+                                "value": "#{0}",
                                 "type": "string",
                                 "columnType": "primaryKey"
                             },
                             {
                                 "name": "vertex_propKey",
-                                "value": "${2}",
+                                "value": "#{2}",
                                 "type": "string",
                                 "columnType": "vertexSetProperty"
                             },
                             {
                                 "name": "vertex_propKey",
-                                "value": "${3}",
+                                "value": "#{3}",
                                 "type": "long",
                                 "columnType": "vertexSetProperty"
                             },
                             {
                                 "name": "vertex_propKey2",
-                                "value": "${4}",
+                                "value": "#{4}",
                                 "type": "string",
                                 "columnType": "vertexProperty"
                             }
@@ -154,7 +154,7 @@ GDBWriter通过DataX框架获取Reader生成的协议数据，使用`g.addV/E(GD
                         "password": "***",
                         "writeMode": "INSERT",
                         "labelType": "EDGE",
-                        "label": "${3}",
+                        "label": "#{3}",
                         "idTransRule": "none",
                         "srcIdTransRule": "labelPrefix",
                         "dstIdTransRule": "labelPrefix",
@@ -164,25 +164,25 @@ GDBWriter通过DataX框架获取Reader生成的协议数据，使用`g.addV/E(GD
                         "column": [
                             {
                                 "name": "id",
-                                "value": "${0}",
+                                "value": "#{0}",
                                 "type": "string",
                                 "columnType": "primaryKey"
                             },
                             {
                                 "name": "id",
-                                "value": "${1}",
+                                "value": "#{1}",
                                 "type": "string",
                                 "columnType": "srcPrimaryKey"
                             },
                             {
                                 "name": "id",
-                                "value": "${2}",
+                                "value": "#{2}",
                                 "type": "string",
                                 "columnType": "dstPrimaryKey"
                             },
                             {
                                 "name": "edge_propKey",
-                                "value": "${4}",
+                                "value": "#{4}",
                                 "type": "string",
                                 "columnType": "edgeProperty"
                             }
@@ -219,7 +219,7 @@ GDBWriter通过DataX框架获取Reader生成的协议数据，使用`g.addV/E(GD
   * 默认值：无
 
 * **label**
-  * 描述：类型名，即点/边名称； label支持从源列中读取，如${0}，表示取第一列字段作为label名。源列索引从0开始；
+  * 描述：类型名，即点/边名称； label支持从源列中读取，如#{0}，表示取第一列字段作为label名。源列索引从0开始；
   * 必选：是
   * 默认值：无
 
@@ -231,12 +231,12 @@ GDBWriter通过DataX框架获取Reader生成的协议数据，使用`g.addV/E(GD
   * 默认值：无
 
 * **srcLabel**
-  * 描述：当label为边时，表示起点的点名称；srcLabel支持从源列中读取，如${0}，表示取第一列字段作为label名。源列索引从0开始；
+  * 描述：当label为边时，表示起点的点名称；srcLabel支持从源列中读取，如#{0}，表示取第一列字段作为label名。源列索引从0开始；
   * 必选：labelType为边，srcIdTransRule为none时可不填写，否则必填；
   * 默认值：无
 
 * **dstLabel**
-  * 描述：当label为边时，表示终点的点名称；dstLabel支持从源列中读取，如${0}，表示取第一列字段作为label名。源列索引从0开始；
+  * 描述：当label为边时，表示终点的点名称；dstLabel支持从源列中读取，如#{0}，表示取第一列字段作为label名。源列索引从0开始；
   * 必选：labelType为边，dstIdTransRule为none时可不填写，否则必填；
   * 默认值：无
 
@@ -291,9 +291,9 @@ GDBWriter通过DataX框架获取Reader生成的协议数据，使用`g.addV/E(GD
 
 * **column -> value**
   * 描述：点/边映射关系的字段值；
-    * ${N}表示直接映射源端值，N为源端column索引，从0开始；${0}表示映射源端column第1个字段；
-    * test-${0} 表示源端值做拼接转换，${0}值前/后可添加固定字符串;
-    * ${0}-${1}表示做多字段拼接，也可在任意位置添加固定字符串，如test-${0}-test1-${1}-test2
+    * #{N}表示直接映射源端值，N为源端column索引，从0开始；#{0}表示映射源端column第1个字段；
+    * test-#{0} 表示源端值做拼接转换，#{0}值前/后可添加固定字符串;
+    * #{0}-#{1}表示做多字段拼接，也可在任意位置添加固定字符串，如test-#{0}-test1-#{1}-test2
   * 必选：是
   * 默认值：无
 

hbase11xsqlreader/src/main/java/com/alibaba/datax/plugin/reader/hbase11xsqlreader/HbaseSQLHelper.java

@@ -34,6 +34,14 @@ import java.util.Map;
 public class HbaseSQLHelper {
     private static final Logger LOG = LoggerFactory.getLogger(HbaseSQLHelper.class);
 
+    static {
+        try {
+            Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
+        } catch (Throwable t) {
+            throw new RuntimeException("faild load org.apache.phoenix.jdbc.PhoenixDriver", t);
+        }
+    }
+
     public static org.apache.hadoop.conf.Configuration generatePhoenixConf(HbaseSQLReaderConfig readerConfig) {
         org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
 

hbase11xsqlwriter/pom.xml

@@ -18,7 +18,7 @@
         <phoenix.version>4.11.0-HBase-1.1</phoenix.version>
         <hadoop.version>2.7.1</hadoop.version>
         <commons-codec.version>1.8</commons-codec.version>
-        <protobuf.version>3.2.0</protobuf.version>
+        <protobuf.version>3.16.1</protobuf.version>
         <httpclient.version>4.4.1</httpclient.version>
     </properties>
 

hbase11xwriter/doc/hbase11xwriter.md

@@ -203,19 +203,20 @@ HbaseWriter 插件实现了从向Hbase中写取数据。在底层实现上，Hba
 	* 描述：要写入的hbase字段。index：指定该列对应reader端column的索引，从0开始；name：指定hbase表中的列，必须为 列族:列名 的格式；type：指定写入数据类型，用于转换HBase byte[]。配置格式如下：
 	
 	```
-"column": [
-              {
-                "index":1,
-                "name": "cf1:q1",
-                "type": "string"
-              },
-              {
-                "index":2,
-                "name": "cf1:q2",
-                "type": "string"
-              }
-           ］ 	
-		            
+
+	"column": [
+		      {
+			"index":1,
+			"name": "cf1:q1",
+			"type": "string"
+		      },
+		      {
+			"index":2,
+			"name": "cf1:q2",
+			"type": "string"
+		      }
+		   ］ 	
+
 	```
 
 	* 必选：是<br />
@@ -227,17 +228,17 @@ HbaseWriter 插件实现了从向Hbase中写取数据。在底层实现上，Hba
 	* 描述：要写入的hbase的rowkey列。index：指定该列对应reader端column的索引，从0开始，若为常量index为－1；type：指定写入数据类型，用于转换HBase byte[]；value：配置常量，常作为多个字段的拼接符。hbasewriter会将rowkeyColumn中所有列按照配置顺序进行拼接作为写入hbase的rowkey，不能全为常量。配置格式如下：
 	
 	```
-"rowkeyColumn": [
-                {
-                  "index":0,
-                  "type":"string"
-                },
-                {
-                  "index":-1,
-                  "type":"string",
-                  "value":"_"
-                }
-            ] 	
+	"rowkeyColumn": [
+			{
+			  "index":0,
+			  "type":"string"
+			},
+			{
+			  "index":-1,
+			  "type":"string",
+			  "value":"_"
+			}
+		    ] 	
 		            
 	```
 
@@ -250,19 +251,19 @@ HbaseWriter 插件实现了从向Hbase中写取数据。在底层实现上，Hba
 	* 描述：指定写入hbase的时间戳。支持：当前时间、指定时间列，指定时间，三者选一。若不配置表示用当前时间。index：指定对应reader端column的索引，从0开始，需保证能转换为long,若是Date类型，会尝试用yyyy-MM-dd HH:mm:ss和yyyy-MM-dd HH:mm:ss SSS去解析；若为指定时间index为－1；value：指定时间的值,long值。配置格式如下：
 	
 	```
-"versionColumn":{
-	"index":1
-}
+	"versionColumn":{
+		"index":1
+	}
 		            
 	```
 	
 	或者
 	
 	```
-"versionColumn":{
-	"index":－1,
-	"value":123456789
-}
+	"versionColumn":{
+		"index":－1,
+		"value":123456789
+	}
 		            
 	```
 

hbase20xsqlreader/doc/hbase20xsqlreader.md

@@ -58,7 +58,9 @@ hbase20xsqlreader插件实现了从Phoenix(HBase SQL)读取数据，对应版本
 * **queryServerAddress**
 
 	* 描述：hbase20xsqlreader需要通过Phoenix轻客户端去连接Phoenix QueryServer，因此这里需要填写对应QueryServer地址。
- 
+           增强版/Lindorm 用户若需透传user, password参数，可以在queryServerAddress后增加对应可选属性.
+           格式参考：http://127.0.0.1:8765;user=root;password=root
+            
 	* 必选：是 <br />
  
 	* 默认值：无 <br />

hbase20xsqlreader/pom.xml

@@ -14,7 +14,7 @@
     <packaging>jar</packaging>
 
     <properties>
-        <phoenix.version>5.1.0-HBase-2.0.0.2</phoenix.version>
+        <phoenix.version>5.2.5-HBase-2.x</phoenix.version>
     </properties>
 
     <dependencies>

hbase20xsqlwriter/doc/hbase20xsqlwriter.md

@@ -120,7 +120,9 @@ HBase20xsqlwriter实现了向hbase中的SQL表(phoenix)批量导入数据的功
 
 * **queryServerAddress**
 
-   * 描述：Phoenix QueryServer地址，为必填项，格式：http://${hostName}:${ip}，如http://172.16.34.58:8765
+   * 描述：Phoenix QueryServer地址，为必填项，格式：http://${hostName}:${ip}，如http://172.16.34.58:8765。
+          增强版/Lindorm 用户若需透传user, password参数，可以在queryServerAddress后增加对应可选属性.
+              格式参考：http://127.0.0.1:8765;user=root;password=root
    * 必选：是
    * 默认值：无
    

hbase20xsqlwriter/pom.xml

@@ -14,7 +14,7 @@
     <packaging>jar</packaging>
 
     <properties>
-        <phoenix.version>5.1.0-HBase-2.0.0.2</phoenix.version>
+        <phoenix.version>5.2.5-HBase-2.x</phoenix.version>
         <commons-codec.version>1.8</commons-codec.version>
     </properties>
 

hbase20xsqlwriter/src/main/java/com/alibaba/datax/plugin/writer/hbase20xsqlwriter/HBase20xSQLWriterTask.java

@@ -6,12 +6,12 @@ import com.alibaba.datax.common.exception.DataXException;
 import com.alibaba.datax.common.plugin.RecordReceiver;
 import com.alibaba.datax.common.plugin.TaskPluginCollector;
 import com.alibaba.datax.common.util.Configuration;
-import com.google.common.collect.Lists;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.math.BigDecimal;
 import java.sql.*;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
@@ -154,7 +154,7 @@ public class HBase20xSQLWriterTask {
      * 从接收器中获取每条记录，写入Phoenix
      */
     private void writeData(RecordReceiver lineReceiver) throws SQLException {
-        List<Record> buffer = Lists.newArrayListWithExpectedSize(batchSize);
+        List<Record> buffer = new ArrayList<Record>(batchSize);
         Record record = null;
         while ((record = lineReceiver.getFromReader()) != null) {
             // 校验列数量是否符合预期

hdfswriter/src/main/java/com/alibaba/datax/plugin/writer/hdfswriter/HdfsWriter.java

@@ -81,10 +81,10 @@ public class HdfsWriter extends Writer {
             //writeMode check
             this.writeMode = this.writerSliceConfig.getNecessaryValue(Key.WRITE_MODE, HdfsWriterErrorCode.REQUIRED_VALUE);
             writeMode = writeMode.toLowerCase().trim();
-            Set<String> supportedWriteModes = Sets.newHashSet("append", "nonconflict");
+            Set<String> supportedWriteModes = Sets.newHashSet("append", "nonconflict", "truncate");
             if (!supportedWriteModes.contains(writeMode)) {
                 throw DataXException.asDataXException(HdfsWriterErrorCode.ILLEGAL_VALUE,
-                        String.format("仅支持append, nonConflict两种模式, 不支持您配置的 writeMode 模式 : [%s]",
+                        String.format("仅支持append, nonConflict, truncate三种模式, 不支持您配置的 writeMode 模式 : [%s]",
                                 writeMode));
             }
             this.writerSliceConfig.set(Key.WRITE_MODE, writeMode);
@@ -179,6 +179,9 @@ public class HdfsWriter extends Writer {
                     LOG.error(String.format("冲突文件列表为: [%s]", StringUtils.join(allFiles, ",")));
                     throw DataXException.asDataXException(HdfsWriterErrorCode.ILLEGAL_VALUE,
                             String.format("由于您配置了writeMode nonConflict,但您配置的path: [%s] 目录不为空, 下面存在其他文件或文件夹.", path));
+                }else if ("truncate".equalsIgnoreCase(writeMode) && isExistFile) {
+                    LOG.info(String.format("由于您配置了writeMode truncate,  [%s] 下面的内容将被覆盖重写", path));
+                    hdfsHelper.deleteFiles(existFilePaths);
                 }
             }else{
                 throw DataXException.asDataXException(HdfsWriterErrorCode.ILLEGAL_VALUE,

introduction.md

@@ -36,6 +36,7 @@ DataX本身作为离线数据同步框架，采用Framework + plugin架构构建
 | ------------ | ---------- | :-------: | :-------: |:-------: |
 | RDBMS 关系型数据库 | MySQL      |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/mysqlreader/doc/mysqlreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/mysqlwriter/doc/mysqlwriter.md)|
 |              | Oracle     |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/oraclereader/doc/oraclereader.md) 、[写](https://github.com/alibaba/DataX/blob/master/oraclewriter/doc/oraclewriter.md)|
+|              | OceanBase  |     √     |     √     |[读](https://open.oceanbase.com/docs/community/oceanbase-database/V3.1.0/use-datax-to-full-migration-data-to-oceanbase) 、[写](https://open.oceanbase.com/docs/community/oceanbase-database/V3.1.0/use-datax-to-full-migration-data-to-oceanbase)|
 |              | SQLServer  |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/sqlserverreader/doc/sqlserverreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/sqlserverwriter/doc/sqlserverwriter.md)|
 |              | PostgreSQL |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/postgresqlreader/doc/postgresqlreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/postgresqlwriter/doc/postgresqlwriter.md)|
 |              | DRDS |     √     |     √     |[读](https://github.com/alibaba/DataX/blob/master/drdsreader/doc/drdsreader.md) 、[写](https://github.com/alibaba/DataX/blob/master/drdswriter/doc/drdswriter.md)|

kingbaseesreader/doc/kingbaseesreader.md

@@ -0,0 +1,241 @@
+
+# KingbaseesReader 插件文档
+
+
+___
+
+
+## 1 快速介绍
+
+KingbaseesReader插件实现了从KingbaseES读取数据。在底层实现上，KingbaseesReader通过JDBC连接远程KingbaseES数据库，并执行相应的sql语句将数据从KingbaseES库中SELECT出来。
+
+## 2 实现原理
+
+简而言之，KingbaseesReader通过JDBC连接器连接到远程的KingbaseES数据库，并根据用户配置的信息生成查询SELECT SQL语句并发送到远程KingbaseES数据库，并将该SQL执行返回结果使用DataX自定义的数据类型拼装为抽象的数据集，并传递给下游Writer处理。
+
+对于用户配置Table、Column、Where的信息，KingbaseesReader将其拼接为SQL语句发送到KingbaseES数据库；对于用户配置querySql信息，KingbaseesReader直接将其发送到KingbaseES数据库。
+
+
+## 3 功能说明
+
+### 3.1 配置样例
+
+* 配置一个从KingbaseES数据库同步抽取数据到本地的作业:
+
+```
+{
+    "job": {
+        "setting": {
+            "speed": {
+            //设置传输速度，单位为byte/s，DataX运行会尽可能达到该速度但是不超过它.
+                 "byte": 1048576
+            },
+            //出错限制
+                "errorLimit": {
+                //出错的record条数上限，当大于该值即报错。
+                "record": 0,
+                //出错的record百分比上限 1.0表示100%，0.02表示2%
+                "percentage": 0.02
+            }
+        },
+        "content": [
+            {
+                "reader": {
+                    "name": "kingbaseesreader",
+                    "parameter": {
+                        // 数据库连接用户名
+                        "username": "xx",
+                        // 数据库连接密码
+                        "password": "xx",
+                        "column": [
+                            "id"，"name"
+                        ],
+                        //切分主键
+                        "splitPk": "id",
+                        "connection": [
+                            {
+                                "table": [
+                                    "table"
+                                ],
+                                "jdbcUrl": [
+     "jdbc:kingbase8://host:port/database"
+                                ]
+                            }
+                        ]
+                    }
+                },
+               "writer": {
+                    //writer类型
+                    "name": "streamwriter",
+                    //是否打印内容
+                    "parameter": {
+                        "print":true,
+                    }
+                }
+            }
+        ]
+    }
+}
+
+```
+
+* 配置一个自定义SQL的数据库同步任务到本地内容的作业：
+
+```
+{
+    "job": {
+        "setting": {
+            "speed": 1048576
+        },
+        "content": [
+            {
+                "reader": {
+                    "name": "kingbaseesreader",
+                    "parameter": {
+                        "username": "xx",
+                        "password": "xx",
+                        "where": "",
+                        "connection": [
+                            {
+                                "querySql": [
+                                    "select db_id,on_line_flag from db_info where db_id < 10;"
+                                ],
+                                "jdbcUrl": [
+                                    "jdbc:kingbase8://host:port/database", "jdbc:kingbase8://host:port/database"
+                                ]
+                            }
+                        ]
+                    }
+                },
+                "writer": {
+                    "name": "streamwriter",
+                    "parameter": {
+                        "print": false,
+                        "encoding": "UTF-8"
+                    }
+                }
+            }
+        ]
+    }
+}
+```
+
+
+### 3.2 参数说明
+
+* **jdbcUrl**
+
+	* 描述：描述的是到对端数据库的JDBC连接信息，使用JSON的数组描述，并支持一个库填写多个连接地址。之所以使用JSON数组描述连接信息，是因为阿里集团内部支持多个IP探测，如果配置了多个，KingbaseesReader可以依次探测ip的可连接性，直到选择一个合法的IP。如果全部连接失败，KingbaseesReader报错。 注意，jdbcUrl必须包含在connection配置单元中。对于阿里集团外部使用情况，JSON数组填写一个JDBC连接即可。
+
+		jdbcUrl按照KingbaseES官方规范，并可以填写连接附件控制信息。具体请参看[KingbaseES官方文档](https://help.kingbase.com.cn/doc-view-5683.html)。
+
+	* 必选：是 <br />
+
+	* 默认值：无 <br />
+
+* **username**
+
+	* 描述：数据源的用户名 <br />
+
+	* 必选：是 <br />
+
+	* 默认值：无 <br />
+
+* **password**
+
+	* 描述：数据源指定用户名的密码 <br />
+
+	* 必选：是 <br />
+
+	* 默认值：无 <br />
+
+* **table**
+
+	* 描述：所选取的需要同步的表。使用JSON的数组描述，因此支持多张表同时抽取。当配置为多张表时，用户自己需保证多张表是同一schema结构，KingbaseesReader不予检查表是否同一逻辑表。注意，table必须包含在connection配置单元中。<br />
+
+	* 必选：是 <br />
+
+	* 默认值：无 <br />
+
+* **column**
+
+	* 描述：所配置的表中需要同步的列名集合，使用JSON的数组描述字段信息。用户使用\*代表默认使用所有列配置，例如['\*']。
+
+	  支持列裁剪，即列可以挑选部分列进行导出。
+
+      支持列换序，即列可以不按照表schema信息进行导出。
+
+	  支持常量配置，用户需要按照KingbaseES语法格式:
+	  ["id", "'hello'::varchar", "true", "2.5::real", "power(2,3)"]
+	  id为普通列名，'hello'::varchar为字符串常量，true为布尔值，2.5为浮点数, power(2,3)为函数。
+
+		**column必须用户显示指定同步的列集合，不允许为空！**
+
+	* 必选：是 <br />
+
+	* 默认值：无 <br />
+
+* **splitPk**
+
+	* 描述：KingbaseesReader进行数据抽取时，如果指定splitPk，表示用户希望使用splitPk代表的字段进行数据分片，DataX因此会启动并发任务进行数据同步，这样可以大大提供数据同步的效能。
+
+	  推荐splitPk用户使用表主键，因为表主键通常情况下比较均匀，因此切分出来的分片也不容易出现数据热点。
+
+	  目前splitPk仅支持整形数据切分，`不支持浮点、字符串型、日期等其他类型`。如果用户指定其他非支持类型，KingbaseesReader将报错！
+
+	 	splitPk设置为空，底层将视作用户不允许对单表进行切分，因此使用单通道进行抽取。
+
+	* 必选：否 <br />
+
+	* 默认值：空 <br />
+
+* **where**
+
+	* 描述：筛选条件，KingbaseesReader根据指定的column、table、where条件拼接SQL，并根据这个SQL进行数据抽取。在实际业务场景中，往往会选择当天的数据进行同步，可以将where条件指定为gmt_create > $bizdate 。注意：不可以将where条件指定为limit 10，limit不是SQL的合法where子句。<br />
+
+          where条件可以有效地进行业务增量同步。		where条件不配置或者为空，视作全表同步数据。
+
+	* 必选：否 <br />
+
+	* 默认值：无 <br />
+
+* **querySql**
+
+	* 描述：在有些业务场景下，where这一配置项不足以描述所筛选的条件，用户可以通过该配置型来自定义筛选SQL。当用户配置了这一项之后，DataX系统就会忽略table，column这些配置型，直接使用这个配置项的内容对数据进行筛选，例如需要进行多表join后同步数据，使用select a,b from table_a join table_b on table_a.id = table_b.id <br />
+
+	 `当用户配置querySql时，KingbaseesReader直接忽略table、column、where条件的配置`。
+
+	* 必选：否 <br />
+
+	* 默认值：无 <br />
+
+* **fetchSize**
+
+	* 描述：该配置项定义了插件和数据库服务器端每次批量数据获取条数，该值决定了DataX和服务器端的网络交互次数，能够较大的提升数据抽取性能。<br />
+
+	 `注意，该值过大(>2048)可能造成DataX进程OOM。`。
+
+	* 必选：否 <br />
+
+	* 默认值：1024 <br />
+
+
+### 3.3 类型转换
+
+目前KingbaseesReader支持大部分KingbaseES类型，但也存在部分个别类型没有支持的情况，请注意检查你的类型。
+
+下面列出KingbaseesReader针对KingbaseES类型转换列表:
+
+
+| DataX 内部类型| KingbaseES 数据类型    |
+| -------- | -----  |
+| Long     |bigint, bigserial, integer, smallint, serial |
+| Double   |double precision, money, numeric, real |
+| String   |varchar, char, text, bit, inet|
+| Date     |date, time, timestamp |
+| Boolean  |bool|
+| Bytes    |bytea|
+
+请注意:
+
+* `除上述罗列字段类型外，其他类型均不支持; money,inet,bit需用户使用a_inet::varchar类似的语法转换`。

kingbaseesreader/pom.xml

@@ -0,0 +1,88 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>datax-all</artifactId>
+        <groupId>com.alibaba.datax</groupId>
+        <version>0.0.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>kingbaseesreader</artifactId>
+    <name>kingbaseesreader</name>
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.alibaba.datax</groupId>
+            <artifactId>datax-common</artifactId>
+            <version>${datax-project-version}</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>slf4j-log4j12</artifactId>
+                    <groupId>org.slf4j</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+        
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.alibaba.datax</groupId>
+            <artifactId>plugin-rdbms-util</artifactId>
+            <version>${datax-project-version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.kingbase8</groupId>
+            <artifactId>kingbase8</artifactId>
+            <version>8.2.0</version>
+            <scope>system</scope>
+            <systemPath>${basedir}/src/main/libs/kingbase8-8.2.0.jar</systemPath>
+        </dependency>
+
+    </dependencies>
+
+    <build>
+        <plugins>
+            <!-- compiler plugin -->
+            <plugin>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>${jdk-version}</source>
+                    <target>${jdk-version}</target>
+                    <encoding>${project-sourceEncoding}</encoding>
+                </configuration>
+            </plugin>
+            <!-- assembly plugin -->
+            <plugin>
+                <artifactId>maven-assembly-plugin</artifactId>
+                <configuration>
+                    <descriptors>
+                        <descriptor>src/main/assembly/package.xml</descriptor>
+                    </descriptors>
+                    <finalName>datax</finalName>
+                </configuration>
+                <executions>
+                    <execution>
+                        <id>dwzip</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>single</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

kingbaseesreader/src/main/assembly/package.xml

@@ -0,0 +1,42 @@
+<assembly
+        xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
+        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+        xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
+    <id></id>
+    <formats>
+        <format>dir</format>
+    </formats>
+    <includeBaseDirectory>false</includeBaseDirectory>
+    <fileSets>
+        <fileSet>
+            <directory>src/main/resources</directory>
+            <includes>
+                <include>plugin.json</include>
+                <include>plugin_job_template.json</include>
+            </includes>
+            <outputDirectory>plugin/reader/kingbaseesreader</outputDirectory>
+        </fileSet>
+        <fileSet>
+            <directory>target/</directory>
+            <includes>
+                <include>kingbaseesreader-0.0.1-SNAPSHOT.jar</include>
+            </includes>
+            <outputDirectory>plugin/reader/kingbaseesreader</outputDirectory>
+        </fileSet>
+        <fileSet>
+            <directory>src/main/libs</directory>
+            <includes>
+                <include>*.*</include>
+            </includes>
+            <outputDirectory>plugin/reader/kingbaseesreader/libs</outputDirectory>
+        </fileSet>
+    </fileSets>
+
+    <dependencySets>
+        <dependencySet>
+            <useProjectArtifact>false</useProjectArtifact>
+            <outputDirectory>plugin/reader/kingbaseesreader/libs</outputDirectory>
+            <scope>runtime</scope>
+        </dependencySet>
+    </dependencySets>
+</assembly>

kingbaseesreader/src/main/java/com/alibaba/datax/plugin/reader/kingbaseesreader/Constant.java

@@ -0,0 +1,7 @@
+package com.alibaba.datax.plugin.reader.kingbaseesreader;
+
+public class Constant {
+
+    public static final int DEFAULT_FETCH_SIZE = 1000;
+
+}

kingbaseesreader/src/main/java/com/alibaba/datax/plugin/reader/kingbaseesreader/KingbaseesReader.java

@@ -0,0 +1,86 @@
+package com.alibaba.datax.plugin.reader.kingbaseesreader;
+
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.common.plugin.RecordSender;
+import com.alibaba.datax.common.spi.Reader;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.reader.CommonRdbmsReader;
+import com.alibaba.datax.plugin.rdbms.util.DBUtilErrorCode;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+
+import java.util.List;
+
+public class KingbaseesReader extends Reader {
+
+    private static final DataBaseType DATABASE_TYPE = DataBaseType.KingbaseES;
+
+    public static class Job extends Reader.Job {
+
+        private Configuration originalConfig;
+        private CommonRdbmsReader.Job commonRdbmsReaderMaster;
+
+        @Override
+        public void init() {
+            this.originalConfig = super.getPluginJobConf();
+            int fetchSize = this.originalConfig.getInt(com.alibaba.datax.plugin.rdbms.reader.Constant.FETCH_SIZE,
+                    Constant.DEFAULT_FETCH_SIZE);
+            if (fetchSize < 1) {
+            	throw DataXException.asDataXException(DBUtilErrorCode.REQUIRED_VALUE,
+					String.format("您配置的fetchSize有误，根据DataX的设计，fetchSize : [%d] 设置值不能小于 1.", fetchSize));
+            }
+            this.originalConfig.set(com.alibaba.datax.plugin.rdbms.reader.Constant.FETCH_SIZE, fetchSize);
+
+            this.commonRdbmsReaderMaster = new CommonRdbmsReader.Job(DATABASE_TYPE);
+            this.commonRdbmsReaderMaster.init(this.originalConfig);
+        }
+
+        @Override
+        public List<Configuration> split(int adviceNumber) {
+            return this.commonRdbmsReaderMaster.split(this.originalConfig, adviceNumber);
+        }
+
+        @Override
+        public void post() {
+            this.commonRdbmsReaderMaster.post(this.originalConfig);
+        }
+
+        @Override
+        public void destroy() {
+            this.commonRdbmsReaderMaster.destroy(this.originalConfig);
+        }
+
+    }
+
+    public static class Task extends Reader.Task {
+
+        private Configuration readerSliceConfig;
+        private CommonRdbmsReader.Task commonRdbmsReaderSlave;
+
+        @Override
+        public void init() {
+            this.readerSliceConfig = super.getPluginJobConf();
+            this.commonRdbmsReaderSlave = new CommonRdbmsReader.Task(DATABASE_TYPE, super.getTaskGroupId(), super.getTaskId());
+            this.commonRdbmsReaderSlave.init(this.readerSliceConfig);
+        }
+
+        @Override
+        public void startRead(RecordSender recordSender) {
+            int fetchSize = this.readerSliceConfig.getInt(com.alibaba.datax.plugin.rdbms.reader.Constant.FETCH_SIZE);
+
+            this.commonRdbmsReaderSlave.startRead(this.readerSliceConfig, recordSender,
+                    super.getTaskPluginCollector(), fetchSize);
+        }
+
+        @Override
+        public void post() {
+            this.commonRdbmsReaderSlave.post(this.readerSliceConfig);
+        }
+
+        @Override
+        public void destroy() {
+            this.commonRdbmsReaderSlave.destroy(this.readerSliceConfig);
+        }
+
+    }
+
+}

kingbaseesreader/src/main/resources/plugin.json

@@ -0,0 +1,6 @@
+{
+    "name": "kingbaseesreader",
+    "class": "com.alibaba.datax.plugin.reader.kingbaseesreader.KingbaseesReader",
+    "description": "useScene: prod. mechanism: Jdbc connection using the database, execute select sql, retrieve data from the ResultSet. warn: The more you know about the database, the less problems you encounter.",
+    "developer": "alibaba"
+}

kingbaseesreader/src/main/resources/plugin_job_template.json

@@ -0,0 +1,13 @@
+{
+    "name": "kingbaseesreader",
+    "parameter": {
+        "username": "",
+        "password": "",
+        "connection": [
+            {
+                "table": [],
+                "jdbcUrl": []
+            }
+        ]
+    }
+}

kingbaseeswriter/doc/kingbaseeswriter.md

@@ -0,0 +1,208 @@
+# DataX KingbaseesWriter
+
+
+---
+
+
+## 1 快速介绍
+
+KingbaseesWriter插件实现了写入数据到 KingbaseES主库目的表的功能。在底层实现上，KingbaseesWriter通过JDBC连接远程 KingbaseES 数据库，并执行相应的 insert into ... sql 语句将数据写入 KingbaseES，内部会分批次提交入库。
+
+KingbaseesWriter面向ETL开发工程师，他们使用KingbaseesWriter从数仓导入数据到KingbaseES。同时 KingbaseesWriter亦可以作为数据迁移工具为DBA等用户提供服务。
+
+
+## 2 实现原理
+
+KingbaseesWriter通过 DataX 框架获取 Reader 生成的协议数据，根据你配置生成相应的SQL插入语句
+
+
+* `insert into...`(当主键/唯一性索引冲突时会写不进去冲突的行)
+
+<br />
+
+    注意：
+    1. 目的表所在数据库必须是主库才能写入数据；整个任务至少需具备 insert into...的权限，是否需要其他权限，取决于你任务配置中在 preSql 和 postSql 中指定的语句。
+    2. KingbaseesWriter和MysqlWriter不同，不支持配置writeMode参数。
+
+
+## 3 功能说明
+
+### 3.1 配置样例
+
+* 这里使用一份从内存产生到 KingbaseesWriter导入的数据。
+
+```json
+{
+    "job": {
+        "setting": {
+            "speed": {
+                "channel": 1
+            }
+        },
+        "content": [
+            {
+                 "reader": {
+                    "name": "streamreader",
+                    "parameter": {
+                        "column" : [
+                            {
+                                "value": "DataX",
+                                "type": "string"
+                            },
+                            {
+                                "value": 19880808,
+                                "type": "long"
+                            },
+                            {
+                                "value": "1988-08-08 08:08:08",
+                                "type": "date"
+                            },
+                            {
+                                "value": true,
+                                "type": "bool"
+                            },
+                            {
+                                "value": "test",
+                                "type": "bytes"
+                            }
+                        ],
+                        "sliceRecordCount": 1000
+                    }
+                },
+                "writer": {
+                    "name": "kingbaseeswriter",
+                    "parameter": {
+                        "username": "xx",
+                        "password": "xx",
+                        "column": [
+                            "id",
+                            "name"
+                        ],
+                        "preSql": [
+                            "delete from test"
+                        ],
+                        "connection": [
+                            {
+                                "jdbcUrl": "jdbc:kingbase8://127.0.0.1:3002/datax",
+                                "table": [
+                                    "test"
+                                ]
+                            }
+                        ]
+                    }
+                }
+            }
+        ]
+    }
+}
+
+```
+
+
+### 3.2 参数说明
+
+* **jdbcUrl**
+
+    * 描述：目的数据库的 JDBC 连接信息 ,jdbcUrl必须包含在connection配置单元中。
+
+      注意：1、在一个数据库上只能配置一个值。
+      2、jdbcUrl按照KingbaseES官方规范，并可以填写连接附加参数信息。具体请参看KingbaseES官方文档或者咨询对应 DBA。
+
+
+  * 必选：是 <br />
+
+  * 默认值：无 <br />
+
+* **username**
+
+  * 描述：目的数据库的用户名 <br />
+
+  * 必选：是 <br />
+
+  * 默认值：无 <br />
+
+* **password**
+
+  * 描述：目的数据库的密码 <br />
+
+  * 必选：是 <br />
+
+  * 默认值：无 <br />
+
+* **table**
+
+  * 描述：目的表的表名称。支持写入一个或者多个表。当配置为多张表时，必须确保所有表结构保持一致。
+
+               注意：table 和 jdbcUrl 必须包含在 connection 配置单元中
+
+  * 必选：是 <br />
+
+  * 默认值：无 <br />
+
+* **column**
+
+  * 描述：目的表需要写入数据的字段,字段之间用英文逗号分隔。例如: "column": ["id","name","age"]。如果要依次写入全部列，使用\*表示, 例如: "column": ["\*"]
+
+               注意：1、我们强烈不推荐你这样配置，因为当你目的表字段个数、类型等有改动时，你的任务可能运行不正确或者失败
+                    2、此处 column 不能配置任何常量值
+
+  * 必选：是 <br />
+
+  * 默认值：否 <br />
+
+* **preSql**
+
+  * 描述：写入数据到目的表前，会先执行这里的标准语句。如果 Sql 中有你需要操作到的表名称，请使用 `@table` 表示，这样在实际执行 Sql 语句时，会对变量按照实际表名称进行替换。比如你的任务是要写入到目的端的100个同构分表(表名称为:datax_00,datax01, ... datax_98,datax_99)，并且你希望导入数据前，先对表中数据进行删除操作，那么你可以这样配置：`"preSql":["delete from @table"]`，效果是：在执行到每个表写入数据前，会先执行对应的 delete from 对应表名称 <br />
+
+  * 必选：否 <br />
+
+  * 默认值：无 <br />
+
+* **postSql**
+
+  * 描述：写入数据到目的表后，会执行这里的标准语句。（原理同 preSql ） <br />
+
+  * 必选：否 <br />
+
+  * 默认值：无 <br />
+
+* **batchSize**
+
+	* 描述：一次性批量提交的记录数大小，该值可以极大减少DataX与KingbaseES的网络交互次数，并提升整体吞吐量。但是该值设置过大可能会造成DataX运行进程OOM情况。<br />
+
+	* 必选：否 <br />
+
+	* 默认值：1024 <br />
+
+### 3.3 类型转换
+
+目前 KingbaseesWriter支持大部分 KingbaseES类型，但也存在部分没有支持的情况，请注意检查你的类型。
+
+下面列出 KingbaseesWriter针对 KingbaseES类型转换列表:
+
+| DataX 内部类型| KingbaseES 数据类型    |
+| -------- | -----  |
+| Long     |bigint, bigserial, integer, smallint, serial |
+| Double   |double precision, money, numeric, real |
+| String   |varchar, char, text, bit|
+| Date     |date, time, timestamp |
+| Boolean  |bool|
+| Bytes    |bytea|
+
+
+## FAQ
+
+***
+
+**Q: KingbaseesWriter 执行 postSql 语句报错，那么数据导入到目标数据库了吗?**
+
+A: DataX 导入过程存在三块逻辑，pre 操作、导入操作、post 操作，其中任意一环报错，DataX 作业报错。由于 DataX 不能保证在同一个事务完成上述几个操作，因此有可能数据已经落入到目标端。
+
+***
+
+**Q: 按照上述说法，那么有部分脏数据导入数据库，如果影响到线上数据库怎么办?**
+
+A: 目前有两种解法，第一种配置 pre 语句，该 sql 可以清理当天导入数据， DataX 每次导入时候可以把上次清理干净并导入完整数据。
+第二种，向临时表导入数据，完成后再 rename 到线上表。
+
+***

kingbaseeswriter/pom.xml

@@ -0,0 +1,84 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>com.alibaba.datax</groupId>
+		<artifactId>datax-all</artifactId>
+		<version>0.0.1-SNAPSHOT</version>
+	</parent>
+	<artifactId>kingbaseeswriter</artifactId>
+	<name>kingbaseeswriter</name>
+	<packaging>jar</packaging>
+	<description>writer data into kingbasees database</description>
+
+	<dependencies>
+		<dependency>
+			<groupId>com.alibaba.datax</groupId>
+			<artifactId>datax-common</artifactId>
+			<version>${datax-project-version}</version>
+			<exclusions>
+				<exclusion>
+					<artifactId>slf4j-log4j12</artifactId>
+					<groupId>org.slf4j</groupId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		
+		<dependency>
+			<groupId>org.slf4j</groupId>
+			<artifactId>slf4j-api</artifactId>
+		</dependency>
+		
+		<dependency>
+			<groupId>ch.qos.logback</groupId>
+			<artifactId>logback-classic</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>com.alibaba.datax</groupId>
+			<artifactId>plugin-rdbms-util</artifactId>
+			<version>${datax-project-version}</version>
+		</dependency>
+
+		<dependency>
+			<groupId>com.kingbase8</groupId>
+			<artifactId>kingbase8</artifactId>
+			<version>8.2.0</version>
+			<scope>system</scope>
+			<systemPath>${basedir}/src/main/libs/kingbase8-8.2.0.jar</systemPath>
+		</dependency>
+
+	</dependencies>
+	<build>
+		<plugins>
+			<!-- compiler plugin -->
+			<plugin>
+				<artifactId>maven-compiler-plugin</artifactId>
+				<configuration>
+					<source>${jdk-version}</source>
+					<target>${jdk-version}</target>
+					<encoding>${project-sourceEncoding}</encoding>
+				</configuration>
+			</plugin>
+			<!-- assembly plugin -->
+			<plugin>
+				<artifactId>maven-assembly-plugin</artifactId>
+				<configuration>
+					<descriptors>
+						<descriptor>src/main/assembly/package.xml</descriptor>
+					</descriptors>
+					<finalName>datax</finalName>
+				</configuration>
+				<executions>
+					<execution>
+						<id>dwzip</id>
+						<phase>package</phase>
+						<goals>
+							<goal>single</goal>
+						</goals>
+					</execution>
+				</executions>
+			</plugin>
+		</plugins>
+	</build>
+</project>

kingbaseeswriter/src/main/assembly/package.xml

@@ -0,0 +1,42 @@
+<assembly
+        xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
+        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+        xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
+    <id></id>
+    <formats>
+        <format>dir</format>
+    </formats>
+    <includeBaseDirectory>false</includeBaseDirectory>
+    <fileSets>
+        <fileSet>
+            <directory>src/main/resources</directory>
+            <includes>
+                <include>plugin.json</include>
+                <include>plugin_job_template.json</include>
+            </includes>
+            <outputDirectory>plugin/writer/kingbaseeswriter</outputDirectory>
+        </fileSet>
+        <fileSet>
+            <directory>target/</directory>
+            <includes>
+                <include>kingbaseeswriter-0.0.1-SNAPSHOT.jar</include>
+            </includes>
+            <outputDirectory>plugin/writer/kingbaseeswriter</outputDirectory>
+        </fileSet>
+        <fileSet>
+            <directory>src/main/libs</directory>
+            <includes>
+                <include>*.*</include>
+            </includes>
+            <outputDirectory>plugin/writer/kingbaseeswriter/libs</outputDirectory>
+        </fileSet>
+    </fileSets>
+
+    <dependencySets>
+        <dependencySet>
+            <useProjectArtifact>false</useProjectArtifact>
+            <outputDirectory>plugin/writer/kingbaseeswriter/libs</outputDirectory>
+            <scope>runtime</scope>
+        </dependencySet>
+    </dependencySets>
+</assembly>

kingbaseeswriter/src/main/java/com/alibaba/datax/plugin/writer/kingbaseeswriter/KingbaseesWriter.java

@@ -0,0 +1,100 @@
+package com.alibaba.datax.plugin.writer.kingbaseeswriter;
+
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.common.plugin.RecordReceiver;
+import com.alibaba.datax.common.spi.Writer;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.util.DBUtilErrorCode;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.datax.plugin.rdbms.writer.CommonRdbmsWriter;
+import com.alibaba.datax.plugin.rdbms.writer.Key;
+
+import java.util.List;
+
+public class KingbaseesWriter extends Writer {
+	private static final DataBaseType DATABASE_TYPE = DataBaseType.KingbaseES;
+
+	public static class Job extends Writer.Job {
+		private Configuration originalConfig = null;
+		private CommonRdbmsWriter.Job commonRdbmsWriterMaster;
+
+		@Override
+		public void init() {
+			this.originalConfig = super.getPluginJobConf();
+
+			// warn：not like mysql, KingbaseES only support insert mode, don't use
+			String writeMode = this.originalConfig.getString(Key.WRITE_MODE);
+			if (null != writeMode) {
+				throw DataXException.asDataXException(DBUtilErrorCode.CONF_ERROR,
+					String.format("写入模式(writeMode)配置有误. 因为KingbaseES不支持配置参数项 writeMode: %s, KingbaseES仅使用insert sql 插入数据. 请检查您的配置并作出修改.", writeMode));
+			}
+
+			this.commonRdbmsWriterMaster = new CommonRdbmsWriter.Job(DATABASE_TYPE);
+			this.commonRdbmsWriterMaster.init(this.originalConfig);
+		}
+
+		@Override
+		public void prepare() {
+			this.commonRdbmsWriterMaster.prepare(this.originalConfig);
+		}
+
+		@Override
+		public List<Configuration> split(int mandatoryNumber) {
+			return this.commonRdbmsWriterMaster.split(this.originalConfig, mandatoryNumber);
+		}
+
+		@Override
+		public void post() {
+			this.commonRdbmsWriterMaster.post(this.originalConfig);
+		}
+
+		@Override
+		public void destroy() {
+			this.commonRdbmsWriterMaster.destroy(this.originalConfig);
+		}
+
+	}
+
+	public static class Task extends Writer.Task {
+		private Configuration writerSliceConfig;
+		private CommonRdbmsWriter.Task commonRdbmsWriterSlave;
+
+		@Override
+		public void init() {
+			this.writerSliceConfig = super.getPluginJobConf();
+			this.commonRdbmsWriterSlave = new CommonRdbmsWriter.Task(DATABASE_TYPE){
+				@Override
+				public String calcValueHolder(String columnType){
+					if("serial".equalsIgnoreCase(columnType)){
+						return "?::int";
+					}else if("bit".equalsIgnoreCase(columnType)){
+						return "?::bit varying";
+					}
+					return "?::" + columnType;
+				}
+			};
+			this.commonRdbmsWriterSlave.init(this.writerSliceConfig);
+		}
+
+		@Override
+		public void prepare() {
+			this.commonRdbmsWriterSlave.prepare(this.writerSliceConfig);
+		}
+
+		public void startWrite(RecordReceiver recordReceiver) {
+			this.commonRdbmsWriterSlave.startWrite(recordReceiver, this.writerSliceConfig, super.getTaskPluginCollector());
+		}
+
+		@Override
+		public void post() {
+			this.commonRdbmsWriterSlave.post(this.writerSliceConfig);
+		}
+
+		@Override
+		public void destroy() {
+			this.commonRdbmsWriterSlave.destroy(this.writerSliceConfig);
+		}
+
+	}
+
+}

kingbaseeswriter/src/main/resources/plugin.json

@@ -0,0 +1,6 @@
+{
+    "name": "kingbaseeswriter",
+    "class": "com.alibaba.datax.plugin.writer.kingbaseeswriter.KingbaseesWriter",
+    "description": "useScene: prod. mechanism: Jdbc connection using the database, execute insert sql. warn: The more you know about the database, the less problems you encounter.",
+    "developer": "alibaba"
+}

kingbaseeswriter/src/main/resources/plugin_job_template.json

@@ -0,0 +1,17 @@
+{
+    "name": "kingbaseeswriter",
+    "parameter": {
+        "username": "",
+        "password": "",
+        "column": [],
+        "preSql": [],
+        "connection": [
+            {
+                "jdbcUrl": "",
+                "table": []
+            }
+        ],
+        "preSql": [],
+        "postSql": []
+    }
+}

kuduwriter/README.md

@@ -0,0 +1,6 @@
+# datax-kudu-plugin
+datax kudu的writer插件
+
+
+
+仅在kudu11进行过测试

kuduwriter/doc/kuduwirter.md

@@ -0,0 +1,143 @@
+# datax-kudu-plugins
+datax kudu的writer插件
+
+
+
+eg:
+
+```json
+{
+  "name": "kuduwriter",
+  "parameter": {
+    "kuduConfig": {
+      "kudu.master_addresses": "***",
+      "timeout": 60000,
+      "sessionTimeout": 60000
+
+    },
+    "table": "",
+    "replicaCount": 3,
+    "truncate": false,
+    "writeMode": "upsert",
+    "partition": {
+      "range": {
+        "column1": [
+          {
+            "lower": "2020-08-25",
+            "upper": "2020-08-26"
+          },
+          {
+            "lower": "2020-08-26",
+            "upper": "2020-08-27"
+          },
+          {
+            "lower": "2020-08-27",
+            "upper": "2020-08-28"
+          }
+        ]
+      },
+      "hash": {
+        "column": [
+          "column1"
+        ],
+        "number": 3
+      }
+    },
+    "column": [
+      {
+        "index": 0,
+        "name": "c1",
+        "type": "string",
+        "primaryKey": true
+      },
+      {
+        "index": 1,
+        "name": "c2",
+        "type": "string",
+        "compress": "DEFAULT_COMPRESSION",
+        "encoding": "AUTO_ENCODING",
+        "comment": "注解xxxx"
+      }
+    ],
+    "batchSize": 1024,
+    "bufferSize": 2048,
+    "skipFail": false,
+    "encoding": "UTF-8"
+  }
+}
+```
+
+必须参数：
+
+```json
+        "writer": {
+          "name": "kuduwriter",
+          "parameter": {
+            "kuduConfig": {
+              "kudu.master_addresses": "***"
+            },
+            "table": "***",
+            "column": [
+              {
+                "name": "c1",
+                "type": "string",
+                "primaryKey": true
+              },
+              {
+                "name": "c2",
+                "type": "string",
+              },
+              {
+                "name": "c3",
+                "type": "string"
+              },
+              {
+                "name": "c4",
+                "type": "string"
+              }
+            ]
+          }
+        }
+```
+
+主键列请写到最前面
+
+
+
+![image-20200901193148188](./image-20200901193148188.png)
+
+##### 配置列表
+
+| name           | default             | description                                                  | 是否必须 |
+| -------------- | ------------------- | ------------------------------------------------------------ | -------- |
+| kuduConfig     |                     | kudu配置 （kudu.master_addresses等）                         | 是       |
+| table          |                     | 导入目标表名                                                 | 是       |
+| partition      |                     | 分区                                                         | 否       |
+| column         |                     | 列                                                           | 是       |
+| name           |                     | 列名                                                         | 是       |
+| type           | string              | 列的类型，现支持INT, FLOAT, STRING, BIGINT, DOUBLE, BOOLEAN, LONG。 | 否       |
+| index          | 升序排列            | 列索引位置(要么全部列都写，要么都不写)，如reader中取到的某一字段在第二位置（eg： name， id， age）但kudu目标表结构不同（eg：id，name， age），此时就需要将index赋值为（1，0，2），默认顺序（0，1，2） | 否       |
+| primaryKey     | false               | 是否为主键（请将所有的主键列写在前面）,不表明主键将不会检查过滤脏数据 | 否       |
+| compress       | DEFAULT_COMPRESSION | 压缩格式                                                     | 否       |
+| encoding       | AUTO_ENCODING       | 编码                                                         | 否       |
+| replicaCount   | 3                   | 保留副本个数                                                 | 否       |
+| hash           |                     | hash分区                                                     | 否       |
+| number         | 3                   | hash分区个数                                                 | 否       |
+| range          |                     | range分区                                                    | 否       |
+| lower          |                     | range分区下限 (eg: sql建表：partition value='haha' 对应：“lower”：“haha”，“upper”：“haha\000”) | 否       |
+| upper          |                     | range分区上限(eg: sql建表：partition "10" <= VALUES < "20"                                                                     对应：“lower”：“10”，“upper”：“20”) | 否       |
+| truncate       | false               | 是否清空表，本质上是删表重建                                 | 否       |
+| writeMode      | upsert              | upsert，insert，update                                       | 否       |
+| batchSize      | 512                 | 每xx行数据flush一次结果（最好不要超过1024）                  | 否       |
+| bufferSize     | 3072                | 缓冲区大小                                                   | 否       |
+| skipFail       | false               | 是否跳过插入不成功的数据                                     | 否       |
+| timeout        | 60000               | client超时时间,如创建表，删除表操作的超时时间。单位：ms      | 否       |
+| sessionTimeout | 60000               | session超时时间 单位：ms                                     | 否       |
+
+
+
+
+
+
+
+

kuduwriter/pom.xml

@@ -0,0 +1,82 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>datax-all</artifactId>
+        <groupId>com.alibaba.datax</groupId>
+        <version>0.0.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>kuduwriter</artifactId>
+    <dependencies>
+        <dependency>
+            <groupId>com.alibaba.datax</groupId>
+            <artifactId>datax-common</artifactId>
+            <version>${datax-project-version}</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>slf4j-log4j12</artifactId>
+                    <groupId>org.slf4j</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.kudu</groupId>
+            <artifactId>kudu-client</artifactId>
+            <version>1.11.1</version>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.13.1</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba.datax</groupId>
+            <artifactId>datax-core</artifactId>
+            <version>${datax-project-version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.alibaba.datax</groupId>
+                    <artifactId>datax-service-face</artifactId>
+                </exclusion>
+            </exclusions>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <!-- compiler plugin -->
+            <plugin>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>${jdk-version}</source>
+                    <target>${jdk-version}</target>
+                    <encoding>${project-sourceEncoding}</encoding>
+                </configuration>
+            </plugin>
+            <!-- assembly plugin -->
+            <plugin>
+                <artifactId>maven-assembly-plugin</artifactId>
+                <configuration>
+                    <descriptors>
+                        <descriptor>src/main/assembly/package.xml</descriptor>
+                    </descriptors>
+                    <finalName>datax</finalName>
+                </configuration>
+                <executions>
+                    <execution>
+                        <id>dwzip</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>single</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

kuduwriter/src/main/assembly/package.xml

@@ -0,0 +1,35 @@
+<assembly
+		xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
+		xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
+	<id></id>
+	<formats>
+		<format>dir</format>
+	</formats>
+	<includeBaseDirectory>false</includeBaseDirectory>
+	<fileSets>
+		<fileSet>
+			<directory>src/main/resources</directory>
+			<includes>
+				<include>plugin.json</include>
+				<include>plugin_job_template.json</include>
+			</includes>
+			<outputDirectory>plugin/writer/kuduwriter</outputDirectory>
+		</fileSet>
+		<fileSet>
+			<directory>target/</directory>
+			<includes>
+				<include>kuduwriter-0.0.1-SNAPSHOT.jar</include>
+			</includes>
+			<outputDirectory>plugin/writer/kuduwriter</outputDirectory>
+		</fileSet>
+	</fileSets>
+
+	<dependencySets>
+		<dependencySet>
+			<useProjectArtifact>false</useProjectArtifact>
+			<outputDirectory>plugin/writer/kuduwriter/libs</outputDirectory>
+			<scope>runtime</scope>
+		</dependencySet>
+	</dependencySets>
+</assembly>

kuduwriter/src/main/java/com/q1/datax/plugin/writer/kudu11xwriter/ColumnType.java

@@ -0,0 +1,37 @@
+package com.q1.datax.plugin.writer.kudu11xwriter;
+
+import com.alibaba.datax.common.exception.DataXException;
+
+import java.util.Arrays;
+
+/**
+ * @author daizihao
+ * @create 2020-08-31 19:12
+ **/
+public enum ColumnType {
+    INT("int"),
+    FLOAT("float"),
+    STRING("string"),
+    BIGINT("bigint"),
+    DOUBLE("double"),
+    BOOLEAN("boolean"),
+    LONG("long");
+    private String mode;
+    ColumnType(String mode) {
+        this.mode = mode.toLowerCase();
+    }
+
+    public String getMode() {
+        return mode;
+    }
+
+    public static ColumnType getByTypeName(String modeName) {
+        for (ColumnType modeType : values()) {
+            if (modeType.mode.equalsIgnoreCase(modeName)) {
+                return modeType;
+            }
+        }
+        throw DataXException.asDataXException(Kudu11xWriterErrorcode.ILLEGAL_VALUE,
+                String.format("Kuduwriter does not support the type:%s, currently supported types are:%s", modeName, Arrays.asList(values())));
+    }
+}

kuduwriter/src/main/java/com/q1/datax/plugin/writer/kudu11xwriter/Constant.java

@@ -0,0 +1,21 @@
+package com.q1.datax.plugin.writer.kudu11xwriter;
+
+/**
+ * @author daizihao
+ * @create 2020-08-31 14:42
+ **/
+public class Constant {
+    public static final String DEFAULT_ENCODING = "UTF-8";
+//    public static final String DEFAULT_DATA_FORMAT = "yyyy-MM-dd HH:mm:ss";
+
+    public static final String COMPRESSION = "DEFAULT_COMPRESSION";
+    public static final String ENCODING = "AUTO_ENCODING";
+    public static final Long ADMIN_TIMEOUTMS = 60000L;
+    public static final Long SESSION_TIMEOUTMS = 60000L;
+
+
+    public static final String INSERT_MODE = "upsert";
+    public static final long DEFAULT_WRITE_BATCH_SIZE = 512L;
+    public static final long DEFAULT_MUTATION_BUFFER_SPACE = 3072L;
+
+}

kuduwriter/src/main/java/com/q1/datax/plugin/writer/kudu11xwriter/InsertModeType.java

@@ -0,0 +1,34 @@
+package com.q1.datax.plugin.writer.kudu11xwriter;
+
+import com.alibaba.datax.common.exception.DataXException;
+
+import java.util.Arrays;
+
+/**
+ * @author daizihao
+ * @create 2020-08-31 14:47
+ **/
+public enum InsertModeType {
+    Insert("insert"),
+    Upsert("upsert"),
+    Update("update");
+    private String mode;
+
+    InsertModeType(String mode) {
+        this.mode = mode.toLowerCase();
+    }
+
+    public String getMode() {
+        return mode;
+    }
+
+    public static InsertModeType getByTypeName(String modeName) {
+        for (InsertModeType modeType : values()) {
+            if (modeType.mode.equalsIgnoreCase(modeName)) {
+                return modeType;
+            }
+        }
+        throw DataXException.asDataXException(Kudu11xWriterErrorcode.ILLEGAL_VALUE,
+                String.format("Kuduwriter does not support the mode :[%s], currently supported mode types are :%s", modeName, Arrays.asList(values())));
+    }
+}

kuduwriter/src/main/java/com/q1/datax/plugin/writer/kudu11xwriter/Key.java

@@ -0,0 +1,45 @@
+package com.q1.datax.plugin.writer.kudu11xwriter;
+
+/**
+ * @author daizihao
+ * @create 2020-08-31 14:17
+ **/
+public class Key {
+    public final static String KUDU_CONFIG = "kuduConfig";
+    public final static String KUDU_MASTER = "kudu.master_addresses";
+    public final static String KUDU_ADMIN_TIMEOUT = "timeout";
+    public final static String KUDU_SESSION_TIMEOUT = "sessionTimeout";
+
+    public final static String TABLE = "table";
+    public final static String PARTITION = "partition";
+    public final static String COLUMN = "column";
+
+    public static final String NAME = "name";
+    public static final String TYPE = "type";
+    public static final String INDEX = "index";
+    public static final String PRIMARYKEY = "primaryKey";
+    public static final String COMPRESSION = "compress";
+    public static final String COMMENT = "comment";
+    public final static String ENCODING = "encoding";
+
+
+
+    public static final String NUM_REPLICAS = "replicaCount";
+    public static final String HASH = "hash";
+    public static final String HASH_NUM = "number";
+
+    public static final String RANGE = "range";
+    public static final String LOWER = "lower";
+    public static final String UPPER = "upper";
+
+
+
+    public static  final String TRUNCATE = "truncate";
+
+    public static final String INSERT_MODE = "writeMode";
+
+    public static  final String WRITE_BATCH_SIZE = "batchSize";
+
+    public static  final String MUTATION_BUFFER_SPACE = "bufferSize";
+    public static  final String SKIP_FAIL = "skipFail";
+}

kuduwriter/src/main/java/com/q1/datax/plugin/writer/kudu11xwriter/Kudu11xHelper.java

@@ -0,0 +1,369 @@
+package com.q1.datax.plugin.writer.kudu11xwriter;
+
+import com.alibaba.datax.common.element.Column;
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.fastjson.JSON;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.Validate;
+import org.apache.kudu.ColumnSchema;
+import org.apache.kudu.Schema;
+import org.apache.kudu.Type;
+import org.apache.kudu.client.*;
+import org.apache.kudu.shaded.org.checkerframework.checker.units.qual.K;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.rmi.runtime.Log;
+
+import java.nio.charset.Charset;
+import java.util.*;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * @author daizihao
+ * @create 2020-08-27 18:30
+ **/
+public class Kudu11xHelper {
+
+    private static final Logger LOG = LoggerFactory.getLogger(Kudu11xHelper.class);
+
+    public static Map<String, Object> getKuduConfiguration(String kuduConfig) {
+        if (StringUtils.isBlank(kuduConfig)) {
+            throw DataXException.asDataXException(Kudu11xWriterErrorcode.REQUIRED_VALUE,
+                    "Connection configuration information required.");
+        }
+        Map<String, Object> kConfiguration;
+        try {
+            kConfiguration = JSON.parseObject(kuduConfig, HashMap.class);
+            Validate.isTrue(kConfiguration != null, "kuduConfig is null!");
+            kConfiguration.put(Key.KUDU_ADMIN_TIMEOUT, kConfiguration.getOrDefault(Key.KUDU_ADMIN_TIMEOUT, Constant.ADMIN_TIMEOUTMS));
+            kConfiguration.put(Key.KUDU_SESSION_TIMEOUT, kConfiguration.getOrDefault(Key.KUDU_SESSION_TIMEOUT, Constant.SESSION_TIMEOUTMS));
+        } catch (Exception e) {
+            throw DataXException.asDataXException(Kudu11xWriterErrorcode.GET_KUDU_CONNECTION_ERROR, e);
+        }
+
+        return kConfiguration;
+    }
+
+    public static KuduClient getKuduClient(String kuduConfig) {
+        Map<String, Object> conf = Kudu11xHelper.getKuduConfiguration(kuduConfig);
+        KuduClient kuduClient = null;
+        try {
+            String masterAddress = (String) conf.get(Key.KUDU_MASTER);
+            kuduClient = new KuduClient.KuduClientBuilder(masterAddress)
+                    .defaultAdminOperationTimeoutMs((Long) conf.get(Key.KUDU_ADMIN_TIMEOUT))
+                    .defaultOperationTimeoutMs((Long) conf.get(Key.KUDU_SESSION_TIMEOUT))
+                    .build();
+        } catch (Exception e) {
+            throw DataXException.asDataXException(Kudu11xWriterErrorcode.GET_KUDU_CONNECTION_ERROR, e);
+        }
+        return kuduClient;
+    }
+
+    public static KuduTable getKuduTable(Configuration configuration, KuduClient kuduClient) {
+        String tableName = configuration.getString(Key.TABLE);
+
+        KuduTable table = null;
+        try {
+            if (kuduClient.tableExists(tableName)) {
+                table = kuduClient.openTable(tableName);
+            } else {
+                synchronized (Kudu11xHelper.class) {
+                    if (!kuduClient.tableExists(tableName)) {
+                        Schema schema = Kudu11xHelper.getSchema(configuration);
+                        CreateTableOptions tableOptions = new CreateTableOptions();
+
+                        Kudu11xHelper.setTablePartition(configuration, tableOptions, schema);
+                        //副本数
+                        Integer numReplicas = configuration.getInt(Key.NUM_REPLICAS, 3);
+                        tableOptions.setNumReplicas(numReplicas);
+                        table = kuduClient.createTable(tableName, schema, tableOptions);
+                    } else {
+                        table = kuduClient.openTable(tableName);
+                    }
+                }
+            }
+
+
+        } catch (Exception e) {
+            throw DataXException.asDataXException(Kudu11xWriterErrorcode.GET_KUDU_TABLE_ERROR, e);
+        }
+        return table;
+    }
+
+    public static void createTable(Configuration configuration) {
+        String tableName = configuration.getString(Key.TABLE);
+        String kuduConfig = configuration.getString(Key.KUDU_CONFIG);
+        KuduClient kuduClient = Kudu11xHelper.getKuduClient(kuduConfig);
+        try {
+            Schema schema = Kudu11xHelper.getSchema(configuration);
+            CreateTableOptions tableOptions = new CreateTableOptions();
+
+            Kudu11xHelper.setTablePartition(configuration, tableOptions, schema);
+            //副本数
+            Integer numReplicas = configuration.getInt(Key.NUM_REPLICAS, 3);
+            tableOptions.setNumReplicas(numReplicas);
+            kuduClient.createTable(tableName, schema, tableOptions);
+        } catch (Exception e) {
+            throw DataXException.asDataXException(Kudu11xWriterErrorcode.GREATE_KUDU_TABLE_ERROR, e);
+        } finally {
+            AtomicInteger i = new AtomicInteger(10);
+            while (i.get() > 0) {
+                try {
+                    if (kuduClient.isCreateTableDone(tableName)) {
+                        Kudu11xHelper.closeClient(kuduClient);
+                        LOG.info("Table " + tableName + " is created!");
+                        break;
+                    }
+                    i.decrementAndGet();
+                    LOG.error("timeout!");
+                } catch (KuduException e) {
+                    LOG.info("Wait for the table to be created..... " + i);
+                    try {
+                        Thread.sleep(100L);
+                    } catch (InterruptedException ex) {
+                        ex.printStackTrace();
+                    }
+                    i.decrementAndGet();
+                }
+            }
+            try {
+                if (kuduClient != null) {
+                    kuduClient.close();
+                }
+            } catch (KuduException e) {
+                LOG.info("Kudu client has been shut down!");
+            }
+        }
+    }
+
+    public static ThreadPoolExecutor createRowAddThreadPool(int coreSize) {
+        return new ThreadPoolExecutor(coreSize,
+                coreSize,
+                60L,
+                TimeUnit.SECONDS,
+                new SynchronousQueue<Runnable>(),
+                new ThreadFactory() {
+                    private final ThreadGroup group = System.getSecurityManager() == null ? Thread.currentThread().getThreadGroup() : System.getSecurityManager().getThreadGroup();
+                    private final AtomicInteger threadNumber = new AtomicInteger(1);
+
+                    @Override
+                    public Thread newThread(Runnable r) {
+                        Thread t = new Thread(group, r,
+                                "pool-kudu_rows_add-thread-" + threadNumber.getAndIncrement(),
+                                0);
+                        if (t.isDaemon())
+                            t.setDaemon(false);
+                        if (t.getPriority() != Thread.NORM_PRIORITY)
+                            t.setPriority(Thread.NORM_PRIORITY);
+                        return t;
+                    }
+                }, new ThreadPoolExecutor.CallerRunsPolicy());
+    }
+
+    public static List<List<Configuration>> getColumnLists(List<Configuration> columns) {
+        int quota = 8;
+        int num = (columns.size() - 1) / quota + 1;
+        int gap = columns.size() / num;
+        List<List<Configuration>> columnLists = new ArrayList<>(num);
+        for (int j = 0; j < num - 1; j++) {
+            List<Configuration> destList = new ArrayList<>(columns.subList(j * gap, (j + 1) * gap));
+            columnLists.add(destList);
+        }
+        List<Configuration> destList = new ArrayList<>(columns.subList(gap * (num - 1), columns.size()));
+        columnLists.add(destList);
+        return columnLists;
+    }
+
+    public static boolean isTableExists(Configuration configuration) {
+        String tableName = configuration.getString(Key.TABLE);
+        String kuduConfig = configuration.getString(Key.KUDU_CONFIG);
+        KuduClient kuduClient = Kudu11xHelper.getKuduClient(kuduConfig);
+        try {
+            return kuduClient.tableExists(tableName);
+        } catch (Exception e) {
+            throw DataXException.asDataXException(Kudu11xWriterErrorcode.GET_KUDU_CONNECTION_ERROR, e);
+        } finally {
+            Kudu11xHelper.closeClient(kuduClient);
+        }
+    }
+
+    public static void closeClient(KuduClient kuduClient) {
+        try {
+            if (kuduClient != null) {
+                kuduClient.close();
+            }
+        } catch (KuduException e) {
+            LOG.warn("The \"kudu client\" was not stopped gracefully. !");
+
+        }
+
+    }
+
+    public static Schema getSchema(Configuration configuration) {
+        List<Configuration> columns = configuration.getListConfiguration(Key.COLUMN);
+        List<ColumnSchema> columnSchemas = new ArrayList<>();
+        Schema schema = null;
+        if (columns == null || columns.isEmpty()) {
+            throw DataXException.asDataXException(Kudu11xWriterErrorcode.REQUIRED_VALUE, "column is not defined，eg：column:[{\"name\": \"cf0:column0\",\"type\": \"string\"},{\"name\": \"cf1:column1\",\"type\": \"long\"}]");
+        }
+        try {
+            for (Configuration column : columns) {
+
+                String type = "BIGINT".equals(column.getNecessaryValue(Key.TYPE, Kudu11xWriterErrorcode.REQUIRED_VALUE).toUpperCase()) ||
+                        "LONG".equals(column.getNecessaryValue(Key.TYPE, Kudu11xWriterErrorcode.REQUIRED_VALUE).toUpperCase()) ?
+                        "INT64" : "INT".equals(column.getNecessaryValue(Key.TYPE, Kudu11xWriterErrorcode.REQUIRED_VALUE).toUpperCase()) ?
+                        "INT32" : column.getNecessaryValue(Key.TYPE, Kudu11xWriterErrorcode.REQUIRED_VALUE).toUpperCase();
+                String name = column.getNecessaryValue(Key.NAME, Kudu11xWriterErrorcode.REQUIRED_VALUE);
+                Boolean key = column.getBool(Key.PRIMARYKEY, false);
+                String encoding = column.getString(Key.ENCODING, Constant.ENCODING).toUpperCase();
+                String compression = column.getString(Key.COMPRESSION, Constant.COMPRESSION).toUpperCase();
+                String comment = column.getString(Key.COMMENT, "");
+
+                columnSchemas.add(new ColumnSchema.ColumnSchemaBuilder(name, Type.getTypeForName(type))
+                        .key(key)
+                        .encoding(ColumnSchema.Encoding.valueOf(encoding))
+                        .compressionAlgorithm(ColumnSchema.CompressionAlgorithm.valueOf(compression))
+                        .comment(comment)
+                        .build());
+            }
+            schema = new Schema(columnSchemas);
+        } catch (Exception e) {
+            throw DataXException.asDataXException(Kudu11xWriterErrorcode.REQUIRED_VALUE, e);
+        }
+        return schema;
+    }
+
+    public static Integer getPrimaryKeyIndexUntil(List<Configuration> columns) {
+        int i = 0;
+        while (i < columns.size()) {
+            Configuration col = columns.get(i);
+            if (!col.getBool(Key.PRIMARYKEY, false)) {
+                break;
+            }
+            i++;
+        }
+        return i;
+    }
+
+    public static void setTablePartition(Configuration configuration,
+                                         CreateTableOptions tableOptions,
+                                         Schema schema) {
+        Configuration partition = configuration.getConfiguration(Key.PARTITION);
+        if (partition == null) {
+            ColumnSchema columnSchema = schema.getColumns().get(0);
+            tableOptions.addHashPartitions(Collections.singletonList(columnSchema.getName()), 3);
+            return;
+        }
+        //range分区
+        Configuration range = partition.getConfiguration(Key.RANGE);
+        if (range != null) {
+            List<String> rangeColums = new ArrayList<>(range.getKeys());
+            tableOptions.setRangePartitionColumns(rangeColums);
+            for (String rangeColum : rangeColums) {
+                List<Configuration> lowerAndUppers = range.getListConfiguration(rangeColum);
+                for (Configuration lowerAndUpper : lowerAndUppers) {
+                    PartialRow lower = schema.newPartialRow();
+                    lower.addString(rangeColum, lowerAndUpper.getNecessaryValue(Key.LOWER, Kudu11xWriterErrorcode.REQUIRED_VALUE));
+                    PartialRow upper = schema.newPartialRow();
+                    upper.addString(rangeColum, lowerAndUpper.getNecessaryValue(Key.UPPER, Kudu11xWriterErrorcode.REQUIRED_VALUE));
+                    tableOptions.addRangePartition(lower, upper);
+                }
+            }
+            LOG.info("Set range partition complete!");
+        }
+
+        // 设置Hash分区
+        Configuration hash = partition.getConfiguration(Key.HASH);
+        if (hash != null) {
+            List<String> hashColums = hash.getList(Key.COLUMN, String.class);
+            Integer hashPartitionNum = configuration.getInt(Key.HASH_NUM, 3);
+            tableOptions.addHashPartitions(hashColums, hashPartitionNum);
+            LOG.info("Set hash partition complete!");
+        }
+    }
+
+    public static void validateParameter(Configuration configuration) {
+        LOG.info("Start validating parameters！");
+        configuration.getNecessaryValue(Key.KUDU_CONFIG, Kudu11xWriterErrorcode.REQUIRED_VALUE);
+        configuration.getNecessaryValue(Key.TABLE, Kudu11xWriterErrorcode.REQUIRED_VALUE);
+        String encoding = configuration.getString(Key.ENCODING, Constant.DEFAULT_ENCODING);
+        if (!Charset.isSupported(encoding)) {
+            throw DataXException.asDataXException(Kudu11xWriterErrorcode.ILLEGAL_VALUE,
+                    String.format("Encoding is not supported:[%s] .", encoding));
+        }
+        configuration.set(Key.ENCODING, encoding);
+        String insertMode = configuration.getString(Key.INSERT_MODE, Constant.INSERT_MODE);
+        try {
+            InsertModeType.getByTypeName(insertMode);
+        } catch (Exception e) {
+            insertMode = Constant.INSERT_MODE;
+        }
+        configuration.set(Key.INSERT_MODE, insertMode);
+
+        Long writeBufferSize = configuration.getLong(Key.WRITE_BATCH_SIZE, Constant.DEFAULT_WRITE_BATCH_SIZE);
+        configuration.set(Key.WRITE_BATCH_SIZE, writeBufferSize);
+
+        Long mutationBufferSpace = configuration.getLong(Key.MUTATION_BUFFER_SPACE, Constant.DEFAULT_MUTATION_BUFFER_SPACE);
+        configuration.set(Key.MUTATION_BUFFER_SPACE, mutationBufferSpace);
+
+        Boolean isSkipFail = configuration.getBool(Key.SKIP_FAIL, false);
+        configuration.set(Key.SKIP_FAIL, isSkipFail);
+        List<Configuration> columns = configuration.getListConfiguration(Key.COLUMN);
+        List<Configuration> goalColumns = new ArrayList<>();
+        //column参数验证
+        int indexFlag = 0;
+        boolean primaryKey = true;
+        int primaryKeyFlag = 0;
+        for (int i = 0; i < columns.size(); i++) {
+            Configuration col = columns.get(i);
+            String index = col.getString(Key.INDEX);
+            if (index == null) {
+                index = String.valueOf(i);
+                col.set(Key.INDEX, index);
+                indexFlag++;
+            }
+            if(primaryKey != col.getBool(Key.PRIMARYKEY, false)){
+                primaryKey = col.getBool(Key.PRIMARYKEY, false);
+                primaryKeyFlag++;
+            }
+            goalColumns.add(col);
+        }
+        if (indexFlag != 0 && indexFlag != columns.size()) {
+            throw DataXException.asDataXException(Kudu11xWriterErrorcode.ILLEGAL_VALUE,
+                    "\"index\" either has values for all of them, or all of them are null!");
+        }
+        if (primaryKeyFlag > 1){
+            throw DataXException.asDataXException(Kudu11xWriterErrorcode.ILLEGAL_VALUE,
+                    "\"primaryKey\" must be written in the front！");
+        }
+        configuration.set(Key.COLUMN, goalColumns);
+//        LOG.info("------------------------------------");
+//        LOG.info(configuration.toString());
+//        LOG.info("------------------------------------");
+        LOG.info("validate parameter complete！");
+    }
+
+    public static void truncateTable(Configuration configuration) {
+        String kuduConfig = configuration.getString(Key.KUDU_CONFIG);
+        String userTable = configuration.getString(Key.TABLE);
+        LOG.info(String.format("Because you have configured truncate is true,KuduWriter begins to truncate table %s .", userTable));
+        KuduClient kuduClient = Kudu11xHelper.getKuduClient(kuduConfig);
+
+        try {
+            if (kuduClient.tableExists(userTable)) {
+                kuduClient.deleteTable(userTable);
+                LOG.info(String.format("table  %s has been deleted.", userTable));
+            }
+        } catch (KuduException e) {
+            throw DataXException.asDataXException(Kudu11xWriterErrorcode.DELETE_KUDU_ERROR, e);
+        } finally {
+            Kudu11xHelper.closeClient(kuduClient);
+        }
+
+    }
+}

kuduwriter/src/main/java/com/q1/datax/plugin/writer/kudu11xwriter/Kudu11xWriter.java

@@ -0,0 +1,85 @@
+package com.q1.datax.plugin.writer.kudu11xwriter;
+
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.common.plugin.RecordReceiver;
+import com.alibaba.datax.common.spi.Writer;
+import com.alibaba.datax.common.util.Configuration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author daizihao
+ * @create 2020-08-27 16:58
+ **/
+public class Kudu11xWriter extends Writer {
+    public static class Job extends Writer.Job{
+        private static final Logger LOG = LoggerFactory.getLogger(Job.class);
+        private Configuration config = null;
+        @Override
+        public void init() {
+            this.config = this.getPluginJobConf();
+            Kudu11xHelper.validateParameter(this.config);
+        }
+
+        @Override
+        public void prepare() {
+            Boolean truncate = config.getBool(Key.TRUNCATE,false);
+            if(truncate){
+                Kudu11xHelper.truncateTable(this.config);
+            }
+
+            if (!Kudu11xHelper.isTableExists(config)){
+                Kudu11xHelper.createTable(config);
+            }
+        }
+
+        @Override
+        public List<Configuration> split(int i) {
+            List<Configuration> splitResultConfigs = new ArrayList<>();
+            for (int j = 0; j < i; j++) {
+                splitResultConfigs.add(config.clone());
+            }
+
+            return splitResultConfigs;
+        }
+
+
+
+        @Override
+        public void destroy() {
+
+        }
+    }
+
+    public static class Task extends Writer.Task{
+        private Configuration taskConfig;
+        private KuduWriterTask kuduTaskProxy;
+        private static final Logger LOG = LoggerFactory.getLogger(Job.class);
+        @Override
+        public void init() {
+            this.taskConfig = super.getPluginJobConf();
+            this.kuduTaskProxy = new KuduWriterTask(this.taskConfig);
+        }
+        @Override
+        public void startWrite(RecordReceiver lineReceiver) {
+            this.kuduTaskProxy.startWriter(lineReceiver,super.getTaskPluginCollector());
+        }
+
+
+        @Override
+        public void destroy() {
+            try {
+                if (kuduTaskProxy.session != null) {
+                    kuduTaskProxy.session.close();
+                }
+            }catch (Exception e){
+                LOG.warn("The \"kudu session\" was not stopped gracefully !");
+            }
+            Kudu11xHelper.closeClient(kuduTaskProxy.kuduClient);
+
+        }
+    }
+}

kuduwriter/src/main/java/com/q1/datax/plugin/writer/kudu11xwriter/Kudu11xWriterErrorcode.java

@@ -0,0 +1,39 @@
+package com.q1.datax.plugin.writer.kudu11xwriter;
+
+import com.alibaba.datax.common.spi.ErrorCode;
+
+/**
+ * @author daizihao
+ * @create 2020-08-27 19:25
+ **/
+public enum Kudu11xWriterErrorcode implements ErrorCode {
+    REQUIRED_VALUE("Kuduwriter-00", "You are missing a required parameter value."),
+    ILLEGAL_VALUE("Kuduwriter-01", "You fill in the parameter values are not legitimate."),
+    GET_KUDU_CONNECTION_ERROR("Kuduwriter-02", "Error getting Kudu connection."),
+    GET_KUDU_TABLE_ERROR("Kuduwriter-03", "Error getting Kudu table."),
+    CLOSE_KUDU_CONNECTION_ERROR("Kuduwriter-04", "Error closing Kudu connection."),
+    CLOSE_KUDU_SESSION_ERROR("Kuduwriter-06", "Error closing Kudu table connection."),
+    PUT_KUDU_ERROR("Kuduwriter-07", "IO exception occurred when writing to Kudu."),
+    DELETE_KUDU_ERROR("Kuduwriter-08", "An exception occurred while delete Kudu table."),
+    GREATE_KUDU_TABLE_ERROR("Kuduwriter-09", "Error creating Kudu table."),
+    PARAMETER_NUM_ERROR("Kuduwriter-10","The number of parameters does not match.")
+    ;
+
+    private final String code;
+    private final String description;
+
+
+    Kudu11xWriterErrorcode(String code, String description) {
+        this.code = code;
+        this.description = description;
+    }
+    @Override
+    public String getCode() {
+        return code;
+    }
+
+    @Override
+    public String getDescription() {
+        return description;
+    }
+}

kuduwriter/src/main/java/com/q1/datax/plugin/writer/kudu11xwriter/KuduWriterTask.java

@@ -0,0 +1,216 @@
+package com.q1.datax.plugin.writer.kudu11xwriter;
+
+import com.alibaba.datax.common.element.Column;
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.common.plugin.RecordReceiver;
+import com.alibaba.datax.common.plugin.TaskPluginCollector;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.common.util.RetryUtil;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.kudu.client.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.LongAdder;
+
+/**
+ * @author daizihao
+ * @create 2020-08-31 16:55
+ **/
+public class KuduWriterTask {
+    private final static Logger LOG = LoggerFactory.getLogger(KuduWriterTask.class);
+
+    private List<Configuration> columns;
+    private List<List<Configuration>> columnLists;
+    private ThreadPoolExecutor pool;
+    private String encoding;
+    private Double batchSize;
+    private Boolean isUpsert;
+    private Boolean isSkipFail;
+    public KuduClient kuduClient;
+    public KuduSession session;
+    private KuduTable table;
+    private Integer primaryKeyIndexUntil;
+
+    private final Object lock = new Object();
+
+    public KuduWriterTask(Configuration configuration) {
+        columns = configuration.getListConfiguration(Key.COLUMN);
+        columnLists = Kudu11xHelper.getColumnLists(columns);
+        pool = Kudu11xHelper.createRowAddThreadPool(columnLists.size());
+
+        this.encoding = configuration.getString(Key.ENCODING);
+        this.batchSize = configuration.getDouble(Key.WRITE_BATCH_SIZE);
+        this.isUpsert = !configuration.getString(Key.INSERT_MODE).equalsIgnoreCase("insert");
+        this.isSkipFail = configuration.getBool(Key.SKIP_FAIL);
+        long mutationBufferSpace = configuration.getLong(Key.MUTATION_BUFFER_SPACE);
+
+        this.kuduClient = Kudu11xHelper.getKuduClient(configuration.getString(Key.KUDU_CONFIG));
+        this.table = Kudu11xHelper.getKuduTable(configuration, kuduClient);
+        this.session = kuduClient.newSession();
+        session.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
+        session.setMutationBufferSpace((int) mutationBufferSpace);
+        this.primaryKeyIndexUntil = Kudu11xHelper.getPrimaryKeyIndexUntil(columns);
+//        tableName = configuration.getString(Key.TABLE);
+    }
+
+    public void startWriter(RecordReceiver lineReceiver, TaskPluginCollector taskPluginCollector) {
+        LOG.info("kuduwriter began to write!");
+        Record record;
+        LongAdder counter = new LongAdder();
+        try {
+            while ((record = lineReceiver.getFromReader()) != null) {
+                if (record.getColumnNumber() != columns.size()) {
+                    throw DataXException.asDataXException(Kudu11xWriterErrorcode.PARAMETER_NUM_ERROR, " number of record fields:" + record.getColumnNumber() + " number of configuration fields:" + columns.size());
+                }
+                boolean isDirtyRecord = false;
+
+
+                for (int i = 0; i < primaryKeyIndexUntil && !isDirtyRecord; i++) {
+                    Column column = record.getColumn(i);
+                    isDirtyRecord = StringUtils.isBlank(column.asString());
+                }
+
+                if (isDirtyRecord) {
+                    taskPluginCollector.collectDirtyRecord(record, "primarykey field is null");
+                    continue;
+                }
+
+                CountDownLatch countDownLatch = new CountDownLatch(columnLists.size());
+                Upsert upsert = table.newUpsert();
+                Insert insert = table.newInsert();
+                PartialRow row;
+                if (isUpsert) {
+                    //覆盖更新
+                    row = upsert.getRow();
+                } else {
+                    //增量更新
+                    row = insert.getRow();
+                }
+                List<Future<?>> futures = new ArrayList<>();
+                for (List<Configuration> columnList : columnLists) {
+                    Record finalRecord = record;
+                    Future<?> future = pool.submit(() -> {
+                        try {
+                            for (Configuration col : columnList) {
+                                String name = col.getString(Key.NAME);
+                                ColumnType type = ColumnType.getByTypeName(col.getString(Key.TYPE, "string"));
+                                Column column = finalRecord.getColumn(col.getInt(Key.INDEX));
+                                String rawData = column.asString();
+                                if (rawData == null) {
+                                    synchronized (lock) {
+                                        row.setNull(name);
+                                    }
+                                    continue;
+                                }
+                                switch (type) {
+                                    case INT:
+                                        synchronized (lock) {
+                                            row.addInt(name, Integer.parseInt(rawData));
+                                        }
+                                        break;
+                                    case LONG:
+                                    case BIGINT:
+                                        synchronized (lock) {
+                                            row.addLong(name, Long.parseLong(rawData));
+                                        }
+                                        break;
+                                    case FLOAT:
+                                        synchronized (lock) {
+                                            row.addFloat(name, Float.parseFloat(rawData));
+                                        }
+                                        break;
+                                    case DOUBLE:
+                                        synchronized (lock) {
+                                            row.addDouble(name, Double.parseDouble(rawData));
+                                        }
+                                        break;
+                                    case BOOLEAN:
+                                        synchronized (lock) {
+                                            row.addBoolean(name, Boolean.getBoolean(rawData));
+                                        }
+                                        break;
+                                    case STRING:
+                                    default:
+                                        synchronized (lock) {
+                                            row.addString(name, rawData);
+                                        }
+                                }
+                            }
+                        } finally {
+                            countDownLatch.countDown();
+                        }
+                    });
+                    futures.add(future);
+                }
+                countDownLatch.await();
+                for (Future<?> future : futures) {
+                    future.get();
+                }
+                try {
+                    RetryUtil.executeWithRetry(() -> {
+                        if (isUpsert) {
+                            //覆盖更新
+                            session.apply(upsert);
+                        } else {
+                            //增量更新
+                            session.apply(insert);
+                        }
+                        //flush
+                        if (counter.longValue() > (batchSize * 0.8)) {
+                            session.flush();
+                            counter.reset();
+                        }
+                        counter.increment();
+                        return true;
+                    }, 5, 500L, true);
+
+                } catch (Exception e) {
+                    LOG.error("Record Write Failure!", e);
+                    if (isSkipFail) {
+                        LOG.warn("Since you have configured \"skipFail\" to be true, this record will be skipped !");
+                        taskPluginCollector.collectDirtyRecord(record, e.getMessage());
+                    } else {
+                        throw DataXException.asDataXException(Kudu11xWriterErrorcode.PUT_KUDU_ERROR, e.getMessage());
+                    }
+                }
+            }
+        } catch (Exception e) {
+            LOG.error("write failure! the task will exit!");
+            throw DataXException.asDataXException(Kudu11xWriterErrorcode.PUT_KUDU_ERROR, e.getMessage());
+        }
+        AtomicInteger i = new AtomicInteger(10);
+        try {
+            while (i.get() > 0) {
+                if (session.hasPendingOperations()) {
+                    session.flush();
+                    break;
+                }
+                Thread.sleep(20L);
+                i.decrementAndGet();
+            }
+        } catch (Exception e) {
+            LOG.info("Waiting for data to be written to kudu...... " + i + "s");
+
+        } finally {
+            try {
+                pool.shutdown();
+                //强制刷写
+                session.flush();
+            } catch (KuduException e) {
+                LOG.error("kuduwriter flush error! The results may be incomplete！");
+                throw DataXException.asDataXException(Kudu11xWriterErrorcode.PUT_KUDU_ERROR, e.getMessage());
+            }
+        }
+
+    }
+
+
+}

kuduwriter/src/main/java/com/q1/kudu/conf/KuduConfig.java

@@ -0,0 +1,9 @@
+package com.q1.kudu.conf;
+
+/**
+ * @author daizihao
+ * @create 2020-09-16 11:39
+ **/
+public class KuduConfig {
+
+}

kuduwriter/src/main/resources/plugin.json

@@ -0,0 +1,7 @@
+{
+    "name": "kuduwriter",
+    "class": "com.q1.datax.plugin.writer.kudu11xwriter.Kudu11xWriter",
+    "description": "use put: prod. mechanism: use kudu java api put data.",
+    "developer": "com.q1.daizihao"
+}
+

kuduwriter/src/main/resources/plugin_job_template.json

@@ -0,0 +1,59 @@
+{
+  "name": "kuduwriter",
+  "parameter": {
+    "kuduConfig": {
+      "kudu.master_addresses": "***",
+      "timeout": 60000,
+      "sessionTimeout": 60000
+
+    },
+    "table": "",
+    "replicaCount": 3,
+    "truncate": false,
+    "writeMode": "upsert",
+    "partition": {
+      "range": {
+        "column1": [
+          {
+            "lower": "2020-08-25",
+            "upper": "2020-08-26"
+          },
+          {
+            "lower": "2020-08-26",
+            "upper": "2020-08-27"
+          },
+          {
+            "lower": "2020-08-27",
+            "upper": "2020-08-28"
+          }
+        ]
+      },
+      "hash": {
+        "column": [
+          "column1"
+        ],
+        "number": 3
+      }
+    },
+    "column": [
+      {
+        "index": 0,
+        "name": "c1",
+        "type": "string",
+        "primaryKey": true
+      },
+      {
+        "index": 1,
+        "name": "c2",
+        "type": "string",
+        "compress": "DEFAULT_COMPRESSION",
+        "encoding": "AUTO_ENCODING",
+        "comment": "注解xxxx"
+      }
+    ],
+    "batchSize": 1024,
+    "bufferSize": 2048,
+    "skipFail": false,
+    "encoding": "UTF-8"
+  }
+}

kuduwriter/src/test/java/com/dai/test.java

@@ -0,0 +1,40 @@
+package com.dai;
+
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.common.util.RetryUtil;
+import com.q1.datax.plugin.writer.kudu11xwriter.*;
+import static org.apache.kudu.client.AsyncKuduClient.LOG;
+
+/**
+ * @author daizihao
+ * @create 2020-08-28 11:03
+ **/
+public class test {
+    static boolean isSkipFail;
+
+
+    public static void main(String[] args) {
+        try {
+            while (true) {
+                try {
+                    RetryUtil.executeWithRetry(()->{
+                        throw new RuntimeException();
+                    },5,1000L,true);
+
+                } catch (Exception e) {
+                    LOG.error("Data write failed!", e);
+                    System.out.println(isSkipFail);
+                    if (isSkipFail) {
+                        LOG.warn("Because you have configured skipFail is true,this data will be skipped!");
+                    }else {
+                        System.out.println("异常抛出");
+                        throw e;
+                    }
+                }
+            }
+        } catch (Exception e) {
+            LOG.error("write failed! the task will exit!");
+            throw DataXException.asDataXException(Kudu11xWriterErrorcode.PUT_KUDU_ERROR, e);
+        }
+    }
+}

mongodbreader/doc/mongodbreader.md

@@ -8,7 +8,7 @@ MongoDBReader 插件利用 MongoDB 的java客户端MongoClient进行MongoDB的
 MongoDBReader通过Datax框架从MongoDB并行的读取数据，通过主控的JOB程序按照指定的规则对MongoDB中的数据进行分片，并行读取，然后将MongoDB支持的类型通过逐一判断转换成Datax支持的类型。
 
 #### 3 功能说明
-* 该示例从ODPS读一份数据到MongoDB。
+* 该示例从MongoDB读一份数据到ODPS。
 
 	    {
 	    "job": {
@@ -127,11 +127,13 @@ MongoDBReader通过Datax框架从MongoDB并行的读取数据，通过主控的J
 * address： MongoDB的数据地址信息，因为MonogDB可能是个集群，则ip端口信息需要以Json数组的形式给出。【必填】
 * userName：MongoDB的用户名。【选填】
 * userPassword： MongoDB的密码。【选填】
+* authDb: MongoDB认证数据库【选填】
 * collectionName： MonogoDB的集合名。【必填】
 * column：MongoDB的文档列名。【必填】
 * name：Column的名字。【必填】
 * type：Column的类型。【选填】
 * splitter：因为MongoDB支持数组类型，但是Datax框架本身不支持数组类型，所以mongoDB读出来的数组类型要通过这个分隔符合并成字符串。【选填】
+* query: MongoDB的额外查询条件。【选填】
 
 #### 5 类型转换
 
@@ -146,4 +148,4 @@ MongoDBReader通过Datax框架从MongoDB并行的读取数据，通过主控的J
 
 
 #### 6 性能报告
-#### 7 测试报告
+#### 7 测试报告

mongodbwriter/doc/mongodbwriter.md

@@ -139,7 +139,7 @@ MongoDBWriter通过Datax框架获取Reader生成的数据，然后将Datax支持
 * splitter：特殊分隔符，当且仅当要处理的字符串要用分隔符分隔为字符数组时，才使用这个参数，通过这个参数指定的分隔符，将字符串分隔存储到MongoDB的数组中。【选填】
 * upsertInfo：指定了传输数据时更新的信息。【选填】
 * isUpsert：当设置为true时，表示针对相同的upsertKey做更新操作。【选填】
-* upsertKey：upsertKey指定了没行记录的业务主键。用来做更新时使用。【选填】
+* upsertKey：upsertKey指定了每行记录的业务主键。用来做更新时使用。【选填】
 
 #### 5 类型转换
 
@@ -154,4 +154,4 @@ MongoDBWriter通过Datax框架获取Reader生成的数据，然后将Datax支持
 
 
 #### 6 性能报告
-#### 7 测试报告
+#### 7 测试报告

mysqlreader/doc/mysqlreader.md

@@ -165,7 +165,7 @@ MysqlReader插件实现了从Mysql读取数据。在底层实现上，MysqlReade
 
 	  支持常量配置，用户需要按照Mysql SQL语法格式:
 	  ["id", "\`table\`", "1", "'bazhen.csy'", "null", "to_char(a + 1)", "2.3" , "true"]
-	  id为普通列名，\`table\`为包含保留在的列名，1为整形数字常量，'bazhen.csy'为字符串常量，null为空指针，to_char(a + 1)为表达式，2.3为浮点数，true为布尔值。
+	  id为普通列名，\`table\`为包含保留字的列名，1为整形数字常量，'bazhen.csy'为字符串常量，null为空指针，to_char(a + 1)为表达式，2.3为浮点数，true为布尔值。
 
 	* 必选：是 <br />
 

mysqlreader/pom.xml

@@ -40,7 +40,7 @@
 		<dependency>
             <groupId>mysql</groupId>
             <artifactId>mysql-connector-java</artifactId>
-            <version>5.1.34</version>
+            <version>${mysql.driver.version}</version>
         </dependency>
 
 

mysqlwriter/doc/mysqlwriter.md

@@ -147,7 +147,7 @@ MysqlWriter 通过 DataX 框架获取 Reader 生成的协议数据，根据你
 
 * **column**
 
-	* 描述：目的表需要写入数据的字段,字段之间用英文逗号分隔。例如: "column": ["id","name","age"]。如果要依次写入全部列，使用*表示, 例如: "column": ["*"]。
+	* 描述：目的表需要写入数据的字段,字段之间用英文逗号分隔。例如: "column": ["id","name","age"]。如果要依次写入全部列，使用`*`表示, 例如: `"column": ["*"]`。
 
 			**column配置项必须指定，不能留空！**
 

mysqlwriter/pom.xml

@@ -40,7 +40,7 @@
 		<dependency>
             <groupId>mysql</groupId>
             <artifactId>mysql-connector-java</artifactId>
-            <version>5.1.34</version>
+            <version>${mysql.driver.version}</version>
         </dependency>
 	</dependencies>
 

oceanbasev10reader/pom.xml

@@ -0,0 +1,97 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>datax-all</artifactId>
+        <groupId>com.alibaba.datax</groupId>
+        <version>0.0.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>oceanbasev10reader</artifactId>
+    <groupId>com.alibaba.datax</groupId>
+    <version>0.0.1-SNAPSHOT</version>
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.alibaba.datax</groupId>
+            <artifactId>datax-common</artifactId>
+            <version>${datax-project-version}</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>slf4j-log4j12</artifactId>
+                    <groupId>org.slf4j</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+        </dependency>
+		<dependency>
+			<groupId>com.alibaba.datax</groupId>
+			<artifactId>plugin-rdbms-util</artifactId>
+			<version>${datax-project-version}</version>
+		</dependency>
+		<dependency>
+            <groupId>mysql</groupId>
+            <artifactId>mysql-connector-java</artifactId>
+            <version>5.1.40</version>
+        </dependency>
+        <dependency>
+            <groupId>log4j</groupId>
+            <artifactId>log4j</artifactId>
+            <version>1.2.16</version>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.11</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <resources>
+            <resource>
+                <directory>src/main/java</directory>
+                <includes>
+                    <include>**/*.properties</include>
+                </includes>
+            </resource>
+        </resources>
+        <plugins>
+            <!-- compiler plugin -->
+            <plugin>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>${jdk-version}</source>
+                    <target>${jdk-version}</target>
+                    <encoding>${project-sourceEncoding}</encoding>
+                </configuration>
+            </plugin>
+            <!-- assembly plugin -->
+            <plugin>
+                <artifactId>maven-assembly-plugin</artifactId>
+                <configuration>
+                    <descriptors>
+                        <descriptor>src/main/assembly/package.xml</descriptor>
+                    </descriptors>
+                    <finalName>datax</finalName>
+                </configuration>
+                <executions>
+                    <execution>
+                        <id>dwzip</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>single</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

oceanbasev10reader/src/main/assembly/package.xml

@@ -0,0 +1,42 @@
+<assembly
+        xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
+        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+        xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
+    <id></id>
+    <formats>
+        <format>dir</format>
+    </formats>
+    <includeBaseDirectory>false</includeBaseDirectory>
+    <fileSets>
+        <fileSet>
+            <directory>src/main/resources</directory>
+            <includes>
+                <include>plugin.json</include>
+                <include>plugin_job_template.json</include>
+            </includes>
+            <outputDirectory>plugin/reader/oceanbasev10reader</outputDirectory>
+        </fileSet>
+        <fileSet>
+            <directory>target/</directory>
+            <includes>
+                <include>oceanbasev10reader-0.0.1-SNAPSHOT.jar</include>
+            </includes>
+            <outputDirectory>plugin/reader/oceanbasev10reader</outputDirectory>
+        </fileSet>
+        <fileSet>
+            <directory>src/main/libs/</directory>
+            <includes>
+                <include>*.jar</include>
+            </includes>
+            <outputDirectory>plugin/reader/oceanbasev10reader/libs</outputDirectory>
+        </fileSet>
+    </fileSets>
+
+    <dependencySets>
+        <dependencySet>
+            <useProjectArtifact>false</useProjectArtifact>
+            <outputDirectory>plugin/reader/oceanbasev10reader/libs</outputDirectory>
+            <scope>runtime</scope>
+        </dependencySet>
+    </dependencySets>
+</assembly>

oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/Config.java

@@ -0,0 +1,16 @@
+package com.alibaba.datax.plugin.reader.oceanbasev10reader;
+
+public interface Config {
+    // queryTimeoutSecond
+    String QUERY_TIMEOUT_SECOND = "memstoreCheckIntervalSecond";
+
+    int DEFAULT_QUERY_TIMEOUT_SECOND = 60 * 60 * 48;// 2天
+
+    // readBatchSize
+    String READ_BATCH_SIZE = "readBatchSize";
+
+    int DEFAULT_READ_BATCH_SIZE = 100000;// 10万
+
+    String RETRY_LIMIT = "retryLimit";
+    int DEFAULT_RETRY_LIMIT = 10;
+}

oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/OceanBaseReader.java

@@ -0,0 +1,130 @@
+package com.alibaba.datax.plugin.reader.oceanbasev10reader;
+
+import java.sql.Connection;
+import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.datax.common.plugin.RecordSender;
+import com.alibaba.datax.common.spi.Reader;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.reader.Constant;
+import com.alibaba.datax.plugin.rdbms.reader.Key;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.datax.plugin.reader.oceanbasev10reader.ext.ReaderJob;
+import com.alibaba.datax.plugin.reader.oceanbasev10reader.ext.ReaderTask;
+import com.alibaba.datax.plugin.reader.oceanbasev10reader.util.ObReaderUtils;
+
+public class OceanBaseReader extends Reader {
+
+    public static class Job extends Reader.Job {
+        private Configuration originalConfig = null;
+        private ReaderJob readerJob;
+        private static final Logger LOG = LoggerFactory.getLogger(Task.class);
+
+        @Override
+        public void init() {
+            this.originalConfig = super.getPluginJobConf();
+
+            Integer userConfigedFetchSize = this.originalConfig.getInt(Constant.FETCH_SIZE);
+            if (userConfigedFetchSize != null) {
+                LOG.warn("The [fetchSize] is not recognized, please use readBatchSize instead.");
+            }
+            this.originalConfig.set(Constant.FETCH_SIZE, Integer.MIN_VALUE);
+            setDatabaseType(originalConfig);
+            this.readerJob = new ReaderJob();
+            this.readerJob.init(this.originalConfig);
+        }
+
+        @Override
+        public void prepare() {
+            //ObReaderUtils.DATABASE_TYPE获取当前数据库的语法模式
+        }
+
+        @Override
+        public void preCheck() {
+            init();
+            this.readerJob.preCheck(this.originalConfig, ObReaderUtils.databaseType);
+
+        }
+
+        @Override
+        public List<Configuration> split(int adviceNumber) {
+            return this.readerJob.split(this.originalConfig, adviceNumber);
+        }
+
+        @Override
+        public void post() {
+            this.readerJob.post(this.originalConfig);
+        }
+
+        @Override
+        public void destroy() {
+            this.readerJob.destroy(this.originalConfig);
+        }
+
+        private void setDatabaseType(Configuration config) {
+            String username = config.getString(Key.USERNAME);
+            String password = config.getString(Key.PASSWORD);
+            List<Object> conns = originalConfig.getList(Constant.CONN_MARK, Object.class);
+            Configuration connConf = Configuration.from(conns.get(0).toString());
+            List<String> jdbcUrls = connConf.getList(Key.JDBC_URL, String.class);
+            String jdbcUrl = jdbcUrls.get(0);
+            if (jdbcUrl.startsWith(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING)) {
+                String[] ss = jdbcUrl.split(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING_PATTERN);
+                if (ss.length != 3) {
+                    LOG.warn("unrecognized jdbc url: " + jdbcUrl);
+                    return;
+                }
+                username = ss[1].trim() + ":" + username;
+                jdbcUrl = ss[2];
+            }
+            // Use ob-client to get compatible mode.
+            try {
+                String obJdbcUrl = jdbcUrl.replace("jdbc:mysql:", "jdbc:oceanbase:");
+                Connection conn = DBUtil.getConnection(DataBaseType.OceanBase, obJdbcUrl, username, password);
+                String compatibleMode = ObReaderUtils.getCompatibleMode(conn);
+                if (ObReaderUtils.isOracleMode(compatibleMode)) {
+                    ObReaderUtils.compatibleMode = ObReaderUtils.OB_COMPATIBLE_MODE_ORACLE;
+                }
+
+            } catch (Exception e) {
+                LOG.warn("error in get compatible mode, using mysql as default: " + e.getMessage());
+            }
+        }
+    }
+
+    public static class Task extends Reader.Task {
+        private Configuration readerSliceConfig;
+        private ReaderTask commonRdbmsReaderTask;
+        private static final Logger LOG = LoggerFactory.getLogger(Task.class);
+
+        @Override
+        public void init() {
+            this.readerSliceConfig = super.getPluginJobConf();
+            this.commonRdbmsReaderTask = new ReaderTask(super.getTaskGroupId(), super.getTaskId());
+            this.commonRdbmsReaderTask.init(this.readerSliceConfig);
+
+        }
+
+        @Override
+        public void startRead(RecordSender recordSender) {
+            int fetchSize = this.readerSliceConfig.getInt(Constant.FETCH_SIZE);
+            this.commonRdbmsReaderTask.startRead(this.readerSliceConfig, recordSender, super.getTaskPluginCollector(),
+                    fetchSize);
+        }
+
+        @Override
+        public void post() {
+            this.commonRdbmsReaderTask.post(this.readerSliceConfig);
+        }
+
+        @Override
+        public void destroy() {
+            this.commonRdbmsReaderTask.destroy(this.readerSliceConfig);
+        }
+    }
+
+}

oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/ext/ReaderJob.java

@@ -0,0 +1,65 @@
+package com.alibaba.datax.plugin.reader.oceanbasev10reader.ext;
+
+import java.util.List;
+
+import com.alibaba.datax.common.constant.CommonConstant;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.reader.CommonRdbmsReader;
+import com.alibaba.datax.plugin.rdbms.reader.Key;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.datax.plugin.rdbms.writer.Constant;
+import com.alibaba.datax.plugin.reader.oceanbasev10reader.OceanBaseReader;
+import com.alibaba.datax.plugin.reader.oceanbasev10reader.util.ObReaderUtils;
+import com.alibaba.fastjson.JSONObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class ReaderJob extends CommonRdbmsReader.Job {
+    private Logger LOG = LoggerFactory.getLogger(OceanBaseReader.Task.class);
+
+    public ReaderJob() {
+        super(ObReaderUtils.databaseType);
+    }
+
+    @Override
+    public void init(Configuration originalConfig) {
+        //将config中的column和table中的关键字进行转义
+        List<String> columns = originalConfig.getList(Key.COLUMN, String.class);
+        ObReaderUtils.escapeDatabaseKeywords(columns);
+        originalConfig.set(Key.COLUMN, columns);
+
+        List<JSONObject> conns = originalConfig.getList(com.alibaba.datax.plugin.rdbms.reader.Constant.CONN_MARK, JSONObject.class);
+        for (int i = 0; i < conns.size(); i++) {
+            JSONObject conn = conns.get(i);
+            Configuration connConfig = Configuration.from(conn.toString());
+            List<String> tables = connConfig.getList(Key.TABLE, String.class);
+            ObReaderUtils.escapeDatabaseKeywords(tables);
+            originalConfig.set(String.format("%s[%d].%s", com.alibaba.datax.plugin.rdbms.reader.Constant.CONN_MARK, i, Key.TABLE), tables);
+        }
+        super.init(originalConfig);
+    }
+
+    @Override
+    public List<Configuration> split(Configuration originalConfig, int adviceNumber) {
+        List<Configuration> list = super.split(originalConfig, adviceNumber);
+        for (Configuration config : list) {
+            String jdbcUrl = config.getString(Key.JDBC_URL);
+            String obRegionName = getObRegionName(jdbcUrl);
+            config.set(CommonConstant.LOAD_BALANCE_RESOURCE_MARK, obRegionName);
+        }
+        return list;
+    }
+
+    private String getObRegionName(String jdbcUrl) {
+        if (jdbcUrl.startsWith(Constant.OB10_SPLIT_STRING)) {
+            String[] ss = jdbcUrl.split(Constant.OB10_SPLIT_STRING_PATTERN);
+            if (ss.length >= 2) {
+                String tenant = ss[1].trim();
+                String[] sss = tenant.split(":");
+                return sss[0];
+            }
+        }
+        return null;
+    }
+}

oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/ext/ReaderTask.java

@@ -0,0 +1,297 @@
+package com.alibaba.datax.plugin.reader.oceanbasev10reader.ext;
+
+import com.alibaba.datax.common.element.Column;
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.datax.common.plugin.RecordSender;
+import com.alibaba.datax.common.plugin.TaskPluginCollector;
+import com.alibaba.datax.common.statistics.PerfRecord;
+import com.alibaba.datax.common.statistics.PerfTrace;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.reader.CommonRdbmsReader;
+import com.alibaba.datax.plugin.rdbms.reader.Constant;
+import com.alibaba.datax.plugin.rdbms.reader.Key;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.RdbmsException;
+import com.alibaba.datax.plugin.reader.oceanbasev10reader.Config;
+import com.alibaba.datax.plugin.reader.oceanbasev10reader.util.ObReaderUtils;
+import com.alibaba.datax.plugin.reader.oceanbasev10reader.util.TaskContext;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.*;
+import java.util.ArrayList;
+import java.util.List;
+
+public class ReaderTask extends CommonRdbmsReader.Task {
+    private static final Logger LOG = LoggerFactory.getLogger(ReaderTask.class);
+    private int taskGroupId = -1;
+    private int taskId = -1;
+
+    private String username;
+    private String password;
+    private String jdbcUrl;
+    private String mandatoryEncoding;
+    private int queryTimeoutSeconds;// 查询超时 默认48小时
+    private int readBatchSize;
+    private int retryLimit = 0;
+    private String compatibleMode = ObReaderUtils.OB_COMPATIBLE_MODE_MYSQL;
+    private boolean reuseConn = false;
+
+    public ReaderTask(int taskGroupId, int taskId) {
+        super(ObReaderUtils.databaseType, taskGroupId, taskId);
+        this.taskGroupId = taskGroupId;
+        this.taskId = taskId;
+    }
+
+    @Override
+    public void init(Configuration readerSliceConfig) {
+        /* for database connection */
+        username = readerSliceConfig.getString(Key.USERNAME);
+        password = readerSliceConfig.getString(Key.PASSWORD);
+        jdbcUrl = readerSliceConfig.getString(Key.JDBC_URL);
+        queryTimeoutSeconds = readerSliceConfig.getInt(Config.QUERY_TIMEOUT_SECOND,
+                Config.DEFAULT_QUERY_TIMEOUT_SECOND);
+        // ob10的处理
+        if (jdbcUrl.startsWith(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING)) {
+            String[] ss = jdbcUrl.split(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING_PATTERN);
+            if (ss.length == 3) {
+                LOG.info("this is ob1_0 jdbc url.");
+                username = ss[1].trim() + ":" + username;
+                jdbcUrl = ss[2];
+            }
+        }
+
+        jdbcUrl = jdbcUrl.replace("jdbc:mysql:", "jdbc:oceanbase:") + "&socketTimeout=1800000&connectTimeout=60000"; //socketTimeout 半个小时
+        if (ObReaderUtils.compatibleMode.equals(ObReaderUtils.OB_COMPATIBLE_MODE_ORACLE)) {
+            compatibleMode = ObReaderUtils.OB_COMPATIBLE_MODE_ORACLE;
+        }
+        LOG.info("this is ob1_0 jdbc url. user=" + username + " :url=" + jdbcUrl);
+        mandatoryEncoding = readerSliceConfig.getString(Key.MANDATORY_ENCODING, "");
+        retryLimit = readerSliceConfig.getInt(Config.RETRY_LIMIT, Config.DEFAULT_RETRY_LIMIT);
+        LOG.info("retryLimit: " + retryLimit);
+    }
+
+    private void buildSavePoint(TaskContext context) {
+        if (!ObReaderUtils.isUserSavePointValid(context)) {
+            LOG.info("user save point is not valid, set to null.");
+            context.setUserSavePoint(null);
+        }
+    }
+
+    /**
+     * 如果isTableMode && table有PK
+     * <p>
+     * 则支持断点续读 (若pk不在原始的columns中,则追加到尾部,但不传给下游)
+     * <p>
+     * 否则,则使用旧模式
+     */
+    @Override
+    public void startRead(Configuration readerSliceConfig, RecordSender recordSender,
+                          TaskPluginCollector taskPluginCollector, int fetchSize) {
+        String querySql = readerSliceConfig.getString(Key.QUERY_SQL);
+        String table = readerSliceConfig.getString(Key.TABLE);
+        PerfTrace.getInstance().addTaskDetails(taskId, table + "," + jdbcUrl);
+        List<String> columns = readerSliceConfig.getList(Key.COLUMN_LIST, String.class);
+        String where = readerSliceConfig.getString(Key.WHERE);
+        boolean weakRead = readerSliceConfig.getBool(Key.WEAK_READ, true); // default true, using weak read
+        String userSavePoint = readerSliceConfig.getString(Key.SAVE_POINT, null);
+        reuseConn = readerSliceConfig.getBool(Key.REUSE_CONN, false);
+        String partitionName = readerSliceConfig.getString(Key.PARTITION_NAME, null);
+        // 从配置文件中取readBatchSize,若无则用默认值
+        readBatchSize = readerSliceConfig.getInt(Config.READ_BATCH_SIZE, Config.DEFAULT_READ_BATCH_SIZE);
+        // 不能少于1万
+        if (readBatchSize < 10000) {
+            readBatchSize = 10000;
+        }
+        TaskContext context = new TaskContext(table, columns, where, fetchSize);
+        context.setQuerySql(querySql);
+        context.setWeakRead(weakRead);
+        context.setCompatibleMode(compatibleMode);
+        if (partitionName != null) {
+            context.setPartitionName(partitionName);
+        }
+        // Add the user save point into the context
+        context.setUserSavePoint(userSavePoint);
+        PerfRecord allPerf = new PerfRecord(taskGroupId, taskId, PerfRecord.PHASE.RESULT_NEXT_ALL);
+        allPerf.start();
+        boolean isTableMode = readerSliceConfig.getBool(Constant.IS_TABLE_MODE);
+        try {
+            startRead0(isTableMode, context, recordSender, taskPluginCollector);
+        } finally {
+            ObReaderUtils.close(null, null, context.getConn());
+        }
+        allPerf.end(context.getCost());
+        // 目前大盘是依赖这个打印，而之前这个Finish read record是包含了sql查询和result next的全部时间
+        LOG.info("finished read record by Sql: [{}\n] {}.", context.getQuerySql(), jdbcUrl);
+    }
+
+    private void startRead0(boolean isTableMode, TaskContext context, RecordSender recordSender,
+                            TaskPluginCollector taskPluginCollector) {
+        // 不是table模式 直接使用原来的做法
+        if (!isTableMode) {
+            doRead(recordSender, taskPluginCollector, context);
+            return;
+        }
+        // check primary key index
+        Connection conn = DBUtil.getConnection(ObReaderUtils.databaseType, jdbcUrl, username, password);
+        ObReaderUtils.initConn4Reader(conn, queryTimeoutSeconds);
+        context.setConn(conn);
+        try {
+            ObReaderUtils.initIndex(conn, context);
+            ObReaderUtils.matchPkIndexs(conn, context);
+        } catch (Throwable e) {
+            LOG.warn("fetch PkIndexs fail,table=" + context.getTable(), e);
+        }
+        // 如果不是table 且 pk不存在 则仍然使用原来的做法
+        if (context.getPkIndexs() == null) {
+            doRead(recordSender, taskPluginCollector, context);
+            return;
+        }
+
+        // setup the user defined save point
+        buildSavePoint(context);
+
+        // 从这里开始就是 断点续读功能
+        // while(true) {
+        // 正常读 (需 order by pk asc)
+        // 如果遇到失败,分两种情况:
+        // a)已读出记录,则开始走增量读逻辑
+        // b)未读出记录,则走正常读逻辑(仍然需要order by pk asc)
+        // 正常结束 则 break
+        // }
+        context.setReadBatchSize(readBatchSize);
+        String getFirstQuerySql = ObReaderUtils.buildFirstQuerySql(context);
+        String appendQuerySql = ObReaderUtils.buildAppendQuerySql(conn, context);
+        LOG.warn("start table scan key : {}", context.getIndexName() == null ? "primary" : context.getIndexName());
+        context.setQuerySql(getFirstQuerySql);
+        boolean firstQuery = true;
+        // 原来打算firstQuery时 limit 1 减少
+        // 后来经过对比发现其实是多余的,因为:
+        // 1.假如走gmt_modified辅助索引,则直接索引扫描 不需要topN的order by
+        // 2.假如不走辅助索引,而是pk table scan,则减少排序规模并没有好处,因为下一次仍然要排序
+        // 减少这个多余的优化tip 可以让代码更易读
+        int retryCount = 0;
+        while (true) {
+            try {
+                boolean finish = doRead(recordSender, taskPluginCollector, context);
+                if (finish) {
+                    break;
+                }
+            } catch (Throwable e) {
+                if (retryLimit == ++retryCount) {
+                    throw RdbmsException.asQueryException(ObReaderUtils.databaseType, new Exception(e),
+                            context.getQuerySql(), context.getTable(), username);
+                }
+                LOG.error("read fail, retry count " + retryCount + ", sleep 60 second, save point:" +
+                        context.getSavePoint() + ", error: " + e.getMessage());
+                ObReaderUtils.sleep(60000); // sleep 10s
+            }
+            // 假如原来的查询有查出数据,则改成增量查询
+            if (firstQuery && context.getPkIndexs() != null && context.getSavePoint() != null) {
+                context.setQuerySql(appendQuerySql);
+                firstQuery = false;
+            }
+        }
+        DBUtil.closeDBResources(null, context.getConn());
+    }
+
+    private boolean isConnectionAlive(Connection conn) {
+        if (conn == null) {
+            return false;
+        }
+        Statement stmt = null;
+        ResultSet rs = null;
+        String sql = "select 1" + (compatibleMode == ObReaderUtils.OB_COMPATIBLE_MODE_ORACLE ? " from dual" : "");
+        try {
+            stmt = conn.createStatement();
+            rs = stmt.executeQuery(sql);
+            rs.next();
+        } catch (Exception ex) {
+            LOG.info("connection is not alive: " + ex.getMessage());
+            return false;
+        } finally {
+            DBUtil.closeDBResources(rs, stmt, null);
+        }
+        return true;
+    }
+
+    private boolean doRead(RecordSender recordSender, TaskPluginCollector taskPluginCollector, TaskContext context) {
+        LOG.info("exe sql: {}", context.getQuerySql());
+        Connection conn = context.getConn();
+        if (reuseConn && isConnectionAlive(conn)) {
+            LOG.info("connection is alive, will reuse this connection.");
+        } else {
+            LOG.info("Create new connection for reader.");
+            conn = DBUtil.getConnection(ObReaderUtils.databaseType, jdbcUrl, username, password);
+            ObReaderUtils.initConn4Reader(conn, queryTimeoutSeconds);
+            context.setConn(conn);
+        }
+        PreparedStatement ps = null;
+        ResultSet rs = null;
+        PerfRecord perfRecord = new PerfRecord(taskGroupId, taskId, PerfRecord.PHASE.SQL_QUERY);
+        perfRecord.start();
+        try {
+            ps = conn.prepareStatement(context.getQuerySql(),
+                    ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+            if (context.getPkIndexs() != null && context.getSavePoint() != null) {
+                Record savePoint = context.getSavePoint();
+                List<Column> point = ObReaderUtils.buildPoint(savePoint, context.getPkIndexs());
+                ObReaderUtils.binding(ps, point);
+                if (LOG.isWarnEnabled()) {
+                    List<String> pointForLog = new ArrayList<String>();
+                    for (Column c : point) {
+                        pointForLog.add(c.asString());
+                    }
+                    LOG.warn("{} save point : {}", context.getTable(), StringUtils.join(pointForLog, ','));
+                }
+            }
+            // 打开流式接口
+            ps.setFetchSize(context.getFetchSize());
+            rs = ps.executeQuery();
+            ResultSetMetaData metaData = rs.getMetaData();
+            int columnNumber = metaData.getColumnCount();
+            long lastTime = System.nanoTime();
+            int count = 0;
+            for (; rs.next(); count++) {
+                context.addCost(System.nanoTime() - lastTime);
+                Record row = buildRecord(recordSender, rs, metaData, columnNumber, mandatoryEncoding,
+                        taskPluginCollector);
+                // // 如果第一个record重复了,则不需要发送
+                // if (count == 0 &&
+                // ObReaderUtils.isPkEquals(context.getSavePoint(), row,
+                // context.getPkIndexs())) {
+                // continue;
+                // }
+                // 如果是querySql
+                if (context.getTransferColumnNumber() == -1
+                        || row.getColumnNumber() == context.getTransferColumnNumber()) {
+                    recordSender.sendToWriter(row);
+                } else {
+                    Record newRow = recordSender.createRecord();
+                    for (int i = 0; i < context.getTransferColumnNumber(); i++) {
+                        newRow.addColumn(row.getColumn(i));
+                    }
+                    recordSender.sendToWriter(newRow);
+                }
+                context.setSavePoint(row);
+                lastTime = System.nanoTime();
+            }
+            LOG.info("end of sql: {}, " + count + "rows are read.", context.getQuerySql());
+            return context.getReadBatchSize() <= 0 || count < readBatchSize;
+        } catch (Exception e) {
+            ObReaderUtils.close(null, null, context.getConn());
+            context.setConn(null);
+            LOG.error("reader data fail", e);
+            throw RdbmsException.asQueryException(ObReaderUtils.databaseType, e, context.getQuerySql(),
+                    context.getTable(), username);
+        } finally {
+            perfRecord.end();
+            if (reuseConn) {
+                ObReaderUtils.close(rs, ps, null);
+            } else {
+                ObReaderUtils.close(rs, ps, conn);
+            }
+        }
+    }
+}

oceanbasev10reader/src/main/java/com/alibaba/datax/plugin/reader/oceanbasev10reader/util/TaskContext.java

@@ -0,0 +1,177 @@
+package com.alibaba.datax.plugin.reader.oceanbasev10reader.util;
+
+import java.sql.Connection;
+import java.util.Collections;
+import java.util.List;
+
+import com.alibaba.datax.common.element.Record;
+
+public class TaskContext {
+    private Connection conn;
+    private final String table;
+    private String indexName;
+    // 辅助索引的字段列表
+    private List<String> secondaryIndexColumns = Collections.emptyList();
+    private String querySql;
+    private final String where;
+    private final int fetchSize;
+    private long readBatchSize = -1;
+    private boolean weakRead = true;
+    private String userSavePoint;
+    private String compatibleMode = ObReaderUtils.OB_COMPATIBLE_MODE_MYSQL;
+
+    public String getPartitionName() {
+        return partitionName;
+    }
+
+    public void setPartitionName(String partitionName) {
+        this.partitionName = partitionName;
+    }
+
+    private String partitionName;
+
+    // 断点续读的保存点
+    private volatile Record savePoint;
+
+    // pk在column中的index,用于绑定变量时从savePoint中读取值
+    // 如果这个值为null,则表示 不是断点续读的场景
+    private int[] pkIndexs;
+
+    private final List<String> columns;
+
+    private String[] pkColumns;
+
+    private long cost;
+
+    private final int transferColumnNumber;
+
+    public TaskContext(String table, List<String> columns, String where, int fetchSize) {
+        super();
+        this.table = table;
+        this.columns = columns;
+        // 针对只有querySql的场景
+        this.transferColumnNumber = columns == null ? -1 : columns.size();
+        this.where = where;
+        this.fetchSize = fetchSize;
+    }
+
+    public Connection getConn() {
+        return conn;
+    }
+
+    public void setConn(Connection conn) {
+        this.conn = conn;
+    }
+
+    public String getIndexName() {
+        return indexName;
+    }
+
+    public void setIndexName(String indexName) {
+        this.indexName = indexName;
+    }
+
+    public List<String> getSecondaryIndexColumns() {
+        return secondaryIndexColumns;
+    }
+
+    public void setSecondaryIndexColumns(List<String> secondaryIndexColumns) {
+        this.secondaryIndexColumns = secondaryIndexColumns;
+    }
+
+    public String getQuerySql() {
+        if (readBatchSize == -1 || ObReaderUtils.isOracleMode(compatibleMode)) {
+            return querySql;
+        } else {
+            return querySql + " limit " + readBatchSize;
+        }
+    }
+
+    public void setQuerySql(String querySql) {
+        this.querySql = querySql;
+    }
+
+    public String getWhere() {
+        return where;
+    }
+
+    public Record getSavePoint() {
+        return savePoint;
+    }
+
+    public void setSavePoint(Record savePoint) {
+        this.savePoint = savePoint;
+    }
+
+    public int[] getPkIndexs() {
+        return pkIndexs;
+    }
+
+    public void setPkIndexs(int[] pkIndexs) {
+        this.pkIndexs = pkIndexs;
+    }
+
+    public List<String> getColumns() {
+        return columns;
+    }
+
+    public String[] getPkColumns() {
+        return pkColumns;
+    }
+
+    public void setPkColumns(String[] pkColumns) {
+        this.pkColumns = pkColumns;
+    }
+
+    public String getTable() {
+        return table;
+    }
+
+    public int getFetchSize() {
+        return fetchSize;
+    }
+
+    public long getCost() {
+        return cost;
+    }
+
+    public void addCost(long cost) {
+        this.cost += cost;
+    }
+
+    public int getTransferColumnNumber() {
+        return transferColumnNumber;
+    }
+
+    public long getReadBatchSize() {
+        return readBatchSize;
+    }
+
+    public void setReadBatchSize(long readBatchSize) {
+        this.readBatchSize = readBatchSize;
+    }
+
+    public boolean getWeakRead() {
+        return weakRead;
+    }
+
+    public void setWeakRead(boolean weakRead) {
+        this.weakRead = weakRead;
+    }
+
+    public String getUserSavePoint() {
+        return userSavePoint;
+    }
+
+    public void setUserSavePoint(String userSavePoint) {
+        this.userSavePoint = userSavePoint;
+    }
+
+    public String getCompatibleMode() {
+        return compatibleMode;
+    }
+
+    public void setCompatibleMode(String compatibleMode) {
+        this.compatibleMode = compatibleMode;
+    }
+}

oceanbasev10reader/src/main/resources/plugin.json

@@ -0,0 +1,6 @@
+{
+    "name": "oceanbasev10reader",
+    "class": "com.alibaba.datax.plugin.reader.oceanbasev10reader.OceanBaseReader",
+    "description": "read data from oceanbase with SQL interface",
+    "developer": "oceanbase"
+}

oceanbasev10writer/pom.xml

@@ -0,0 +1,126 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>datax-all</artifactId>
+        <groupId>com.alibaba.datax</groupId>
+        <version>0.0.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>oceanbasev10writer</artifactId>
+
+    <groupId>com.alibaba.datax</groupId>
+    <version>0.0.1-SNAPSHOT</version>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.alibaba.datax</groupId>
+            <artifactId>datax-common</artifactId>
+            <version>${datax-project-version}</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>slf4j-log4j12</artifactId>
+                    <groupId>org.slf4j</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba.datax</groupId>
+            <artifactId>plugin-rdbms-util</artifactId>
+            <version>${datax-project-version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-test</artifactId>
+            <version>4.0.4.RELEASE</version>
+            <scope>test</scope>
+        </dependency>
+        <!-- 
+        <dependency>
+            <groupId>com.alipay.oceanbase</groupId>
+            <artifactId>oceanbase-partition</artifactId>
+            <version>0.0.5</version>
+        </dependency>
+       -->
+
+        <dependency>
+            <groupId>com.alipay.oceanbase</groupId>
+            <artifactId>oceanbase-connector-java</artifactId>
+            <version>3.2.0</version>
+            <scope>system</scope>
+            <systemPath>${basedir}/src/main/libs/oceanbase-connector-java-3.2.0.jar</systemPath>
+            <exclusions>
+            <exclusion>
+                <groupId>com.alipay.oceanbase</groupId>
+                <artifactId>oceanbase-client</artifactId>
+            </exclusion>
+            </exclusions>
+        </dependency>
+
+		<dependency>
+		  <groupId>log4j</groupId>
+		  <artifactId>log4j</artifactId>
+		  <version>1.2.16</version>
+		</dependency>
+		<dependency>
+      		<groupId>org.json</groupId>
+      		<artifactId>json</artifactId>
+      		<version>20160810</version>
+		</dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.11</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <resources>
+            <resource>
+                <directory>src/main/java</directory>
+                <includes>
+                    <include>**/*.properties</include>
+                </includes>
+            </resource>
+        </resources>
+        <plugins>
+            <!-- compiler plugin -->
+            <plugin>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>${jdk-version}</source>
+                    <target>${jdk-version}</target>
+                    <encoding>${project-sourceEncoding}</encoding>
+                </configuration>
+            </plugin>
+            <!-- assembly plugin -->
+            <plugin>
+                <artifactId>maven-assembly-plugin</artifactId>
+                <configuration>
+                    <descriptors>
+                        <descriptor>src/main/assembly/package.xml</descriptor>
+                    </descriptors>
+                    <finalName>datax</finalName>
+                </configuration>
+                <executions>
+                    <execution>
+                        <id>dwzip</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>single</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

oceanbasev10writer/src/main/assembly/package.xml

@@ -0,0 +1,42 @@
+<assembly
+        xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
+        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+        xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
+    <id></id>
+    <formats>
+        <format>dir</format>
+    </formats>
+    <includeBaseDirectory>false</includeBaseDirectory>
+    <fileSets>
+        <fileSet>
+            <directory>src/main/resources</directory>
+            <includes>
+                <include>plugin.json</include>
+ 				<include>plugin_job_template.json</include>
+ 			</includes>
+            <outputDirectory>plugin/writer/oceanbasev10writer</outputDirectory>
+        </fileSet>
+        <fileSet>
+            <directory>target/</directory>
+            <includes>
+                <include>oceanbasev10writer-0.0.1-SNAPSHOT.jar</include>
+            </includes>
+            <outputDirectory>plugin/writer/oceanbasev10writer</outputDirectory>
+        </fileSet>
+        <fileSet>
+            <directory>src/main/libs</directory>
+            <includes>
+                <include>*.jar</include>
+            </includes>
+            <outputDirectory>plugin/writer/oceanbasev10writer/libs</outputDirectory>
+        </fileSet>
+    </fileSets>
+
+    <dependencySets>
+        <dependencySet>
+            <useProjectArtifact>false</useProjectArtifact>
+            <outputDirectory>plugin/writer/oceanbasev10writer/libs</outputDirectory>
+            <scope>runtime</scope>
+        </dependencySet>
+    </dependencySets>
+</assembly>

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/Config.java

@@ -0,0 +1,62 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer;
+
+public interface Config {
+
+	String MEMSTORE_THRESHOLD = "memstoreThreshold";
+
+	double DEFAULT_MEMSTORE_THRESHOLD = 0.9d;
+
+	String MEMSTORE_CHECK_INTERVAL_SECOND = "memstoreCheckIntervalSecond";
+
+	long DEFAULT_MEMSTORE_CHECK_INTERVAL_SECOND = 30;
+
+	int DEFAULT_BATCH_SIZE = 100;
+	int MAX_BATCH_SIZE = 4096;
+
+	String FAIL_TRY_COUNT  = "failTryCount";
+
+	int DEFAULT_FAIL_TRY_COUNT = 10000;
+
+	String WRITER_THREAD_COUNT = "writerThreadCount";
+
+	int DEFAULT_WRITER_THREAD_COUNT = 1;
+
+	String CONCURRENT_WRITE = "concurrentWrite";
+
+	boolean DEFAULT_CONCURRENT_WRITE = true;
+
+	String OB_VERSION = "obVersion";
+	String TIMEOUT = "timeout";
+
+	String PRINT_COST = "printCost";
+	boolean DEFAULT_PRINT_COST = false;
+
+	String COST_BOUND = "costBound";
+	long DEFAULT_COST_BOUND = 20;
+
+	String MAX_ACTIVE_CONNECTION = "maxActiveConnection";
+	int DEFAULT_MAX_ACTIVE_CONNECTION = 2000;
+
+	String WRITER_SUB_TASK_COUNT = "writerSubTaskCount";
+	int DEFAULT_WRITER_SUB_TASK_COUNT = 1;
+	int MAX_WRITER_SUB_TASK_COUNT = 4096;
+
+	String OB_WRITE_MODE = "obWriteMode";
+    String OB_COMPATIBLE_MODE = "obCompatibilityMode";
+    String OB_COMPATIBLE_MODE_ORACLE = "ORACLE";
+    String OB_COMPATIBLE_MODE_MYSQL = "MYSQL";
+
+	String OCJ_GET_CONNECT_TIMEOUT = "ocjGetConnectTimeout";
+	int DEFAULT_OCJ_GET_CONNECT_TIMEOUT = 5000; // 5s
+
+	String OCJ_PROXY_CONNECT_TIMEOUT = "ocjProxyConnectTimeout";
+	int DEFAULT_OCJ_PROXY_CONNECT_TIMEOUT = 5000; // 5s
+
+	String OCJ_CREATE_RESOURCE_TIMEOUT = "ocjCreateResourceTimeout";
+	int DEFAULT_OCJ_CREATE_RESOURCE_TIMEOUT = 60000; // 60s
+
+	String OB_UPDATE_COLUMNS = "obUpdateColumns";
+
+	String USE_PART_CALCULATOR = "usePartCalculator";
+	boolean DEFAULT_USE_PART_CALCULATOR = false;
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/OceanBaseV10Writer.java

@@ -0,0 +1,258 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer;
+
+import com.alibaba.datax.common.plugin.RecordReceiver;
+import com.alibaba.datax.common.spi.Writer;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.datax.plugin.rdbms.writer.CommonRdbmsWriter;
+import com.alibaba.datax.plugin.rdbms.writer.Constant;
+import com.alibaba.datax.plugin.rdbms.writer.Key;
+import com.alibaba.datax.plugin.rdbms.writer.util.WriterUtil;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.task.ConcurrentTableWriterTask;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.DbUtils;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils;
+import com.alibaba.fastjson.JSONObject;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 2016-04-07
+ * <p>
+ * 专门针对OceanBase1.0的Writer
+ * 
+ * @author biliang.wbl
+ *
+ */
+public class OceanBaseV10Writer extends Writer {
+	private static DataBaseType DATABASE_TYPE = DataBaseType.OceanBase;
+
+	/**
+	 * Job 中的方法仅执行一次，Task 中方法会由框架启动多个 Task 线程并行执行。
+	 * <p/>
+	 * 整个 Writer 执行流程是：
+	 * 
+	 * <pre>
+	 * Job类init-->prepare-->split
+	 * 
+	 *                          Task类init-->prepare-->startWrite-->post-->destroy
+	 *                          Task类init-->prepare-->startWrite-->post-->destroy
+	 * 
+	 *                                                                            Job类post-->destroy
+	 * </pre>
+	 */
+	public static class Job extends Writer.Job {
+		private Configuration originalConfig = null;
+		private CommonRdbmsWriter.Job commonJob;
+		private static final Logger LOG = LoggerFactory.getLogger(Job.class);
+
+		/**
+		 * 注意：此方法仅执行一次。 最佳实践：通常在这里对用户的配置进行校验：是否缺失必填项？有无错误值？有没有无关配置项？...
+		 * 并给出清晰的报错/警告提示。校验通常建议采用静态工具类进行，以保证本类结构清晰。
+		 */
+		@Override
+		public void init() {
+			this.originalConfig = super.getPluginJobConf();
+			checkCompatibleMode(originalConfig);
+			//将config中的column和table中的关键字进行转义
+			List<String> columns = originalConfig.getList(Key.COLUMN, String.class);
+			ObWriterUtils.escapeDatabaseKeywords(columns);
+			originalConfig.set(Key.COLUMN, columns);
+
+			List<JSONObject> conns = originalConfig.getList(Constant.CONN_MARK, JSONObject.class);
+			for (int i = 0; i < conns.size(); i++) {
+				JSONObject conn = conns.get(i);
+				Configuration connConfig = Configuration.from(conn.toString());
+				List<String> tables = connConfig.getList(Key.TABLE, String.class);
+				ObWriterUtils.escapeDatabaseKeywords(tables);
+				originalConfig.set(String.format("%s[%d].%s", Constant.CONN_MARK, i, Key.TABLE), tables);
+			}
+			this.commonJob = new CommonRdbmsWriter.Job(DATABASE_TYPE);
+			this.commonJob.init(this.originalConfig);
+		}
+
+		/**
+		 * 注意：此方法仅执行一次。 最佳实践：如果 Job 中有需要进行数据同步之前的处理，可以在此处完成，如果没有必要则可以直接去掉。
+		 */
+		// 一般来说，是需要推迟到 task 中进行pre 的执行（单表情况例外）
+		@Override
+		public void prepare() {
+			int tableNumber = originalConfig.getInt(Constant.TABLE_NUMBER_MARK);
+			if (tableNumber == 1) {
+				this.commonJob.prepare(this.originalConfig);
+				final String version = fetchServerVersion(originalConfig);
+				originalConfig.set(Config.OB_VERSION, version);
+			}
+
+			String username = originalConfig.getString(Key.USERNAME);
+			String password = originalConfig.getString(Key.PASSWORD);
+
+			// 获取presql配置，并执行
+			List<String> preSqls = originalConfig.getList(Key.PRE_SQL, String.class);
+			if (preSqls == null || preSqls.size() == 0) {
+				return;
+			}
+
+			List<Object> conns = originalConfig.getList(Constant.CONN_MARK, Object.class);
+			for (Object connConfObject : conns) {
+				Configuration connConf = Configuration.from(connConfObject.toString());
+				// 这里的 jdbcUrl 已经 append 了合适后缀参数
+				String jdbcUrl = connConf.getString(Key.JDBC_URL);
+
+				List<String> tableList = connConf.getList(Key.TABLE, String.class);
+				for (String table : tableList) {
+					List<String> renderedPreSqls = WriterUtil.renderPreOrPostSqls(preSqls, table);
+					if (null != renderedPreSqls && !renderedPreSqls.isEmpty()) {
+						Connection conn = DBUtil.getConnection(DATABASE_TYPE, jdbcUrl, username, password);
+						LOG.info("Begin to execute preSqls:[{}]. context info:{}.",
+								StringUtils.join(renderedPreSqls, ";"), jdbcUrl);
+						WriterUtil.executeSqls(conn, renderedPreSqls, jdbcUrl, DATABASE_TYPE);
+						ObWriterUtils.asyncClose(null, null, conn);
+					}
+				}
+			}
+			if (LOG.isDebugEnabled()) {
+				LOG.debug("After job prepare(), originalConfig now is:[\n{}\n]", originalConfig.toJSON());
+			}
+		}
+
+		/**
+		 * 注意：此方法仅执行一次。 最佳实践：通常采用工具静态类完成把 Job 配置切分成多个 Task 配置的工作。 这里的
+		 * mandatoryNumber 是强制必须切分的份数。
+		 */
+		@Override
+		public List<Configuration> split(int mandatoryNumber) {
+			int tableNumber = originalConfig.getInt(Constant.TABLE_NUMBER_MARK);
+			if (tableNumber == 1) {
+				return this.commonJob.split(this.originalConfig, mandatoryNumber);
+			}
+			Configuration simplifiedConf = this.originalConfig;
+
+			List<Configuration> splitResultConfigs = new ArrayList<Configuration>();
+			for (int j = 0; j < mandatoryNumber; j++) {
+				splitResultConfigs.add(simplifiedConf.clone());
+			}
+			return splitResultConfigs;
+		}
+
+		/**
+		 * 注意：此方法仅执行一次。 最佳实践：如果 Job 中有需要进行数据同步之后的后续处理，可以在此处完成。
+		 */
+		@Override
+		public void post() {
+			int tableNumber = originalConfig.getInt(Constant.TABLE_NUMBER_MARK);
+			if (tableNumber == 1) {
+				commonJob.post(this.originalConfig);
+				return;
+			}
+			String username = originalConfig.getString(Key.USERNAME);
+			String password = originalConfig.getString(Key.PASSWORD);
+			List<Object> conns = originalConfig.getList(Constant.CONN_MARK, Object.class);
+			List<String> postSqls = originalConfig.getList(Key.POST_SQL, String.class);
+			if (postSqls == null || postSqls.size() == 0) {
+				return;
+			}
+
+			for (Object connConfObject : conns) {
+				Configuration connConf = Configuration.from(connConfObject.toString());
+				String jdbcUrl = connConf.getString(Key.JDBC_URL);
+				List<String> tableList = connConf.getList(Key.TABLE, String.class);
+
+				for (String table : tableList) {
+					List<String> renderedPostSqls = WriterUtil.renderPreOrPostSqls(postSqls, table);
+					if (null != renderedPostSqls && !renderedPostSqls.isEmpty()) {
+						// 说明有 postSql 配置，则此处删除掉
+						Connection conn = DBUtil.getConnection(DATABASE_TYPE, jdbcUrl, username, password);
+						LOG.info("Begin to execute postSqls:[{}]. context info:{}.",
+								StringUtils.join(renderedPostSqls, ";"), jdbcUrl);
+						WriterUtil.executeSqls(conn, renderedPostSqls, jdbcUrl, DATABASE_TYPE);
+						ObWriterUtils.asyncClose(null, null, conn);
+					}
+				}
+			}
+			originalConfig.remove(Key.POST_SQL);
+		}
+		
+		/**
+		 * 注意：此方法仅执行一次。 最佳实践：通常配合 Job 中的 post() 方法一起完成 Job 的资源释放。
+		 */
+		@Override
+		public void destroy() {
+			this.commonJob.destroy(this.originalConfig);
+		}
+
+		private String fetchServerVersion(Configuration config) {
+			final String fetchVersionSql = "show variables like 'version'";
+			return DbUtils.fetchSingleValueWithRetry(config, fetchVersionSql);
+		}
+
+		private void checkCompatibleMode(Configuration configure) {
+			final String fetchCompatibleModeSql = "SHOW VARIABLES LIKE 'ob_compatibility_mode'";
+			String compatibleMode = DbUtils.fetchSingleValueWithRetry(configure, fetchCompatibleModeSql);
+			ObWriterUtils.setCompatibleMode(compatibleMode);
+			configure.set(Config.OB_COMPATIBLE_MODE, compatibleMode);
+		}
+	}
+
+	public static class Task extends Writer.Task {
+		private static final Logger LOG = LoggerFactory.getLogger(Task.class);
+		private Configuration writerSliceConfig;
+		private CommonRdbmsWriter.Task writerTask;
+
+		/**
+		 * 注意：此方法每个 Task 都会执行一次。 最佳实践：此处通过对 taskConfig 配置的读取，进而初始化一些资源为
+		 * startWrite()做准备。
+		 */
+		@Override
+		public void init() {
+			this.writerSliceConfig = super.getPluginJobConf();
+			int tableNumber = writerSliceConfig.getInt(Constant.TABLE_NUMBER_MARK);
+			if (tableNumber == 1) {
+				// always use concurrentTableWriter
+				this.writerTask = new ConcurrentTableWriterTask(DATABASE_TYPE);
+			} else {
+				throw new RuntimeException("writing to multi-tables is not supported.");
+			}
+			LOG.info("tableNumber:" + tableNumber + ",writerTask Class:" + writerTask.getClass().getName());
+			this.writerTask.init(this.writerSliceConfig);
+		}
+
+		/**
+		 * 注意：此方法每个 Task 都会执行一次。 最佳实践：如果 Task
+		 * 中有需要进行数据同步之前的处理，可以在此处完成，如果没有必要则可以直接去掉。
+		 */
+		@Override
+		public void prepare() {
+			this.writerTask.prepare(this.writerSliceConfig);
+		}
+
+		/**
+		 * 注意：此方法每个 Task 都会执行一次。 最佳实践：此处适当封装确保简洁清晰完成数据写入工作。
+		 */
+		@Override
+		public void startWrite(RecordReceiver recordReceiver) {
+			this.writerTask.startWrite(recordReceiver, this.writerSliceConfig, super.getTaskPluginCollector());
+		}
+
+		/**
+		 * 注意：此方法每个 Task 都会执行一次。 最佳实践：如果 Task 中有需要进行数据同步之后的后续处理，可以在此处完成。
+		 */
+		@Override
+		public void post() {
+			this.writerTask.post(this.writerSliceConfig);
+		}
+
+		/**
+		 * 注意：此方法每个 Task 都会执行一次。 最佳实践：通常配合Task 中的 post() 方法一起完成 Task 的资源释放。
+		 */
+		@Override
+		public void destroy() {
+			this.writerTask.destroy(this.writerSliceConfig);
+		}
+	}
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/ConnHolder.java

@@ -0,0 +1,37 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.ext;
+
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+
+import java.sql.Connection;
+
+public abstract class ConnHolder {
+
+    protected final Configuration config;
+    protected Connection conn;
+
+    public ConnHolder(Configuration config) {
+        this.config = config;
+    }
+
+    public abstract Connection initConnection();
+
+    public Configuration getConfig() {
+        return config;
+    }
+
+    public Connection getConn() {
+        return conn;
+    }
+
+    public Connection reconnect() {
+        DBUtil.closeDBResources(null, conn);
+        return initConnection();
+    }
+
+    public abstract String getJdbcUrl();
+
+    public abstract String getUserName();
+
+    public abstract void destroy();
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/DataBaseWriterBuffer.java

@@ -0,0 +1,101 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.ext;
+
+import java.sql.Connection;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.util.DBUtilErrorCode;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * 
+ * @author oceanbase
+ *
+ */
+public class DataBaseWriterBuffer {
+	private static final Logger LOG = LoggerFactory.getLogger(DataBaseWriterBuffer.class);
+
+	private final ConnHolder connHolder;
+	private final String dbName;
+	private Map<String, LinkedList<Record>> tableBuffer = new HashMap<String, LinkedList<Record>>();
+	private long lastCheckMemstoreTime;
+
+	public DataBaseWriterBuffer(Configuration config,String jdbcUrl, String userName, String password,String dbName){
+		this.connHolder = new ObClientConnHolder(config, jdbcUrl, userName, password);
+		this.dbName=dbName;
+	}
+	
+	public ConnHolder getConnHolder(){
+		return connHolder;
+	}
+
+	public void initTableBuffer(List<String> tableList) {
+		for (String table : tableList) {
+			tableBuffer.put(table, new LinkedList<Record>());
+		}
+	}
+	
+	public List<String> getTableList(){
+		return new ArrayList<String>(tableBuffer.keySet());
+	}
+
+	public void addRecord(Record record, String tableName) {
+		LinkedList<Record> recordList = tableBuffer.get(tableName);
+		if (recordList == null) {
+			throw DataXException.asDataXException(DBUtilErrorCode.WRITE_DATA_ERROR,
+					String.format("The [table] calculated based on the rules does not exist. The calculated [tableName]=%s, [db]=%s. Please check the rules you configured.",
+							tableName, connHolder.getJdbcUrl()));
+		}
+		recordList.add(record);
+	}
+
+	public Map<String, LinkedList<Record>> getTableBuffer() {
+		return tableBuffer;
+	}
+
+	public String getDbName() {
+		return dbName;
+	}
+
+	public long getLastCheckMemstoreTime() {
+		return lastCheckMemstoreTime;
+	}
+
+	public void setLastCheckMemstoreTime(long lastCheckMemstoreTime) {
+		this.lastCheckMemstoreTime = lastCheckMemstoreTime;
+	}
+
+	/**
+	 * 检查当前DB的memstore使用状态
+	 * <p>
+	 * 若超过阈值,则休眠
+	 *
+	 * @param memstoreCheckIntervalSecond
+	 * @param memstoreThreshold
+	 */
+	public synchronized void checkMemstore(long memstoreCheckIntervalSecond, double memstoreThreshold) {
+		long now = System.currentTimeMillis();
+		if (now - getLastCheckMemstoreTime() < 1000 * memstoreCheckIntervalSecond) {
+			return;
+		}
+
+		LOG.debug(String.format("checking memstore usage: lastCheckTime=%d, now=%d, check interval=%d, threshold=%f",
+				getLastCheckMemstoreTime(), now, memstoreCheckIntervalSecond, memstoreThreshold));
+
+		Connection conn = getConnHolder().getConn();
+		while (ObWriterUtils.isMemstoreFull(conn, memstoreThreshold)) {
+			LOG.warn("OB memstore is full,sleep 60 seconds, jdbc=" + getConnHolder().getJdbcUrl()
+					+ ",threshold=" + memstoreThreshold);
+			ObWriterUtils.sleep(60000);
+		}
+		setLastCheckMemstoreTime(now);
+	}
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/OBDataSourceV10.java

@@ -0,0 +1,190 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.ext;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import com.alibaba.datax.plugin.rdbms.reader.Key;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.Config;
+import com.alipay.oceanbase.obproxy.datasource.ObGroupDataSource;
+import com.alipay.oceanbase.obproxy.exception.ConnectionPropertiesNotSupportedException;
+import com.alipay.oceanbase.obproxy.util.StringParser.IllegalFormatException;
+import com.google.common.collect.Maps;
+
+public class OBDataSourceV10 {
+	private static final Logger LOG = LoggerFactory.getLogger(OBDataSourceV10.class);
+
+	private static final Map<String, DataSourceHolder> dataSources = Maps.newHashMap();
+
+	private static int ocjGetConnectionTimeout = 0;
+	private static int ocjGlobalProxyroGetConnectionTimeout = 0;
+	private static int ocjMaxWaitOfCreateClusterResourceMs = 0;
+
+	private static Configuration taskConfig;
+
+	public static String genKey(String fullUserName, String dbName) {
+		//username@tenantName#clusterName/dbName
+		return fullUserName + "/" + dbName;
+	}
+	
+	public static synchronized void init(Configuration configuration,
+			final String fullUsername,
+			final String password,
+			final String dbName) {
+		taskConfig = configuration;
+		final String rsUrl = "";
+		final String dataSourceKey = genKey(fullUsername, dbName);
+		final int maxActiveConnection = configuration.getInt(Config.MAX_ACTIVE_CONNECTION, Config.DEFAULT_MAX_ACTIVE_CONNECTION);
+		if (dataSources.containsKey(dataSourceKey)) {
+			dataSources.get(dataSourceKey).increseRefercnce();
+		} else {
+			long timeout = configuration.getInt(Config.TIMEOUT, 30);
+			if (timeout < 30) {
+				timeout = 30;
+			}
+			if (ocjGetConnectionTimeout == 0) {
+				ocjGetConnectionTimeout = configuration.getInt(Config.OCJ_GET_CONNECT_TIMEOUT,
+						Config.DEFAULT_OCJ_GET_CONNECT_TIMEOUT);
+				ocjGlobalProxyroGetConnectionTimeout = configuration.getInt(Config.OCJ_PROXY_CONNECT_TIMEOUT,
+						Config.DEFAULT_OCJ_PROXY_CONNECT_TIMEOUT);
+				ocjMaxWaitOfCreateClusterResourceMs = configuration.getInt(Config.OCJ_CREATE_RESOURCE_TIMEOUT,
+						Config.DEFAULT_OCJ_CREATE_RESOURCE_TIMEOUT);
+
+				LOG.info(String.format("initializing OCJ with ocjGetConnectionTimeout=%d, " +
+								"ocjGlobalProxyroGetConnectionTimeout=%d, ocjMaxWaitOfCreateClusterResourceMs=%d",
+						ocjGetConnectionTimeout, ocjGlobalProxyroGetConnectionTimeout, ocjMaxWaitOfCreateClusterResourceMs));
+			}
+			DataSourceHolder holder = null;
+			try {
+				holder = new DataSourceHolder(rsUrl, fullUsername, password, dbName, maxActiveConnection, timeout);
+				dataSources.put(dataSourceKey, holder);
+			} catch (ConnectionPropertiesNotSupportedException e) {
+				e.printStackTrace();
+				throw new DataXException(ObDataSourceErrorCode.DESC, "connect error");
+			} catch (IllegalArgumentException e) {
+				e.printStackTrace();
+				throw new DataXException(ObDataSourceErrorCode.DESC, "connect error");
+			} catch (IllegalFormatException e) {
+				e.printStackTrace();
+				throw new DataXException(ObDataSourceErrorCode.DESC, "connect error");
+			} catch (SQLException e) {
+				e.printStackTrace();
+				throw new DataXException(ObDataSourceErrorCode.DESC, "connect error");
+			}
+		}
+	}
+
+	public static synchronized void destory(final String dataSourceKey){
+		DataSourceHolder holder = dataSources.get(dataSourceKey);
+		holder.decreaseReference();
+		if (holder.canClose()) {
+			dataSources.remove(dataSourceKey);
+			holder.close();
+			LOG.info(String.format("close datasource success [%s]", dataSourceKey));
+		}
+	}
+	
+	public static Connection getConnection(final String url) {
+		Connection conn = null;
+		try {
+			conn = dataSources.get(url).getconnection();
+		} catch (SQLException e) {
+			e.printStackTrace();
+		}
+		return conn;
+	}
+
+	private static Map<String, String> buildJdbcProperty() {
+	    Map<String, String> property = new HashMap<String, String>();
+	    property.put("useServerPrepStmts", "false");
+	    property.put("characterEncoding", "UTF-8");
+	    property.put("useLocalSessionState", "false");
+	    property.put("rewriteBatchedStatements", "true");
+	    property.put("socketTimeout", "25000");
+
+	    return property;
+	}
+
+	private static class DataSourceHolder {
+		private volatile int reference;
+		private final ObGroupDataSource groupDataSource;
+		public static final Map<String, String> jdbcProperty = buildJdbcProperty();;
+
+		public DataSourceHolder(final String rsUrl,
+				final String fullUsername,
+				final String password,
+				final String dbName,
+				final int maxActive,
+				final long timeout) throws ConnectionPropertiesNotSupportedException, IllegalFormatException, IllegalArgumentException, SQLException {
+			this.reference = 1;
+			this.groupDataSource = new ObGroupDataSource();
+		    this.groupDataSource.setUrl(rsUrl);
+		    this.groupDataSource.setFullUsername(fullUsername);
+		    this.groupDataSource.setPassword(password);
+		    this.groupDataSource.setDatabase(dbName);
+		    this.groupDataSource.setConnectionProperties(jdbcProperty);
+		    this.groupDataSource.setGetConnectionTimeout(ocjGetConnectionTimeout);
+		    this.groupDataSource.setGlobalProxyroGetConnectionTimeout(ocjGlobalProxyroGetConnectionTimeout);
+		    this.groupDataSource.setMaxWaitOfCreateClusterResourceMs(ocjMaxWaitOfCreateClusterResourceMs);
+		    this.groupDataSource.setMaxActive(maxActive);
+		    this.groupDataSource.setGlobalSlowQueryThresholdUs(3000000);      // 3s, sql with response time more than 3s will be logged
+		    this.groupDataSource.setGlobalCleanLogFileEnabled(true);          // enable log cleanup
+		    this.groupDataSource.setGlobalLogFileSizeThreshold(17179869184L); // 16G, log file total size
+		    this.groupDataSource.setGlobalCleanLogFileInterval(10000);        // 10s, check interval
+		    this.groupDataSource.setInitialSize(1);
+
+			List<String> initSqls = new ArrayList<String>();
+			if (taskConfig != null) {
+				List<String> sessionConfig = taskConfig.getList(Key.SESSION, new ArrayList(), String.class);
+				if (sessionConfig != null || sessionConfig.size() > 0) {
+					initSqls.addAll(sessionConfig);
+				}
+			}
+            // set up for writing timestamp columns
+			if (ObWriterUtils.isOracleMode()) {
+				initSqls.add("ALTER SESSION SET NLS_DATE_FORMAT='YYYY-MM-DD HH24:MI:SS';");
+				initSqls.add("ALTER SESSION SET NLS_TIMESTAMP_FORMAT='YYYY-MM-DD HH24:MI:SS.FF';");
+				initSqls.add("ALTER SESSION SET NLS_TIMESTAMP_TZ_FORMAT='YYYY-MM-DD HH24:MI:SS.FF TZR TZD';");
+			}
+
+			this.groupDataSource.setConnectionInitSqls(initSqls);
+
+		    this.groupDataSource.init();
+			// this.groupDataSource;
+		    LOG.info("Create GroupDataSource rsUrl=[{}], fullUserName=[{}], dbName=[{}], getConnectionTimeout= {}ms, maxActive={}",
+		    		rsUrl, fullUsername, dbName, 5000, maxActive);
+		}
+		
+		public Connection getconnection() throws SQLException {
+			return groupDataSource.getConnection();
+		}
+
+		public synchronized void increseRefercnce() {
+			this.reference++;
+		}
+
+		public synchronized void decreaseReference() {
+			this.reference--;
+		}
+
+		public synchronized boolean canClose() {
+			return reference == 0;
+		}
+
+		public synchronized void close() {
+			if (this.canClose()) {
+				groupDataSource.destroy();
+			}
+		}
+	}
+
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/OCJConnHolder.java

@@ -0,0 +1,55 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.ext;
+
+import java.sql.Connection;
+
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+
+/**
+ * wrap oceanbase java client
+ * @author oceanbase
+ */
+
+public class OCJConnHolder extends ConnHolder {
+	private ServerConnectInfo connectInfo;
+	private String dataSourceKey;
+
+	public OCJConnHolder (Configuration config, ServerConnectInfo connInfo) {
+		super(config);
+		this.connectInfo = connInfo;
+		this.dataSourceKey = OBDataSourceV10.genKey(connectInfo.getFullUserName(), connectInfo.databaseName);
+		OBDataSourceV10.init(config, connectInfo.getFullUserName(), connectInfo.password, connectInfo.databaseName);
+	}
+
+	@Override
+	public Connection initConnection() {
+		conn = OBDataSourceV10.getConnection(dataSourceKey);
+		return conn;
+	}
+
+	@Override
+	public Connection reconnect() {
+		DBUtil.closeDBResources(null, conn);
+		return initConnection();
+	}
+
+	@Override
+	public Connection getConn() {
+		return conn;
+	}
+
+	@Override
+	public String getJdbcUrl() {
+		return connectInfo.jdbcUrl;
+	}
+
+	@Override
+	public String getUserName() {
+		return connectInfo.userName;
+	}
+	
+	public void destroy() {
+		OBDataSourceV10.destory(this.dataSourceKey);
+	}
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/ObClientConnHolder.java

@@ -0,0 +1,63 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.ext;
+
+import java.sql.Connection;
+import java.util.ArrayList;
+import java.util.List;
+
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.reader.Key;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils;
+
+/**
+ * 数据库连接代理对象,负责创建连接，重新连接
+ *
+ * @author oceanbase
+ *
+ */
+public class ObClientConnHolder extends ConnHolder {
+	private final String jdbcUrl;
+	private final String userName;
+	private final String password;
+
+	public ObClientConnHolder(Configuration config, String jdbcUrl, String userName, String password) {
+		super(config);
+		this.jdbcUrl = jdbcUrl;
+		this.userName = userName;
+		this.password = password;
+	}
+
+	// Connect to ob with obclient and obproxy
+	@Override
+	public Connection initConnection() {
+		String BASIC_MESSAGE = String.format("jdbcUrl:[%s]", this.jdbcUrl);
+		DataBaseType dbType = DataBaseType.OceanBase;
+		if (ObWriterUtils.isOracleMode()) {
+		    // set up for writing timestamp columns
+		    List<String> sessionConfig = config.getList(Key.SESSION, new ArrayList<String>(), String.class);
+			sessionConfig.add("ALTER SESSION SET NLS_DATE_FORMAT='YYYY-MM-DD HH24:MI:SS'");
+			sessionConfig.add("ALTER SESSION SET NLS_TIMESTAMP_FORMAT='YYYY-MM-DD HH24:MI:SS.FF'");
+			sessionConfig.add("ALTER SESSION SET NLS_TIMESTAMP_TZ_FORMAT='YYYY-MM-DD HH24:MI:SS.FF TZR TZD'");
+		    config.set(Key.SESSION, sessionConfig);
+		}
+		conn = DBUtil.getConnection(dbType, jdbcUrl, userName, password);
+		DBUtil.dealWithSessionConfig(conn, config, dbType, BASIC_MESSAGE);
+		return conn;
+	}
+
+	@Override
+	public String getJdbcUrl() {
+		return jdbcUrl;
+	}
+
+	@Override
+	public String getUserName() {
+		return userName;
+	}
+
+	@Override
+	public void destroy() {
+		DBUtil.closeDBResources(null, conn);
+	}
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/ObDataSourceErrorCode.java

@@ -0,0 +1,31 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.ext;
+
+import com.alibaba.datax.common.spi.ErrorCode;
+
+public enum ObDataSourceErrorCode implements ErrorCode {
+	DESC("ObDataSourceError code","connect error");
+
+    private final String code;
+    private final String describe;
+
+    private ObDataSourceErrorCode(String code, String describe) {
+        this.code = code;
+        this.describe = describe;
+    }
+
+    @Override
+    public String getCode() {
+        return this.code;
+    }
+
+    @Override
+    public String getDescription() {
+        return this.describe;
+    }
+
+    @Override
+    public String toString() {
+        return String.format("Code:[%s], Describe:[%s]. ", this.code,
+                this.describe);
+    }
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/ext/ServerConnectInfo.java

@@ -0,0 +1,79 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.ext;
+
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public class ServerConnectInfo {
+	
+	public String clusterName;
+	public String tenantName;
+	public String userName;
+	public String password;
+	public String databaseName;
+	public String ipPort;
+	public String jdbcUrl;
+
+	public ServerConnectInfo(final String jdbcUrl, final String username, final String password) {
+		if (jdbcUrl.startsWith(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING)) {
+			String[] ss = jdbcUrl.split(com.alibaba.datax.plugin.rdbms.writer.Constant.OB10_SPLIT_STRING_PATTERN);
+			if (ss.length != 3) {
+				throw new RuntimeException("jdbc url format is not correct: " + jdbcUrl);
+			}
+			this.userName = username;
+			this.clusterName = ss[1].trim().split(":")[0];
+			this.tenantName  = ss[1].trim().split(":")[1];
+			this.jdbcUrl = ss[2].replace("jdbc:mysql:", "jdbc:oceanbase:");
+		} else {
+			this.jdbcUrl = jdbcUrl.replace("jdbc:mysql:", "jdbc:oceanbase:");
+			if (username.contains("@") && username.contains("#")) {
+				this.userName = username.substring(0, username.indexOf("@"));
+				this.tenantName = username.substring(username.indexOf("@") + 1, username.indexOf("#"));
+				this.clusterName = username.substring(username.indexOf("#") + 1);
+			} else if (username.contains(":")) {
+				String[] config = username.split(":");
+				if (config.length != 3) {
+					throw new RuntimeException ("username format is not correct: " + username);
+				}
+				this.clusterName = config[0];
+				this.tenantName = config[1];
+				this.userName = config[2];
+			} else {
+				this.clusterName = null;
+				this.tenantName = null;
+				this.userName = username;
+			}
+		}
+
+		this.password = password;
+		parseJdbcUrl(jdbcUrl);
+	}
+
+	private void parseJdbcUrl(final String jdbcUrl) {
+		Pattern pattern = Pattern.compile("//([\\w\\.\\-]+:\\d+)/([\\w-]+)\\?");
+		Matcher matcher = pattern.matcher(jdbcUrl);
+		if (matcher.find()) {
+			String ipPort = matcher.group(1);
+			String dbName = matcher.group(2);
+			this.ipPort = ipPort;
+			this.databaseName = dbName;
+		} else {
+			throw new RuntimeException("Invalid argument:" + jdbcUrl);
+		}
+	}
+
+	public String toString() {
+		StringBuffer strBuffer = new StringBuffer();
+		return strBuffer.append("clusterName:").append(clusterName).append(", tenantName:").append(tenantName)
+				.append(", userName:").append(userName).append(", databaseName:").append(databaseName)
+				.append(", ipPort:").append(ipPort).append(", jdbcUrl:").append(jdbcUrl).toString();
+	}
+
+	public String getFullUserName() {
+		StringBuilder builder = new StringBuilder(userName);
+		if (tenantName != null && clusterName != null) {
+			builder.append("@").append(tenantName).append("#").append(clusterName);
+		}
+
+		return builder.toString();
+	}
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/ColumnMetaCache.java

@@ -0,0 +1,41 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.task;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.List;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Triple;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+
+public class ColumnMetaCache {
+    private static final Logger LOG = LoggerFactory.getLogger(ColumnMetaCache.class);
+	
+	private static String tableName;
+	private static Triple<List<String>, List<Integer>, List<String>> columnMeta = null;
+
+	public ColumnMetaCache() {
+		
+	}
+	
+	public static void init(Connection connection, final String tableName, final List<String> columns) throws SQLException {
+		if (columnMeta == null) {
+			synchronized(ColumnMetaCache.class) {
+				ColumnMetaCache.tableName = tableName;
+				if (columnMeta == null) {
+					columnMeta = DBUtil.getColumnMetaData(connection,
+            				tableName, StringUtils.join(columns, ","));
+					LOG.info("fetch columnMeta of table {} success", tableName);
+				}
+			}
+		}
+	}
+	
+	public static Triple<List<String>, List<Integer>, List<String>> getColumnMeta() {
+		return columnMeta;
+	}
+	
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/ConcurrentTableWriterTask.java

@@ -0,0 +1,522 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.task;
+
+import com.alibaba.datax.common.element.Column;
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.common.plugin.RecordReceiver;
+import com.alibaba.datax.common.plugin.TaskPluginCollector;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.DBUtilErrorCode;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.datax.plugin.rdbms.writer.CommonRdbmsWriter;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.Config;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ConnHolder;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ObClientConnHolder;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ServerConnectInfo;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils;
+import com.alipay.oceanbase.obproxy.data.TableEntryKey;
+import com.alipay.oceanbase.obproxy.util.ObPartitionIdCalculator;
+import org.apache.commons.lang3.tuple.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+//import java.sql.PreparedStatement;
+
+public class ConcurrentTableWriterTask extends CommonRdbmsWriter.Task {
+    private static final Logger LOG = LoggerFactory.getLogger(ConcurrentTableWriterTask.class);
+
+	// memstore_total 与 memstore_limit 比例的阈值,一旦超过这个值,则暂停写入
+	private double memstoreThreshold = Config.DEFAULT_MEMSTORE_THRESHOLD;
+	// memstore检查的间隔
+	private long memstoreCheckIntervalSecond = Config.DEFAULT_MEMSTORE_CHECK_INTERVAL_SECOND;
+	// 最后一次检查
+	private long lastCheckMemstoreTime;
+    
+    private static AtomicLong totalTask = new AtomicLong(0);
+    private long taskId = -1;
+    
+    private AtomicBoolean isMemStoreFull = new AtomicBoolean(false);
+    private ConnHolder checkConnHolder;
+
+    public ConcurrentTableWriterTask(DataBaseType dataBaseType) {
+		super(dataBaseType);
+		taskId = totalTask.getAndIncrement();
+	}
+
+	private ObPartitionIdCalculator partCalculator = null;
+
+	private HashMap<Long, List<Record>> groupInsertValues;
+	List<Record> unknownPartRecords = new ArrayList<Record>();
+//	private List<Record> unknownPartRecords;
+	private List<Integer> partitionKeyIndexes;
+	
+	private ConcurrentTableWriter concurrentWriter = null;
+	
+	private ConnHolder connHolder;
+	
+	private boolean allTaskInQueue = false;
+	
+	private Lock lock = new ReentrantLock();
+	private Condition condition = lock.newCondition();
+	
+	private long startTime;
+	private boolean isOb2 = false;
+	private String obWriteMode = "update";
+	private boolean isOracleCompatibleMode = false;
+	private String obUpdateColumns = null;
+	private List<Pair<String, int[]>> deleteColPos;
+	private String dbName;
+
+	@Override
+	public void init(Configuration config) {
+		super.init(config);
+		// OceanBase 所有操作都是 insert into on duplicate key update 模式
+		// writeMode应该使用enum来定义
+		this.writeMode = "update";
+        obWriteMode = config.getString(Config.OB_WRITE_MODE, "update");
+		ServerConnectInfo connectInfo = new ServerConnectInfo(jdbcUrl, username, password);
+		dbName = connectInfo.databaseName;
+		//init check memstore
+		this.memstoreThreshold = config.getDouble(Config.MEMSTORE_THRESHOLD, Config.DEFAULT_MEMSTORE_THRESHOLD);
+		this.memstoreCheckIntervalSecond = config.getLong(Config.MEMSTORE_CHECK_INTERVAL_SECOND,
+				Config.DEFAULT_MEMSTORE_CHECK_INTERVAL_SECOND);
+        this.isOracleCompatibleMode = ObWriterUtils.isOracleMode();
+
+        LOG.info("configure url is unavailable, use obclient for connections.");
+        this.checkConnHolder = new ObClientConnHolder(config, connectInfo.jdbcUrl,
+				connectInfo.getFullUserName(), connectInfo.password);
+        this.connHolder = new ObClientConnHolder(config, connectInfo.jdbcUrl,
+				connectInfo.getFullUserName(), connectInfo.password);
+        checkConnHolder.initConnection();
+        if (isOracleCompatibleMode) {
+			connectInfo.databaseName = connectInfo.databaseName.toUpperCase();
+			//在转义的情况下不翻译
+			if (!(table.startsWith("\"") && table.endsWith("\""))) {
+				table = table.toUpperCase();
+			}
+
+			LOG.info(String.format("this is oracle compatible mode, change database to %s, table to %s",
+					connectInfo.databaseName, table));
+        }
+
+        if (config.getBool(Config.USE_PART_CALCULATOR, Config.DEFAULT_USE_PART_CALCULATOR)) {
+            initPartCalculator(connectInfo);
+        } else {
+            LOG.info("Disable partition calculation feature.");
+        }
+
+		obUpdateColumns = config.getString(Config.OB_UPDATE_COLUMNS, null);
+		groupInsertValues = new HashMap<Long, List<Record>>();
+		partitionKeyIndexes = new ArrayList<Integer>();
+		rewriteSql();
+
+		if (null == concurrentWriter) {
+			concurrentWriter = new ConcurrentTableWriter(config, connectInfo, writeRecordSql);
+			allTaskInQueue = false;
+		}
+
+		String version = config.getString(Config.OB_VERSION);
+		int pIdx = version.lastIndexOf('.');
+		if ((Float.valueOf(version.substring(0, pIdx)) >= 2.1f)) {
+			isOb2 = true;
+		}
+	}
+
+	private void initPartCalculator(ServerConnectInfo connectInfo) {
+		int retry = 0;
+		LOG.info(String.format("create tableEntryKey with clusterName %s, tenantName %s, databaseName %s, tableName %s",
+				connectInfo.clusterName, connectInfo.tenantName, connectInfo.databaseName, table));
+		TableEntryKey tableEntryKey = new TableEntryKey(connectInfo.clusterName, connectInfo.tenantName,
+				connectInfo.databaseName, table);
+		do {
+			try {
+				if (retry > 0) {
+					int sleep = retry > 8 ? 500 : (1 << retry);
+					TimeUnit.SECONDS.sleep(sleep);
+					LOG.info("retry create new part calculator, the {} times", retry);
+				}
+				LOG.info("create partCalculator with address: " + connectInfo.ipPort);
+				partCalculator = new ObPartitionIdCalculator(connectInfo.ipPort, tableEntryKey);
+			} catch (Exception ex) {
+				++retry;
+				LOG.warn("create new part calculator failed, retry {}: {}", retry, ex.getMessage());
+			}
+		} while (partCalculator == null && retry < 3); // try 3 times
+	}
+
+	public boolean isFinished() {
+		return allTaskInQueue && concurrentWriter.checkFinish();
+	}
+	
+	public boolean allTaskInQueue() {
+		return allTaskInQueue;
+	}
+	
+	public void setPutAllTaskInQueue() {
+		this.allTaskInQueue = true;
+		LOG.info("ConcurrentTableWriter has put all task in queue, queueSize = {},  total = {}, finished = {}",
+				concurrentWriter.getTaskQueueSize(),
+				concurrentWriter.getTotalTaskCount(),
+				concurrentWriter.getFinishTaskCount());
+	}
+	
+	private void rewriteSql() {
+		Connection conn = connHolder.initConnection();
+		if (isOracleCompatibleMode && obWriteMode.equalsIgnoreCase("update")) {
+			// change obWriteMode to insert so the insert statement will be generated.
+			obWriteMode = "insert";
+			deleteColPos = ObWriterUtils.buildDeleteSql(conn, dbName, table, columns);
+		}
+		this.writeRecordSql = ObWriterUtils.buildWriteSql(table, columns, conn, obWriteMode, obUpdateColumns);
+		LOG.info("writeRecordSql :{}", this.writeRecordSql);
+	}
+	
+	public void prepare(Configuration writerSliceConfig) {
+		super.prepare(writerSliceConfig);
+		calPartitionKeyIndex(partitionKeyIndexes);
+		concurrentWriter.start();
+	}
+
+	private void calPartitionKeyIndex(List<Integer> partKeyIndexes) {
+		partKeyIndexes.clear();
+		if (null == partCalculator) {
+			LOG.error("partCalculator is null");
+			return;
+		}
+		for (int i = 0; i < columns.size(); ++i) {
+			if (partCalculator.isPartitionKeyColumn(columns.get(i))) {
+			    LOG.info(columns.get(i) + " is partition key.");
+				partKeyIndexes.add(i);
+			}
+		}
+	}
+
+	private Long calPartitionId(List<Integer> partKeyIndexes, Record record) {
+	    if (partCalculator == null) {
+	        return null;
+	    }
+		for (Integer i : partKeyIndexes) {
+			partCalculator.addColumn(columns.get(i), record.getColumn(i).asString());
+		}
+		return partCalculator.calculate();
+	}
+	
+	@Override
+    public void startWriteWithConnection(RecordReceiver recordReceiver, TaskPluginCollector taskPluginCollector, Connection connection) {
+        this.taskPluginCollector = taskPluginCollector;
+
+        // 用于写入数据的时候的类型根据目的表字段类型转换
+        int retryTimes = 0;
+        boolean needRetry = false;
+        do {
+        	try {
+        		if (retryTimes > 0) {
+        			TimeUnit.SECONDS.sleep((1 << retryTimes));
+    				DBUtil.closeDBResources(null, connection);
+        			connection = DBUtil.getConnection(dataBaseType, jdbcUrl, username, password);
+        			LOG.warn("getColumnMetaData of table {} failed, retry the {} times ...", this.table, retryTimes);
+        		}
+        		ColumnMetaCache.init(connection, this.table, this.columns);
+        		this.resultSetMetaData = ColumnMetaCache.getColumnMeta();
+            	needRetry = false;
+        	} catch (SQLException e) {
+        		needRetry = true;
+        		++retryTimes;
+        		e.printStackTrace();
+        		LOG.warn("fetch column meta of [{}] failed..., retry {} times", this.table, retryTimes);
+        	} catch (InterruptedException e) {
+				LOG.warn("startWriteWithConnection interrupt, ignored");
+			} finally {
+        	}
+        } while (needRetry && retryTimes < 100);
+
+        try {
+            Record record;
+            startTime = System.currentTimeMillis();
+            while ((record = recordReceiver.getFromReader()) != null) {
+                if (record.getColumnNumber() != this.columnNumber) {
+                    // 源头读取字段列数与目的表字段写入列数不相等，直接报错
+                	LOG.error("column not equal {} != {}, record = {}",
+                			this.columnNumber, record.getColumnNumber(), record.toString());
+                    throw DataXException
+                            .asDataXException(
+                                    DBUtilErrorCode.CONF_ERROR,
+                                    String.format("Recoverable exception in OB. Roll back this write and hibernate for one minute. SQLState: %d. ErrorCode: %d",
+                                            record.getColumnNumber(),
+                                            this.columnNumber));
+                }
+                addRecordToCache(record);
+            }
+            addLeftRecords();
+            waitTaskFinish();
+        } catch (Exception e) {
+            throw DataXException.asDataXException(
+                    DBUtilErrorCode.WRITE_DATA_ERROR, e);
+        } finally {
+            DBUtil.closeDBResources(null, null, connection);
+        }
+    }
+
+	public PreparedStatement fillStatement(PreparedStatement preparedStatement, Record record)
+			throws SQLException {
+		return fillPreparedStatement(preparedStatement, record);
+	}
+
+    public PreparedStatement fillStatementIndex(PreparedStatement preparedStatement,
+                                                int prepIdx, int columnIndex, Column column) throws SQLException {
+        int columnSqltype = this.resultSetMetaData.getMiddle().get(columnIndex);
+        String typeName = this.resultSetMetaData.getRight().get(columnIndex);
+        return fillPreparedStatementColumnType(preparedStatement, prepIdx, columnSqltype, typeName, column);
+    }
+
+    public void collectDirtyRecord(Record record, SQLException e) {
+		taskPluginCollector.collectDirtyRecord(record, e);
+	}
+
+	public void insertOneRecord(Connection connection, List<Record> buffer) {
+		doOneInsert(connection, buffer);
+	}
+
+	private void addLeftRecords() {
+		//不需要刷新Cache，已经是最后一批数据了
+		for (List<Record> groupValues : groupInsertValues.values()) {
+			if (groupValues.size() > 0 ) {
+				addRecordsToWriteQueue(groupValues);
+			}
+		}
+		if (unknownPartRecords.size() > 0) {
+			addRecordsToWriteQueue(unknownPartRecords);
+		}
+	}
+	
+	private void addRecordToCache(final Record record) {
+		Long partId =null;
+		try {
+			partId = calPartitionId(partitionKeyIndexes, record);
+		} catch (Exception e1) {
+		    LOG.warn("fail to get partition id: " + e1.getMessage() + ", record: " + record);
+		}
+
+        if (partId == null && isOb2) {
+            LOG.debug("fail to calculate parition id, just put into the default buffer.");
+            partId = Long.MAX_VALUE;
+        }
+
+		if (partId != null) {
+			List<Record> groupValues = groupInsertValues.get(partId);
+			if (groupValues == null) {
+				groupValues = new ArrayList<Record>(batchSize);
+				groupInsertValues.put(partId, groupValues);
+			}
+			groupValues.add(record);
+			if (groupValues.size() >= batchSize) {
+				groupValues = addRecordsToWriteQueue(groupValues);
+				groupInsertValues.put(partId, groupValues);
+			}
+		} else {
+			LOG.debug("add unknown part record {}", record);
+			unknownPartRecords.add(record);
+			if (unknownPartRecords.size() >= batchSize) {
+				unknownPartRecords = addRecordsToWriteQueue(unknownPartRecords);
+			}
+
+		}
+	}
+
+	/**
+	 *
+	 * @param records
+	 * @return 返回一个新的Cache用于存储接下来的数据
+	 */
+	private List<Record> addRecordsToWriteQueue(List<Record> records) {
+		int i = 0;
+		while (true) {
+			if (i > 0) {
+				LOG.info("retry add batch record the {} times", i);
+			}
+			try {
+				concurrentWriter.addBatchRecords(records);
+				break;
+			} catch (InterruptedException e) {
+				i++;
+				LOG.info("Concurrent table writer is interrupted");
+			}
+		}
+		return new ArrayList<Record>(batchSize);
+	}
+	private void checkMemStore() {
+		Connection checkConn = checkConnHolder.reconnect();
+		long now = System.currentTimeMillis();
+		if (now - lastCheckMemstoreTime < 1000 * memstoreCheckIntervalSecond) {
+			return;
+		}
+		boolean isFull = ObWriterUtils.isMemstoreFull(checkConn, memstoreThreshold);
+		this.isMemStoreFull.set(isFull);
+		if (isFull) {
+			LOG.warn("OB memstore is full,sleep 30 seconds, threshold=" + memstoreThreshold);
+		}
+		lastCheckMemstoreTime = now;
+	}
+	
+	public boolean isMemStoreFull() {
+		return isMemStoreFull.get();
+	}
+	
+	public void printEveryTime() {
+		long cost = System.currentTimeMillis() - startTime;
+		if (cost > 10000) { //10s
+			print();
+			startTime = System.currentTimeMillis();
+		}
+	}
+	
+	public void print() {
+		LOG.debug("Statistic total task {}, finished {}, queue Size {}",
+				concurrentWriter.getTotalTaskCount(),
+				concurrentWriter.getFinishTaskCount(),
+				concurrentWriter.getTaskQueueSize());
+		concurrentWriter.printStatistics();
+	}
+	
+	public void waitTaskFinish() {
+		setPutAllTaskInQueue();
+		lock.lock();
+		try {
+			while (!concurrentWriter.checkFinish()) {
+				condition.await(15, TimeUnit.SECONDS);
+				print();
+				checkMemStore();
+			}
+		} catch (InterruptedException e) {
+			LOG.warn("Concurrent table writer wait task finish interrupt");
+		} finally {
+			lock.unlock();
+		}
+		LOG.debug("wait all InsertTask finished ...");
+	}
+	
+	public void singalTaskFinish() {
+		lock.lock();
+		condition.signal();
+		lock.unlock();
+	}
+	
+	@Override
+	public void destroy(Configuration writerSliceConfig) {
+	   if(concurrentWriter!=null) {
+		concurrentWriter.destory();
+		}
+		// 把本级持有的conn关闭掉
+		DBUtil.closeDBResources(null, connHolder.getConn());
+        DBUtil.closeDBResources(null, checkConnHolder.getConn());
+        checkConnHolder.destroy();
+		super.destroy(writerSliceConfig);
+	}
+	
+	public class ConcurrentTableWriter {
+		private BlockingQueue<List<Record>> queue;
+		private List<InsertTask> insertTasks;
+		private Configuration config;
+		private ServerConnectInfo connectInfo;
+		private String rewriteRecordSql;
+		private AtomicLong totalTaskCount;
+		private AtomicLong finishTaskCount;
+		private final int threadCount;
+
+		public ConcurrentTableWriter(Configuration config, ServerConnectInfo connInfo, String rewriteRecordSql) {
+			threadCount = config.getInt(Config.WRITER_THREAD_COUNT, Config.DEFAULT_WRITER_THREAD_COUNT);
+			queue = new LinkedBlockingQueue<List<Record>>(threadCount << 1);
+			insertTasks = new ArrayList<InsertTask>(threadCount);
+			this.config = config;
+			this.connectInfo = connInfo;
+			this.rewriteRecordSql = rewriteRecordSql;
+			this.totalTaskCount = new AtomicLong(0);
+			this.finishTaskCount = new AtomicLong(0);
+		}
+		
+		public long getTotalTaskCount() {
+			return totalTaskCount.get();
+		}
+		
+		public long getFinishTaskCount() {
+			return finishTaskCount.get();
+		}
+		
+		public int getTaskQueueSize() {
+			return queue.size();
+		}
+		
+		public void increFinishCount() {
+			finishTaskCount.incrementAndGet();
+		}
+		
+		//should check after put all the task in the queue
+		public boolean checkFinish() {
+			long finishCount = finishTaskCount.get();
+			long totalCount = totalTaskCount.get();
+			return finishCount == totalCount;
+		}
+		
+		public synchronized void start() {
+			for (int i = 0; i < threadCount; ++i) {
+			    LOG.info("start {} insert task.", (i+1));
+				InsertTask insertTask = new InsertTask(taskId, queue, config, connectInfo, rewriteRecordSql, deleteColPos);
+				insertTask.setWriterTask(ConcurrentTableWriterTask.this);
+				insertTask.setWriter(this);
+				insertTasks.add(insertTask);
+			}
+			WriterThreadPool.executeBatch(insertTasks);
+		}
+		
+		public void printStatistics() {
+			long insertTotalCost = 0;
+			long insertTotalCount = 0;
+			for (InsertTask task: insertTasks) {
+				insertTotalCost += task.getTotalCost();
+				insertTotalCount += task.getInsertCount();
+			}
+			long avgCost = 0;
+			if (insertTotalCount != 0) {
+				avgCost = insertTotalCost / insertTotalCount;
+			}
+			ConcurrentTableWriterTask.LOG.debug("Insert {} times, totalCost {} ms, average {} ms",
+					insertTotalCount, insertTotalCost, avgCost);
+		}
+
+		public void addBatchRecords(final List<Record> records) throws InterruptedException {
+			boolean isSucc = false;
+			while (!isSucc) {
+				isSucc = queue.offer(records, 5, TimeUnit.SECONDS);
+				checkMemStore();
+			}
+			totalTaskCount.incrementAndGet();
+		}
+		
+		public synchronized void destory() {
+			if (insertTasks != null) {
+				for(InsertTask task : insertTasks) {
+					task.setStop();
+				}
+				for(InsertTask task: insertTasks) {
+					task.destroy();
+				}
+			}
+		}
+	}
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/InsertTask.java

@@ -0,0 +1,286 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.task;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.TimeUnit;
+
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.DBUtilErrorCode;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ObClientConnHolder;
+import org.apache.commons.lang3.tuple.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.Config;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ConnHolder;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ServerConnectInfo;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.task.ConcurrentTableWriterTask.ConcurrentTableWriter;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils;
+
+public class InsertTask implements Runnable {
+
+    private static final Logger LOG = LoggerFactory.getLogger(InsertTask.class);
+
+	private ConcurrentTableWriterTask writerTask;
+	private ConcurrentTableWriter writer;
+
+	private String writeRecordSql;
+	private long totalCost = 0;
+	private long insertCount = 0;
+
+	private Queue<List<Record>> queue;
+	private boolean isStop;
+	private ConnHolder connHolder;
+
+	private final long taskId;
+	private ServerConnectInfo connInfo;
+
+	// 失败重试次数
+	private int failTryCount = Config.DEFAULT_FAIL_TRY_COUNT;
+	private boolean printCost = Config.DEFAULT_PRINT_COST;
+	private long costBound = Config.DEFAULT_COST_BOUND;
+	private List<Pair<String, int[]>> deleteMeta;
+
+	public InsertTask(
+			final long taskId,
+			Queue<List<Record>> recordsQueue,
+			Configuration config,
+			ServerConnectInfo connectInfo,
+			String writeRecordSql,
+			List<Pair<String, int[]>> deleteMeta) {
+		this.taskId = taskId;
+		this.queue = recordsQueue;
+		this.connInfo = connectInfo;
+		failTryCount = config.getInt(Config.FAIL_TRY_COUNT, Config.DEFAULT_FAIL_TRY_COUNT);
+		printCost = config.getBool(Config.PRINT_COST, Config.DEFAULT_PRINT_COST);
+		costBound = config.getLong(Config.COST_BOUND, Config.DEFAULT_COST_BOUND);
+		this.connHolder = new ObClientConnHolder(config, connInfo.jdbcUrl,
+                                         connInfo.getFullUserName(), connInfo.password);
+		this.writeRecordSql = writeRecordSql;
+		this.isStop = false;
+		this.deleteMeta = deleteMeta;
+		connHolder.initConnection();
+	}
+	
+	void setWriterTask(ConcurrentTableWriterTask writerTask) {
+		this.writerTask = writerTask;
+	}
+	
+	void setWriter(ConcurrentTableWriter writer) {
+		this.writer = writer;
+	}
+
+	private boolean isStop() { return isStop; }
+	public void setStop() { isStop = true; }
+	public long getTotalCost() { return totalCost; }
+	public long getInsertCount() { return insertCount; }
+	
+	@Override
+	public void run() {
+		Thread.currentThread().setName(String.format("%d-insertTask-%d", taskId, Thread.currentThread().getId()));
+		LOG.debug("Task {} start to execute...", taskId);
+		while (!isStop()) {
+			try {
+				List<Record> records = queue.poll();
+				if (null != records) {
+					doMultiInsert(records, this.printCost, this.costBound);
+
+				} else if (writerTask.isFinished()) {
+					writerTask.singalTaskFinish();
+					LOG.debug("not more task, thread exist ...");
+					break;
+				} else {
+					TimeUnit.MILLISECONDS.sleep(5);
+				}
+			} catch (InterruptedException e) {
+				LOG.debug("TableWriter is interrupt");
+			} catch (Exception e) {
+				LOG.warn("ERROR UNEXPECTED {}", e);
+			}
+		}
+		LOG.debug("Thread exist...");
+	}
+	
+	public void destroy() {
+	    connHolder.destroy();
+	};
+	
+	public void calStatistic(final long cost) {
+		writer.increFinishCount();
+		++insertCount;
+		totalCost += cost;
+		if (this.printCost && cost > this.costBound) {
+			LOG.info("slow multi insert cost {}ms", cost);
+		}
+	}
+
+	private void doDelete(Connection conn, final List<Record> buffer) throws SQLException {
+		if(deleteMeta == null || deleteMeta.size() == 0) {
+			return;
+		}
+		for (int i = 0; i < deleteMeta.size(); i++) {
+			String deleteSql = deleteMeta.get(i).getKey();
+			int[] valueIdx = deleteMeta.get(i).getValue();
+			PreparedStatement ps = null;
+			try {
+				ps = conn.prepareStatement(deleteSql);
+				StringBuilder builder = new StringBuilder();
+				for (Record record : buffer) {
+					int bindIndex = 0;
+					for (int idx : valueIdx) {
+						writerTask.fillStatementIndex(ps, bindIndex++, idx, record.getColumn(idx));
+						builder.append(record.getColumn(idx).asString()).append(",");
+					}
+					ps.addBatch();
+				}
+				LOG.debug("delete values: " + builder.toString());
+				ps.executeBatch();
+			} catch (SQLException ex) {
+				LOG.error("SQL Exception when delete records with {}", deleteSql, ex);
+				throw ex;
+			} finally {
+				DBUtil.closeDBResources(ps, null);
+			}
+		}
+	}
+
+	public void doMultiInsert(final List<Record> buffer, final boolean printCost, final long restrict) {
+		checkMemstore();
+		Connection conn = connHolder.getConn();
+		boolean success = false;
+		long cost = 0;
+		long startTime = 0;
+		try {
+			for (int i = 0; i < failTryCount; ++i) {
+				if (i > 0) {
+					try {
+						int sleep = i >= 9 ? 500 : 1 << i;//不明白为什么要sleep 500s
+						TimeUnit.SECONDS.sleep(sleep);
+					} catch (InterruptedException e) {
+						LOG.info("thread interrupted ..., ignore");
+					}
+					conn = connHolder.getConn();
+					LOG.info("retry {}, start do batch insert, size={}", i, buffer.size());
+					checkMemstore();
+				}
+				startTime = System.currentTimeMillis();
+				PreparedStatement ps = null;
+				try {
+					conn.setAutoCommit(false);
+
+					// do delete if necessary
+					doDelete(conn, buffer);
+
+					ps = conn.prepareStatement(writeRecordSql);
+					for (Record record : buffer) {
+						ps = writerTask.fillStatement(ps, record);
+						ps.addBatch();
+					}
+					ps.executeBatch();
+					conn.commit();
+					success = true;
+					cost = System.currentTimeMillis() - startTime;
+					calStatistic(cost);
+					break;
+				} catch (SQLException e) {
+					LOG.warn("Insert fatal error SqlState ={}, errorCode = {}, {}", e.getSQLState(), e.getErrorCode(), e);
+					if (i == 0 || i > 10 ) {
+						for (Record record : buffer) {
+							LOG.warn("ERROR : record {}", record);
+						}
+					}
+					// 按照错误码分类，分情况处理
+					// 如果是OB系统级异常,则需要重建连接
+					boolean fatalFail = ObWriterUtils.isFatalError(e);
+					if (fatalFail) {
+						ObWriterUtils.sleep(300000);
+						connHolder.reconnect();
+						// 如果是可恢复的异常,则重试
+					} else if (ObWriterUtils.isRecoverableError(e)) {
+						conn.rollback();
+						ObWriterUtils.sleep(60000);
+					} else {// 其它异常直接退出,采用逐条写入方式
+						conn.rollback();
+						ObWriterUtils.sleep(1000);
+						break;
+					}
+				} catch (Exception e) {
+					e.printStackTrace();
+					LOG.warn("Insert error unexpected {}", e);
+				} finally {
+					DBUtil.closeDBResources(ps, null);
+				}
+			}
+		} catch (SQLException e) {
+			LOG.warn("ERROR:retry failSql State ={}, errorCode = {}, {}", e.getSQLState(), e.getErrorCode(), e);
+		}
+
+		if (!success) {
+			try {
+				LOG.info("do one insert");
+				conn = connHolder.reconnect();
+				doOneInsert(conn, buffer);
+				cost = System.currentTimeMillis() - startTime;
+				calStatistic(cost);
+			} finally {
+			}
+		}
+	}
+
+	// process one row, delete before insert
+	private void doOneInsert(Connection connection, List<Record> buffer) {
+		List<PreparedStatement> deletePstmtList = new ArrayList();
+		PreparedStatement preparedStatement = null;
+		try {
+			connection.setAutoCommit(false);
+			if (deleteMeta != null && deleteMeta.size() > 0) {
+				for (int i = 0; i < deleteMeta.size(); i++) {
+					String deleteSql = deleteMeta.get(i).getKey();
+					deletePstmtList.add(connection.prepareStatement(deleteSql));
+				}
+			}
+
+			preparedStatement = connection.prepareStatement(this.writeRecordSql);
+			for (Record record : buffer) {
+				try {
+					for (int i = 0; i < deletePstmtList.size(); i++) {
+						PreparedStatement deleteStmt = deletePstmtList.get(i);
+						int[] valueIdx = deleteMeta.get(i).getValue();
+						int bindIndex = 0;
+						for (int idx : valueIdx) {
+							writerTask.fillStatementIndex(deleteStmt, bindIndex++, idx, record.getColumn(idx));
+						}
+						deleteStmt.execute();
+					}
+					preparedStatement = writerTask.fillStatement(preparedStatement, record);
+					preparedStatement.execute();
+					connection.commit();
+				} catch (SQLException e) {
+					writerTask.collectDirtyRecord(record, e);
+				} finally {
+					// 此处不应该关闭statement，后续的数据还需要用到
+				}
+			}
+		} catch (Exception e) {
+			throw DataXException.asDataXException(
+					DBUtilErrorCode.WRITE_DATA_ERROR, e);
+		} finally {
+			DBUtil.closeDBResources(preparedStatement, null);
+			for (PreparedStatement pstmt : deletePstmtList) {
+				DBUtil.closeDBResources(pstmt, null);
+			}
+		}
+	}
+
+	private void checkMemstore() {
+		while (writerTask.isMemStoreFull()) {
+			ObWriterUtils.sleep(30000);
+		}
+	}
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/SingleTableWriterTask.java

@@ -0,0 +1,152 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.task;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.List;
+
+import com.alibaba.datax.common.element.Record;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.datax.plugin.rdbms.writer.CommonRdbmsWriter;
+import com.alibaba.datax.plugin.rdbms.writer.Key;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.Config;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ConnHolder;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.ext.ObClientConnHolder;
+import com.alibaba.datax.plugin.writer.oceanbasev10writer.util.ObWriterUtils;
+
+public class SingleTableWriterTask extends CommonRdbmsWriter.Task {
+
+	// memstore_total 与 memstore_limit 比例的阈值,一旦超过这个值,则暂停写入
+	private double memstoreThreshold = Config.DEFAULT_MEMSTORE_THRESHOLD;
+
+	// memstore检查的间隔
+	private long memstoreCheckIntervalSecond = Config.DEFAULT_MEMSTORE_CHECK_INTERVAL_SECOND;
+
+	// 最后一次检查
+	private long lastCheckMemstoreTime;
+
+	// 失败重试次数
+	private int failTryCount = Config.DEFAULT_FAIL_TRY_COUNT;
+
+	private ConnHolder connHolder;
+	private String obWriteMode = "update";
+	private boolean isOracleCompatibleMode = false;
+	private String obUpdateColumns = null;
+
+	public SingleTableWriterTask(DataBaseType dataBaseType) {
+		super(dataBaseType);
+	}
+
+	@Override
+	public void init(Configuration config) {
+		super.init(config);
+		this.memstoreThreshold = config.getDouble(Config.MEMSTORE_THRESHOLD, Config.DEFAULT_MEMSTORE_THRESHOLD);
+		this.memstoreCheckIntervalSecond = config.getLong(Config.MEMSTORE_CHECK_INTERVAL_SECOND,
+				Config.DEFAULT_MEMSTORE_CHECK_INTERVAL_SECOND);
+		failTryCount = config.getInt(Config.FAIL_TRY_COUNT, Config.DEFAULT_FAIL_TRY_COUNT);
+		// OceanBase 所有操作都是 insert into on duplicate key update 模式
+		// writeMode应该使用enum来定义
+		this.writeMode = "update";
+		this.connHolder = new ObClientConnHolder(config, jdbcUrl, username, password);
+		//ob1.0里面，
+		this.batchSize = Math.min(128, config.getInt(Key.BATCH_SIZE, 128));
+		LOG.info("In Write OceanBase 1.0, Real Batch Size : " + this.batchSize);
+
+		isOracleCompatibleMode = ObWriterUtils.isOracleMode();
+		LOG.info("isOracleCompatibleMode=" + isOracleCompatibleMode);
+
+		obUpdateColumns = config.getString(Config.OB_UPDATE_COLUMNS, null);
+
+        obWriteMode = config.getString(Config.OB_WRITE_MODE, "update");
+        if (isOracleCompatibleMode) {
+            obWriteMode = "insert";
+        }
+		rewriteSql();
+	}
+
+	private void rewriteSql() {
+		Connection conn = connHolder.initConnection();
+		this.writeRecordSql = ObWriterUtils.buildWriteSql(table, columns, conn, obWriteMode, obUpdateColumns);
+	}
+
+	protected void doBatchInsert(Connection conn, List<Record> buffer) throws SQLException {
+		doBatchInsert(buffer);
+	}
+
+	private void doBatchInsert(List<Record> buffer) {
+		Connection conn = connHolder.getConn();
+		// 检查内存
+		checkMemstore(conn);
+		boolean success = false;
+		try {
+			for (int i = 0; i < failTryCount; i++) {
+				PreparedStatement ps = null;
+				try {
+					conn.setAutoCommit(false);
+					ps = conn.prepareStatement(this.writeRecordSql);
+					for (Record record : buffer) {
+						ps = fillPreparedStatement(ps, record);
+						ps.addBatch();
+					}
+					ps.executeBatch();
+					conn.commit();
+					// 标记执行正常,且退出for循环
+					success = true;
+					break;
+				} catch (SQLException e) {
+					// 如果是OB系统级异常,则需要重建连接
+					boolean fatalFail = ObWriterUtils.isFatalError(e);
+					if (fatalFail) {
+						LOG.warn("Fatal exception in OB. Roll back this write and hibernate for five minutes. SQLState: {}. ErrorCode: {}",
+								e.getSQLState(), e.getErrorCode(), e);
+						ObWriterUtils.sleep(300000);
+						DBUtil.closeDBResources(null, conn);
+						conn = connHolder.reconnect();
+						// 如果是可恢复的异常,则重试
+					} else if (ObWriterUtils.isRecoverableError(e)) {
+						LOG.warn("Recoverable exception in OB. Roll back this write and hibernate for one minute. SQLState: {}. ErrorCode: {}",
+								e.getSQLState(), e.getErrorCode(), e);
+						conn.rollback();
+						ObWriterUtils.sleep(60000);
+						// 其它异常直接退出,采用逐条写入方式
+					} else {
+						LOG.warn("Exception in OB. Roll back this write and hibernate for one second. Write and submit the records one by one. SQLState: {}. ErrorCode: {}",
+								e.getSQLState(), e.getErrorCode(), e);
+						conn.rollback();
+						ObWriterUtils.sleep(1000);
+						break;
+					}
+				} finally {
+					DBUtil.closeDBResources(ps, null);
+				}
+			}
+		} catch (SQLException e) {
+			LOG.warn("Exception in OB. Roll back this write. Write and submit the records one by one. SQLState: {}. ErrorCode: {}",
+					e.getSQLState(), e.getErrorCode(), e);
+		}
+		if (!success) {
+			doOneInsert(conn, buffer);
+		}
+	}
+
+	private void checkMemstore(Connection conn) {
+		long now = System.currentTimeMillis();
+		if (now - lastCheckMemstoreTime < 1000 * memstoreCheckIntervalSecond) {
+			return;
+		}
+		while (ObWriterUtils.isMemstoreFull(conn, memstoreThreshold)) {
+			LOG.warn("OB memstore is full,sleep 60 seconds, threshold=" + memstoreThreshold);
+			ObWriterUtils.sleep(60000);
+		}
+		lastCheckMemstoreTime = now;
+	}
+
+	@Override
+	public void destroy(Configuration writerSliceConfig) {
+		// 把本级持有的conn关闭掉
+		DBUtil.closeDBResources(null, connHolder.getConn());
+		super.destroy(writerSliceConfig);
+	}
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/task/WriterThreadPool.java

@@ -0,0 +1,37 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.task;
+
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class WriterThreadPool {
+    private static final Logger LOG = LoggerFactory.getLogger(InsertTask.class);
+	
+	private static ExecutorService executorService = Executors.newCachedThreadPool();
+	
+	public WriterThreadPool() {
+	}
+	
+	public static ExecutorService getInstance() {
+		return executorService;
+	}
+	
+	public static synchronized void shutdown() {
+		LOG.info("start shutdown executor service...");
+		executorService.shutdown();
+		LOG.info("shutdown executor service success...");
+	}
+	
+	public static synchronized void execute(InsertTask task) {
+		executorService.execute(task);
+	}
+	
+	public static synchronized void executeBatch(List<InsertTask> tasks) {
+		for (InsertTask task : tasks) {
+			executorService.execute(task);
+		}
+	}
+}

oceanbasev10writer/src/main/java/com/alibaba/datax/plugin/writer/oceanbasev10writer/util/DbUtils.java

@@ -0,0 +1,71 @@
+package com.alibaba.datax.plugin.writer.oceanbasev10writer.util;
+
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.util.DBUtil;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.datax.plugin.rdbms.writer.CommonRdbmsWriter;
+import com.alibaba.datax.plugin.rdbms.writer.Constant;
+import com.alibaba.datax.plugin.rdbms.writer.Key;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+public class DbUtils {
+
+    protected static final Logger LOG = LoggerFactory.getLogger(DbUtils.class);
+
+    public static String fetchSingleValueWithRetry(Configuration config, String query) {
+        final String username = config.getString(Key.USERNAME);
+        final String password = config.getString(Key.PASSWORD);
+        String jdbcUrl = config.getString(Key.JDBC_URL);
+
+        if(jdbcUrl == null) {
+            List<Object> conns = config.getList(Constant.CONN_MARK, Object.class);
+            Configuration connConf = Configuration.from(conns.get(0).toString());
+            jdbcUrl = connConf.getString(Key.JDBC_URL);
+        }
+
+        Connection conn = null;
+        PreparedStatement stmt = null;
+        ResultSet result = null;
+        boolean need_retry = false;
+        String value = null;
+        int retry = 0;
+        do {
+            try {
+                if (retry > 0) {
+                    int sleep = retry > 9 ? 500 : 1 << retry;
+                    try {
+                        TimeUnit.SECONDS.sleep(sleep);
+                    } catch (InterruptedException e) {
+                    }
+                    LOG.warn("retry fetch value for {} the {} times", query, retry);
+                }
+                conn = DBUtil.getConnection(DataBaseType.OceanBase, jdbcUrl, username, password);
+                stmt = conn.prepareStatement(query);
+                result = stmt.executeQuery();
+                if (result.next()) {
+                    value = result.getString("Value");
+                } else {
+                    throw new RuntimeException("no values returned for " + query);
+                }
+                LOG.info("value for query [{}] is [{}]", query, value);
+                break;
+            } catch (SQLException e) {
+                need_retry = true;
+                ++retry;
+                LOG.warn("fetch value with {} error {}", query, e);
+            } finally {
+                DBUtil.closeDBResources(result, stmt, null);
+            }
+        } while (need_retry);
+
+        return value;
+    }
+}

oceanbasev10writer/src/main/resources/plugin.json

@@ -0,0 +1,6 @@
+{
+    "name": "oceanbasev10writer",
+    "class": "com.alibaba.datax.plugin.writer.oceanbasev10writer.OceanBaseV10Writer",
+    "description": "write data into oceanbase with sql interface",
+    "developer": "oceanbase"
+}

odpsreader/doc/odpsreader.md

@@ -58,7 +58,8 @@ ODPSReader 支持读取分区表、非分区表，不支持读取虚拟视图。
                         ],
                         "packageAuthorizedProject": "yourCurrentProjectName",
                         "splitMode": "record",
-                        "odpsServer": "http://xxx/api"
+                        "odpsServer": "http://xxx/api",
+                        "tunnelServer": "http://dt.odps.aliyun.com"
                     }
                 },
                 "writer": {

odpsreader/src/main/java/com/alibaba/datax/plugin/reader/odpsreader/util/DESCipher.java

@@ -40,7 +40,7 @@ public class DESCipher {
 	 * 　　
 	 */
 
-	public static final String KEY = "u4Gqu4Z8";
+	public static final String KEY = "DESDES";
 
 	private final static String DES = "DES";