diff --git a/databus-core/databus-core-impl/build.gradle b/databus-core/databus-core-impl/build.gradle
index 9f14e788..bda1cb88 100644
--- a/databus-core/databus-core-impl/build.gradle
+++ b/databus-core/databus-core-impl/build.gradle
@@ -16,6 +16,7 @@ dependencies {
   compile externalDependency.json
   compile externalDependency.log4j
   compile externalDependency.netty
+  compile externalDependency.c3p0
 
   testCompile externalDependency.testng
   testCompile externalDependency.easymock
diff --git a/databus-core/databus-core-impl/src/main/java/com/linkedin/databus2/core/seq/MaxSCNReaderWriterConfig.java b/databus-core/databus-core-impl/src/main/java/com/linkedin/databus2/core/seq/MaxSCNReaderWriterConfig.java
index b3093f62..3dc5fc3a 100644
--- a/databus-core/databus-core-impl/src/main/java/com/linkedin/databus2/core/seq/MaxSCNReaderWriterConfig.java
+++ b/databus-core/databus-core-impl/src/main/java/com/linkedin/databus2/core/seq/MaxSCNReaderWriterConfig.java
@@ -27,6 +27,7 @@ public class MaxSCNReaderWriterConfig implements ConfigBuilder<MaxSCNReaderWrite
 
   private String _type;
   private FileMaxSCNHandler.Config _file;
+  private MysqlMaxSCNHandler.Config _mysql;
   private MaxSCNReaderWriter _existing;
 
   public MaxSCNReaderWriterConfig()
@@ -34,6 +35,15 @@ public MaxSCNReaderWriterConfig()
     _type = MaxSCNReaderWriterStaticConfig.Type.FILE.toString();
     _existing = null;
     _file = new FileMaxSCNHandler.Config();
+    _mysql = new MysqlMaxSCNHandler.Config();
+  }
+
+  public MysqlMaxSCNHandler.Config getMysql() {
+    return _mysql;
+  }
+
+  public void setMysql(MysqlMaxSCNHandler.Config _mysql) {
+    this._mysql = _mysql;
   }
 
   public String getType()
@@ -84,7 +94,7 @@ public MaxSCNReaderWriterStaticConfig build() throws InvalidConfigException
       throw new InvalidConfigException("No existing max scn reader/writer specified ");
     }
 
-    return new MaxSCNReaderWriterStaticConfig(handlerType, _file.build(), _existing);
+    return new MaxSCNReaderWriterStaticConfig(handlerType, _file.build(), _mysql.build(),  _existing);
   }
 
 }
diff --git a/databus-core/databus-core-impl/src/main/java/com/linkedin/databus2/core/seq/MaxSCNReaderWriterStaticConfig.java b/databus-core/databus-core-impl/src/main/java/com/linkedin/databus2/core/seq/MaxSCNReaderWriterStaticConfig.java
index 8df5d141..7152688b 100644
--- a/databus-core/databus-core-impl/src/main/java/com/linkedin/databus2/core/seq/MaxSCNReaderWriterStaticConfig.java
+++ b/databus-core/databus-core-impl/src/main/java/com/linkedin/databus2/core/seq/MaxSCNReaderWriterStaticConfig.java
@@ -19,9 +19,8 @@
 */
 
 
-import org.apache.log4j.Logger;
-
 import com.linkedin.databus2.core.seq.FileMaxSCNHandler.StaticConfig;
+import org.apache.log4j.Logger;
 
 /**
  * Static configuration for the SCN reader/writer
@@ -46,21 +45,25 @@ public enum Type
     DISABLED,
     FILE,
     EXISTING,
-    IN_MEMORY
+    IN_MEMORY,
+    MYSQL
   }
 
   private final Type _type;
   private final FileMaxSCNHandler.StaticConfig _file;
+  private final MysqlMaxSCNHandler.StaticConfig _mysql;
   private final MaxSCNReaderWriter _existing;
 
   public MaxSCNReaderWriterStaticConfig(Type type,
                                      StaticConfig file,
+                                        MysqlMaxSCNHandler.StaticConfig  mysql,
                                      MaxSCNReaderWriter existing)
   {
     super();
     _type = type;
     _file = file;
     _existing = existing;
+    _mysql = mysql;
   }
 
   /** Type of of the MaxSCN handler */
@@ -135,6 +138,21 @@ public SequenceNumberHandlerFactory createFactory()
       break;
       case IN_MEMORY: result = new InMemorySequenceNumberHandlerFactory(-1); break;
       case DISABLED: result = null; break;
+      case MYSQL : {
+        MysqlMaxSCNHandler.Config configBuilder = new MysqlMaxSCNHandler.Config();
+        configBuilder.setJdbcUrl(_mysql.getJdbcUrl());
+        configBuilder.setScnTable(_mysql.getScnTable());
+        configBuilder.setDriverClass(_mysql.getDriverClass());
+        configBuilder.setDbPassword(_mysql.getDbPassword());
+        configBuilder.setDbUser(_mysql.getDbUser());
+        configBuilder.setFlushItvl(_mysql.getFlushItvl());
+        configBuilder.setInitVal(_mysql.getInitVal());
+        configBuilder.setUpsertSCNQuery(_mysql.getUpsertSCNQuery());
+        configBuilder.setGetSCNQuery(_mysql.getGetSCNQuery());
+        configBuilder.setScnColumnName(_mysql.getScnColumnName());
+
+        result = new MysqlMaxSCNHandlerFactory(configBuilder);
+      }break;
       default: throw new RuntimeException("unknown scn reader/writer type: " + _type.toString());
     }
 
diff --git a/databus-core/databus-core-impl/src/main/java/com/linkedin/databus2/core/seq/MysqlMaxSCNHandler.java b/databus-core/databus-core-impl/src/main/java/com/linkedin/databus2/core/seq/MysqlMaxSCNHandler.java
new file mode 100644
index 00000000..8606e96a
--- /dev/null
+++ b/databus-core/databus-core-impl/src/main/java/com/linkedin/databus2/core/seq/MysqlMaxSCNHandler.java
@@ -0,0 +1,330 @@
+package com.linkedin.databus2.core.seq;
+
+import com.linkedin.databus.core.util.ConfigBuilder;
+import com.linkedin.databus.core.util.InvalidConfigException;
+import com.linkedin.databus2.core.DatabusException;
+import com.mchange.v2.c3p0.ComboPooledDataSource;
+import org.apache.log4j.Logger;
+
+import java.beans.PropertyVetoException;
+import java.sql.*;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ *
+ */
+public class MysqlMaxSCNHandler  implements MaxSCNReaderWriter {
+
+    private static final Logger LOG = Logger.getLogger(MysqlMaxSCNHandler.class.getName());
+
+    private final StaticConfig staticConfig;
+    private final AtomicLong _flushCounter;
+    private final AtomicLong    _scn;
+    private final ComboPooledDataSource cpds;
+
+    public MysqlMaxSCNHandler(StaticConfig staticConfig, ComboPooledDataSource cpds){
+        this.staticConfig = staticConfig;
+        _flushCounter = new AtomicLong(0);
+        _scn = new AtomicLong(0);
+        this.cpds = cpds;
+    }
+
+
+    public static MysqlMaxSCNHandler create(StaticConfig config) throws DatabusException {
+        ComboPooledDataSource cpds = createConnectionPool(config);
+        MysqlMaxSCNHandler handler = new MysqlMaxSCNHandler(config, cpds);
+        handler.loadInitialValue(cpds);
+        return handler;
+    }
+
+    private void loadInitialValue(ComboPooledDataSource cpds) throws DatabusException {
+        Statement stmt = null;
+        Connection connection = null;
+        try {
+            connection = cpds.getConnection();
+            stmt = connection.createStatement();
+            ResultSet rs = stmt.executeQuery(staticConfig.getGetSCNQuery());
+            if(!rs.isBeforeFirst()){
+                LOG.info("Initial max SCN does not exist in datastore. Defaulting to initial value from configuration: "
+                        + staticConfig.getInitVal());
+                _scn.set(staticConfig.getInitVal());
+                writeScnToDataStore();
+                return;
+            }
+            while(rs.next()) {
+                _scn.set(rs.getLong(staticConfig.getScnColumnName()));
+            }
+            rs.close();
+        } catch (SQLException e) {
+            throw new DatabusException("unable to load initial SCN value",e);
+        } finally {
+            if(stmt!=null){
+                try {
+                    stmt.close();
+                } catch (SQLException e) {
+                }
+            }
+            if(connection!=null){
+                try {
+                    connection.close();
+                } catch (SQLException e) {
+                }
+            }
+        }
+    }
+
+    private static ComboPooledDataSource createConnectionPool(StaticConfig staticConfig) throws DatabusException {
+        ComboPooledDataSource cpds = new ComboPooledDataSource();
+        try {
+            cpds.setDriverClass( staticConfig.getDriverClass() );
+        } catch (PropertyVetoException e) {
+            throw new DatabusException("Unable to create connection pool",e);
+        }
+        cpds.setJdbcUrl(staticConfig.getJdbcUrl());
+        cpds.setUser(staticConfig.getDbUser());
+        cpds.setPassword(staticConfig.getDbPassword());
+        //TODO : Default connection pool setting, make it configurable
+        return cpds;
+    }
+
+
+    @Override
+    public long getMaxScn() throws DatabusException {
+        return _scn.get();
+    }
+
+    @Override
+    public void saveMaxScn(long scn) throws DatabusException {
+        _scn.set(scn);
+
+       long ctr =  _flushCounter.addAndGet(1);
+       if(ctr % staticConfig.getFlushItvl() == 0) {
+           LOG.info("Flushing counter:" + ctr);
+           writeScnToDataStore();
+        }
+    }
+
+    private void writeScnToDataStore() throws DatabusException {
+        LOG.info("save max scn to mysql");
+        PreparedStatement stmt = null;
+        Connection connection = null;
+        try {
+            connection = cpds.getConnection();
+            stmt = connection.prepareStatement(staticConfig.getUpsertSCNQuery());
+            stmt.setLong(1,_scn.get());
+            int i = stmt.executeUpdate();
+            LOG.info("scn inserted "+_scn.get() +", success : "+i);
+        } catch (SQLException e) {
+            LOG.error("Could not persist SCN value "+_scn.get(),e);
+            throw new DatabusException("Could not persist SCN value "+_scn.get(),e);
+        }finally {
+            if(stmt!=null){
+                try {
+                    stmt.close();
+                } catch (SQLException e) {
+                }
+            }
+            if(connection!=null){
+                try {
+                    connection.close();
+                } catch (SQLException e) {
+                }
+            }
+        }
+    }
+
+    public void destroy() throws DatabusException {
+        LOG.info("destory() called, saving scn file before shutting down.");
+        writeScnToDataStore();
+        cpds.close();
+    }
+
+    public static class Config implements ConfigBuilder<StaticConfig>
+    {
+        private String jdbcUrl= "jdbc:mysql://localhost:3306/databus";
+
+        private String scnTable = "databus_scn_store";
+
+        private String driverClass = "com.mysql.jdbc.Driver";
+
+        private String dbUser = "root";
+
+        private String dbPassword = "";
+
+        private Long flushItvl = 1L;
+
+        private Long initVal = 0L;
+
+        private String upsertSCNQuery = "insert into databus_scn_store (max_scn) values (?)";
+
+        private String getSCNQuery = "select max_scn from databus_scn_store order by updated_at desc limit 1";
+
+        private String scnColumnName = "max_scn";
+
+        public String getScnColumnName() {
+            return scnColumnName;
+        }
+
+        public void setScnColumnName(String scnColumnName) {
+            this.scnColumnName = scnColumnName;
+        }
+
+        public Long getInitVal() {
+            return initVal;
+        }
+
+        public void setInitVal(Long initVal) {
+            this.initVal = initVal;
+        }
+
+        public String getUpsertSCNQuery() {
+            return upsertSCNQuery;
+        }
+
+        public void setUpsertSCNQuery(String upsertSCNQuery) {
+            this.upsertSCNQuery = upsertSCNQuery;
+        }
+
+        public String getGetSCNQuery() {
+            return getSCNQuery;
+        }
+
+        public void setGetSCNQuery(String getSCNQuery) {
+            this.getSCNQuery = getSCNQuery;
+        }
+
+        public Long getFlushItvl() {
+            return flushItvl;
+        }
+
+        public void setFlushItvl(Long flushItvl) {
+            this.flushItvl = flushItvl;
+        }
+
+        public String getDbUser() {
+            return dbUser;
+        }
+
+        public void setDbUser(String dbUser) {
+            this.dbUser = dbUser;
+        }
+
+        public String getDbPassword() {
+            return dbPassword;
+        }
+
+        public void setDbPassword(String dbPassword) {
+            this.dbPassword = dbPassword;
+        }
+
+        public String getJdbcUrl() {
+            return jdbcUrl;
+        }
+
+        public void setJdbcUrl(String jdbcUrl) {
+            this.jdbcUrl = jdbcUrl;
+        }
+
+        public String getScnTable() {
+            return scnTable;
+        }
+
+        public void setScnTable(String scnTable) {
+            this.scnTable = scnTable;
+        }
+
+        public String getDriverClass() {
+            return driverClass;
+        }
+
+        public void setDriverClass(String driverClass) {
+            this.driverClass = driverClass;
+        }
+
+        @Override
+        public StaticConfig build() throws InvalidConfigException {
+            //TODO : verify
+            return new StaticConfig(jdbcUrl,scnTable, driverClass, dbUser, dbPassword, flushItvl, initVal ,
+                    upsertSCNQuery, getSCNQuery ,scnColumnName);
+        }
+    }
+    public static class StaticConfig{
+
+        private String driverClass = "com.mysql.jdbc.Driver";
+
+        private String jdbcUrl;
+
+        private String scnTable;
+
+        private String dbUser;
+
+        private String dbPassword;
+
+        private Long  flushItvl;
+
+        private Long initVal;
+
+        private String upsertSCNQuery;
+
+        private String getSCNQuery;
+
+        private String scnColumnName;
+
+        public String getScnColumnName() {
+            return scnColumnName;
+        }
+
+        public Long getInitVal() {
+            return initVal;
+        }
+
+        public String getUpsertSCNQuery() {
+            return upsertSCNQuery;
+        }
+
+        public String getGetSCNQuery() {
+            return getSCNQuery;
+        }
+
+        public Long getFlushItvl() {
+            return flushItvl;
+        }
+
+        public String getDriverClass() {
+            return driverClass;
+        }
+
+        public String getJdbcUrl() {
+            return jdbcUrl;
+        }
+
+        public String getScnTable() {
+            return scnTable;
+        }
+
+        public String getDbUser() {
+            return dbUser;
+        }
+
+        public String getDbPassword() {
+            return dbPassword;
+        }
+
+        public StaticConfig(String host, String table, String driverClass, String dbUser, String dbPassword, long flushItvl, long initVal,
+        String upsertSCNQuery, String getSCNQuery, String scnColumnName){
+            this.jdbcUrl = host;
+            this.scnTable = table;
+            this.driverClass = driverClass;
+            this.dbUser = dbUser;
+            this.dbPassword = dbPassword;
+            this.flushItvl = flushItvl;
+            this.initVal = initVal;
+            this.upsertSCNQuery = upsertSCNQuery;
+            this.getSCNQuery = getSCNQuery;
+            this.scnColumnName = scnColumnName;
+        }
+
+    }
+
+}
+
diff --git a/databus-core/databus-core-impl/src/main/java/com/linkedin/databus2/core/seq/MysqlMaxSCNHandlerFactory.java b/databus-core/databus-core-impl/src/main/java/com/linkedin/databus2/core/seq/MysqlMaxSCNHandlerFactory.java
new file mode 100644
index 00000000..43a99d79
--- /dev/null
+++ b/databus-core/databus-core-impl/src/main/java/com/linkedin/databus2/core/seq/MysqlMaxSCNHandlerFactory.java
@@ -0,0 +1,26 @@
+package com.linkedin.databus2.core.seq;
+
+import com.linkedin.databus2.core.DatabusException;
+
+/**
+ *
+ */
+public class MysqlMaxSCNHandlerFactory implements SequenceNumberHandlerFactory {
+    private final MysqlMaxSCNHandler.Config _configBuilder;
+
+    public MysqlMaxSCNHandlerFactory(MysqlMaxSCNHandler.Config configBuilder)
+    {
+        _configBuilder = configBuilder;
+    }
+
+    @Override
+    public MaxSCNReaderWriter createHandler(String id) throws DatabusException {
+        MysqlMaxSCNHandler maxSCNHandler;
+        MysqlMaxSCNHandler.StaticConfig config;
+        synchronized (_configBuilder) {
+            config = _configBuilder.build();
+            maxSCNHandler = MysqlMaxSCNHandler.create(config);
+        }
+        return maxSCNHandler;
+    }
+}
diff --git a/subprojects.gradle b/subprojects.gradle
index 6a72a627..2eafa51a 100644
--- a/subprojects.gradle
+++ b/subprojects.gradle
@@ -47,7 +47,8 @@ ext.externalDependency = [
     'zookeeper': 'org.apache.zookeeper:zookeeper:3.3.3',
     'ojdbc6': 'com.oracle:ojdbc6:11.2.0.2.0',
     'helixCore': 'org.apache.helix:helix-core:0.6.2.0',
-    'or': 'com.linkedin.dds-mysql:open-replicator-impl:1.0.63'
+    'or': 'com.linkedin.dds-mysql:open-replicator-impl:1.0.63',
+    'c3p0': 'com.mchange:c3p0:0.9.5'
 ];
 
 if (isDefaultEnvironment) {