Start SCN deployment

Author: erdemcer

README.md

@@ -53,6 +53,7 @@ In order to execute connector successfully, connector must be started with privi
 |table.whitelist|String|A comma separated list of database schema or table names which will be captured.<br />For all schema capture **<SCHEMA_NAME>.*** <br /> For table capture **<SCHEMA_NAME>.<TABLE_NAME>** must be specified.|
 |parse.dml.data|Boolean|If it is true , captured sql DML statement will be parsed into fields and values.If it is false only sql DML statement is published.
 |reset.offset|Boolean|If it is true , offset value will be set to current SCN of database when connector started.If it is false connector will start from last offset value.
+|start.scn|Long|If it is set , offset value will be set this specified value and logminer will start at this SCN.If connector would like to be started from desired SCN , this property can be used.
 |||
 
 

config/OracleSourceConnector.properties

@@ -26,4 +26,5 @@ db.user.password=kminerpass
 db.fetch.size=1
 table.whitelist=TEST.*,TEST2.TABLE2
 parse.dml.data=true
-reset.offset=true
+reset.offset=true
+start.scn=

src/main/java/com/ecer/kafka/connect/oracle/OracleSourceConnectorConfig.java

@@ -21,6 +21,7 @@ public class OracleSourceConnectorConfig extends AbstractConfig {
   public static final String PARSE_DML_DATA = "parse.dml.data";
   public static final String DB_FETCH_SIZE = "db.fetch.size";
   public static final String RESET_OFFSET = "reset.offset";
+  public static final String START_SCN = "start.scn";
 
   public OracleSourceConnectorConfig(ConfigDef config, Map<String, String> parsedConfig) {
     super(config, parsedConfig);
@@ -42,7 +43,8 @@ public static ConfigDef conf() {
         .define(TABLE_WHITELIST,Type.STRING,Importance.HIGH,"TAbles will be mined")
         .define(PARSE_DML_DATA,Type.BOOLEAN,Importance.HIGH,"Parse DML Data")
         .define(DB_FETCH_SIZE,Type.INT,Importance.HIGH,"Database Record Fetch Size")
-        .define(RESET_OFFSET,Type.BOOLEAN,Importance.HIGH,"Reset Offset");
+        .define(RESET_OFFSET,Type.BOOLEAN,Importance.HIGH,"Reset Offset")
+        .define(START_SCN,Type.STRING,"",Importance.LOW,"Start SCN");
   }
 
   /*public String getMy(){
@@ -59,4 +61,5 @@ public static ConfigDef conf() {
   public Boolean getParseDmlData(){return this.getBoolean(PARSE_DML_DATA);}
   public int getDbFetchSize(){return this.getInt(DB_FETCH_SIZE);}
   public Boolean getResetOffset(){return this.getBoolean(RESET_OFFSET);}
+  public String getStartScn(){return this.getString(START_SCN);}
 }

src/main/java/com/ecer/kafka/connect/oracle/OracleSourceTask.java

@@ -69,10 +69,11 @@ public static void closeDbConn() throws SQLException{
   @Override
   public void start(Map<String, String> map) {
     //TODO: Do things here that are required to start your task. This could be open a connection to a database, etc.
-    config=new OracleSourceConnectorConfig(map);
+    config=new OracleSourceConnectorConfig(map);    
     topic=config.getTopic();
     dbName=config.getDbNameAlias();
-    parseDmlData=config.getParseDmlData();    
+    parseDmlData=config.getParseDmlData();
+    String startSCN = config.getStartScn();
     log.info("Oracle Kafka Connector is starting on {}",config.getDbNameAlias());
     try {
       log.info("Connecting to database");
@@ -89,6 +90,12 @@ public void start(Map<String, String> map) {
         Object lastRecordedOffset = offset.get(POSITION_FIELD);
         streamOffset = (lastRecordedOffset != null) ? (Long) lastRecordedOffset : 0L;
       }
+            
+      if (!startSCN.equals("")){
+        log.info("Resetting offset with specified SCN:{}",startSCN);
+        streamOffset=Long.parseLong(startSCN);
+        streamOffset-=1;        
+      }
 
       if (config.getResetOffset()){
         log.info("Resetting offset with new SCN");