update readme & sample configs.

src/main/java/com/github/zzt93/syncer/producer/input/mysql/meta/ConsumerSchemaMeta.java

@@ -52,7 +52,7 @@ public TableMeta findTable(String database, String table) {
       }
     }
     if (count > 1) {
-      logger.warn("Multiple configured schema match `{}`.`{}`. Check your config", database, table);
+      logger.error("Multiple configured schema match `{}`.`{}`. Check your config", database, table);
     }
     return res;
   }

src/main/resources/application.yml

@@ -1,9 +1,5 @@
 spring.application.name: syncer
 
-spring:
-  main:
-    web-application-type: none
-
 
 spring:
   profiles:

src/main/resources/config.yml

@@ -7,7 +7,6 @@ syncer:
       last-run-metadata-dir: //Users/xiong/data/syncer/input/last_position/
 
   filter:
-    # should be at least twice of core
     worker: 12
 
   output:

src/test/resources/sample-consumer.yml

@@ -1,10 +1,14 @@
+version: 1.1
+
+consumerId: sample1
+
+
 input:
   masters:
     - connection:
         address: 192.168.1.100
         port: 1234
-        user: root
-        passwordFile: password
+      scheduler: mod
       schemas:
         - name: "test_*"
           tables:
@@ -22,8 +26,6 @@ input:
     - connection:
         address: 192.168.1.100
         port: 1235
-        user: root
-        password-file: password
       schemas:
         - name: "*_test"
           tables:
@@ -52,21 +54,21 @@ filter:
   - switcher:
       switch: "table"
       case: # support default branch
-        "affair": ["#suffix = '-' + row['id']","#type = 'INDEX_AFFAIR'", "renameColumn('xx', 'yy')" ]
+        "affair": ["#suffix = '-' + row['id']","#type = 'INDEX_AFFAIR'", "renameRecord('xx', 'yy')" ]
         "file": ["#suffix = '-' + row['id']","#type = 'INDEX_FILE'", "addRow('type', '0')"]
         "folder": ["#suffix = '-' + row['id']","#type = 'INDEX_FILE'", "addRow('type', '1')"]
-  - statement: [ "#tags = row['tags']", "updateColumn('tags', new java.util.ArrayList())", "removeColumns('id', 'xid')"]
+  - statement: [ "#tags = row['tags']", "updateRecord('tags', new java.util.ArrayList())", "removeRecords('id', 'xid')"]
   - foreach:
       var: "tag"
       in: "#tags?.split('\n')"
       statement: ["#map = new java.util.HashMap()", "#map.put('des', #tag)", "row.get('tags').add(#map)"]
   - if:
       condition: "table == 'affair'"
-      if-body:
+      ifBody:
         create:
           copy: [row]
           postCreation: ["table = 'role'", "row['id']"]
-      else-body:
+      elseBody:
         drop: {}
 
 # filter result class: com.github.zzt93.syncer.common.data.SyncData

src/test/resources/sample-consumer2.yml

@@ -1,10 +1,12 @@
+version: 1.1
+
+consumerId: searcher
+
 input:
   masters:
     - connection:
         address: 192.168.1.100
         port: 1234
-        user: root
-        passwordFile: password
       schemas:
         - name: "test_*"
           tables:
@@ -22,8 +24,6 @@ input:
     - connection:
         address: 192.168.1.200
         port: 1234
-        user: root
-        password-file: password
       schemas:
         - name: "test_*"
           tables:
@@ -53,13 +53,24 @@ input:
 #}
 
 filter:
+  - if:
+      condition: "containRecord('tags')"
+      ifBody:
+      - statement: ["updateRecord('tags', T(SyncUtil).fromJson(records['tags'], T(String[])))"]
+  - statement: ["#docType=table"]
+
+  - if:
+      condition: "table == 'task' && isWrite() && records[state] != 0"
+      ifBody:
+      - drop: {}
+
   - switcher:
       switch: "table"
       case: # support default branch
-        "affair": ["#suffix = '-' + row['id']","#type = 'INDEX_AFFAIR'", "renameColumn('xx', 'yy')" ]
+        "affair": ["#suffix = '-' + row['id']","#type = 'INDEX_AFFAIR'", "renameRecord('xx', 'yy')" ]
         "file": ["#suffix = '-' + row['id']","#type = 'INDEX_FILE'", "addRow('type', '0')"]
         "folder": ["#suffix = '-' + row['id']","#type = 'INDEX_FILE'", "addRow('type', '1')"]
-  - statement: [ "#tags = row['tags']", "updateColumn('tags', new java.util.ArrayList())", "removeColumns('id', 'xid')"]
+  - statement: [ "#tags = row['tags']", "updateRecord('tags', new java.util.ArrayList())", "removeRecords('id', 'xid')"]
   - foreach:
       var: "tag"
       in: "#tags?.split('\n')"
@@ -93,18 +104,36 @@ output:
       clusterNodes: ["192.168.1.100:9300"]
       user: elastic-user # optional if not enable security
       passwordFile: es-password # optional if not enable security
-    documentMapping: # mapping from input data to document json
-      index: "table + #suffix" # default schema
-      type: "table" # default table
-      documentId: "id" # default id
-      fieldsMapping: # rest row.* => row.*
-        "row": "row.*.flatten"
+    requestMapping: # mapping from input data to es request
+      enableExtraQuery: true
+      retryOnUpdateConflict: 3
+      index: "table + #suffix" # default: schema
+      type: "#docType" # default: table
+      documentId: "id" # default: id
+      fieldsMapping: # default: records.*.flatten
+        "records": "records.*.flatten"
+    batch:
+      size: 100
+      delay: 1000
+      maxRetry: 5
+    refreshInMillis: 1000
 
-  http:
+  mysql:
     connection:
-      address: 192.168.1.100
-      port: 9700
-    jsonMapping:
-      "data": "row.*"
-      "type": "extra['type']"
+      address: ${HOST_ADDRESS}
+      port: 3306
+      user: xxx
+      password: xxx
+    rowMapping:
+      schema: "'auth'"
+      table: "table"
+      id: "id"
+      rows:
+        "records": "records.*.flatten"
+    batch:
+      size: 100
+      delay: 100
+      maxRetry: 5
+    failureLog:
+      countLimit: 1000
 

src/test/resources/sample-producer.yml

@@ -1,15 +1,22 @@
+version: 1.1
 
-input:
+syncer.producer.input:
   masters:
-    - connection:
-        address: localhost
-        port: 3308
-        user: root
-        passwordFile: password
 
-    - connection:
-        address: localhost
-        port: 27017
+  - connection:
+      address: ${HOST_ADDRESS}
+      port: 3306
+      user: xxx
+      password: yyy
 
+  - connection:
+      address: ${HOST_ADDRESS}
+      port: 27018
+    type: mongo
+
+  - connection:
+      address: ${HOST_ADDRESS}
+      port: 27017
+    type: mongo
 
 

src/test/resources/syncer-config.yml

@@ -0,0 +1,17 @@
+syncer:
+  ack:
+    flushPeriod: 100
+  input:
+    max-retry: 5
+    input-meta:
+      last-run-metadata-dir: /data/syncer/input/last_position/
+
+  filter:
+    worker: 12
+
+  output:
+    worker: 2
+    batch:
+      worker: 2
+    output-meta:
+      failure-log-dir: /data/syncer/output/failure/