diff options
Diffstat (limited to 'mdbc-server/src/main/java')
5 files changed, 250 insertions, 173 deletions
diff --git a/mdbc-server/src/main/java/org/onap/music/mdbc/MdbcConnection.java b/mdbc-server/src/main/java/org/onap/music/mdbc/MdbcConnection.java index cca14c0..cac4139 100755 --- a/mdbc-server/src/main/java/org/onap/music/mdbc/MdbcConnection.java +++ b/mdbc-server/src/main/java/org/onap/music/mdbc/MdbcConnection.java @@ -67,7 +67,7 @@ public class MdbcConnection implements Connection { private static EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(MdbcConnection.class); private final String id; // This is the transaction id, assigned to this connection. There is no need to change the id, if connection is reused - private final Connection conn; // the JDBC Connection to the actual underlying database + private final Connection jdbcConn; // the JDBC Connection to the actual underlying database private final MusicInterface mi; private final TxCommitProgress progressKeeper; private final DatabasePartition partition; @@ -83,10 +83,10 @@ public class MdbcConnection implements Connection { if (c == null) { throw new MDBCServiceException("Connection is null"); } - this.conn = c; + this.jdbcConn = c; info.putAll(MDBCUtils.getMdbcProperties()); String mixinDb = info.getProperty(Configuration.KEY_DB_MIXIN_NAME, Configuration.DB_MIXIN_DEFAULT); - this.dbi = MixinFactory.createDBInterface(mixinDb, mi, url, conn, info); + this.dbi = MixinFactory.createDBInterface(mixinDb, mi, url, jdbcConn, info); this.mi = mi; try { this.setAutoCommit(c.getAutoCommit()); @@ -112,39 +112,39 @@ public class MdbcConnection implements Connection { @Override public <T> T unwrap(Class<T> iface) throws SQLException { logger.error(EELFLoggerDelegate.errorLogger, "proxyconn unwrap: " + iface.getName()); - return conn.unwrap(iface); + return jdbcConn.unwrap(iface); } @Override public boolean isWrapperFor(Class<?> iface) throws SQLException { logger.error(EELFLoggerDelegate.errorLogger, "proxystatement iswrapperfor: " + iface.getName()); - return conn.isWrapperFor(iface); + return jdbcConn.isWrapperFor(iface); } @Override public Statement createStatement() throws SQLException { - return new MdbcCallableStatement(conn.createStatement(), this); + return new MdbcCallableStatement(jdbcConn.createStatement(), this); } @Override public PreparedStatement prepareStatement(String sql) throws SQLException { //TODO: grab the sql call from here and all the other preparestatement calls - return new MdbcPreparedStatement(conn.prepareStatement(sql), sql, this); + return new MdbcPreparedStatement(jdbcConn.prepareStatement(sql), sql, this); } @Override public CallableStatement prepareCall(String sql) throws SQLException { - return new MdbcCallableStatement(conn.prepareCall(sql), this); + return new MdbcCallableStatement(jdbcConn.prepareCall(sql), this); } @Override public String nativeSQL(String sql) throws SQLException { - return conn.nativeSQL(sql); + return jdbcConn.nativeSQL(sql); } @Override public void setAutoCommit(boolean autoCommit) throws SQLException { - boolean b = conn.getAutoCommit(); + boolean b = jdbcConn.getAutoCommit(); if (b != autoCommit) { if(progressKeeper!=null) progressKeeper.commitRequested(id); logger.debug(EELFLoggerDelegate.applicationLogger,"autocommit changed to "+b); @@ -165,7 +165,7 @@ public class MdbcConnection implements Connection { if(progressKeeper!=null) { progressKeeper.setMusicDone(id); } - conn.setAutoCommit(autoCommit); + jdbcConn.setAutoCommit(autoCommit); if(progressKeeper!=null) { progressKeeper.setSQLDone(id); } @@ -177,7 +177,7 @@ public class MdbcConnection implements Connection { @Override public boolean getAutoCommit() throws SQLException { - return conn.getAutoCommit(); + return jdbcConn.getAutoCommit(); } /** @@ -208,7 +208,7 @@ public class MdbcConnection implements Connection { progressKeeper.setMusicDone(id); } - conn.commit(); + jdbcConn.commit(); if(progressKeeper != null) { progressKeeper.setSQLDone(id); @@ -227,7 +227,7 @@ public class MdbcConnection implements Connection { public void rollback() throws SQLException { logger.debug(EELFLoggerDelegate.applicationLogger, "Rollback");; transactionDigest.clear(); - conn.rollback(); + jdbcConn.rollback(); progressKeeper.reinitializeTxProgress(id); } @@ -240,230 +240,230 @@ public class MdbcConnection implements Connection { if (dbi != null) { dbi.close(); } - if (conn != null && !conn.isClosed()) { + if (jdbcConn != null && !jdbcConn.isClosed()) { logger.debug("Closing jdbc from mdbc with id:"+id); - conn.close(); + jdbcConn.close(); logger.debug("Connection was closed for id:" + id); } } @Override public boolean isClosed() throws SQLException { - return conn.isClosed(); + return jdbcConn.isClosed(); } @Override public DatabaseMetaData getMetaData() throws SQLException { - return conn.getMetaData(); + return jdbcConn.getMetaData(); } @Override public void setReadOnly(boolean readOnly) throws SQLException { - conn.setReadOnly(readOnly); + jdbcConn.setReadOnly(readOnly); } @Override public boolean isReadOnly() throws SQLException { - return conn.isReadOnly(); + return jdbcConn.isReadOnly(); } @Override public void setCatalog(String catalog) throws SQLException { - conn.setCatalog(catalog); + jdbcConn.setCatalog(catalog); } @Override public String getCatalog() throws SQLException { - return conn.getCatalog(); + return jdbcConn.getCatalog(); } @Override public void setTransactionIsolation(int level) throws SQLException { - conn.setTransactionIsolation(level); + jdbcConn.setTransactionIsolation(level); } @Override public int getTransactionIsolation() throws SQLException { - return conn.getTransactionIsolation(); + return jdbcConn.getTransactionIsolation(); } @Override public SQLWarning getWarnings() throws SQLException { - return conn.getWarnings(); + return jdbcConn.getWarnings(); } @Override public void clearWarnings() throws SQLException { - conn.clearWarnings(); + jdbcConn.clearWarnings(); } @Override public Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException { - return new MdbcCallableStatement(conn.createStatement(resultSetType, resultSetConcurrency), this); + return new MdbcCallableStatement(jdbcConn.createStatement(resultSetType, resultSetConcurrency), this); } @Override public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency) throws SQLException { - return new MdbcCallableStatement(conn.prepareStatement(sql, resultSetType, resultSetConcurrency), sql, this); + return new MdbcCallableStatement(jdbcConn.prepareStatement(sql, resultSetType, resultSetConcurrency), sql, this); } @Override public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency) throws SQLException { - return new MdbcCallableStatement(conn.prepareCall(sql, resultSetType, resultSetConcurrency), this); + return new MdbcCallableStatement(jdbcConn.prepareCall(sql, resultSetType, resultSetConcurrency), this); } @Override public Map<String, Class<?>> getTypeMap() throws SQLException { - return conn.getTypeMap(); + return jdbcConn.getTypeMap(); } @Override public void setTypeMap(Map<String, Class<?>> map) throws SQLException { - conn.setTypeMap(map); + jdbcConn.setTypeMap(map); } @Override public void setHoldability(int holdability) throws SQLException { - conn.setHoldability(holdability); + jdbcConn.setHoldability(holdability); } @Override public int getHoldability() throws SQLException { - return conn.getHoldability(); + return jdbcConn.getHoldability(); } @Override public Savepoint setSavepoint() throws SQLException { - return conn.setSavepoint(); + return jdbcConn.setSavepoint(); } @Override public Savepoint setSavepoint(String name) throws SQLException { - return conn.setSavepoint(name); + return jdbcConn.setSavepoint(name); } @Override public void rollback(Savepoint savepoint) throws SQLException { - conn.rollback(savepoint); + jdbcConn.rollback(savepoint); } @Override public void releaseSavepoint(Savepoint savepoint) throws SQLException { - conn.releaseSavepoint(savepoint); + jdbcConn.releaseSavepoint(savepoint); } @Override public Statement createStatement(int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException { - return new MdbcCallableStatement(conn.createStatement(resultSetType, resultSetConcurrency, resultSetHoldability), this); + return new MdbcCallableStatement(jdbcConn.createStatement(resultSetType, resultSetConcurrency, resultSetHoldability), this); } @Override public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException { - return new MdbcCallableStatement(conn.prepareStatement(sql, resultSetType, resultSetConcurrency, resultSetHoldability), sql, this); + return new MdbcCallableStatement(jdbcConn.prepareStatement(sql, resultSetType, resultSetConcurrency, resultSetHoldability), sql, this); } @Override public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException { - return new MdbcCallableStatement(conn.prepareCall(sql, resultSetType, resultSetConcurrency, resultSetHoldability), this); + return new MdbcCallableStatement(jdbcConn.prepareCall(sql, resultSetType, resultSetConcurrency, resultSetHoldability), this); } @Override public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys) throws SQLException { - return new MdbcPreparedStatement(conn.prepareStatement(sql, autoGeneratedKeys), sql, this); + return new MdbcPreparedStatement(jdbcConn.prepareStatement(sql, autoGeneratedKeys), sql, this); } @Override public PreparedStatement prepareStatement(String sql, int[] columnIndexes) throws SQLException { - return new MdbcPreparedStatement(conn.prepareStatement(sql, columnIndexes), sql, this); + return new MdbcPreparedStatement(jdbcConn.prepareStatement(sql, columnIndexes), sql, this); } @Override public PreparedStatement prepareStatement(String sql, String[] columnNames) throws SQLException { - return new MdbcPreparedStatement(conn.prepareStatement(sql, columnNames), sql, this); + return new MdbcPreparedStatement(jdbcConn.prepareStatement(sql, columnNames), sql, this); } @Override public Clob createClob() throws SQLException { - return conn.createClob(); + return jdbcConn.createClob(); } @Override public Blob createBlob() throws SQLException { - return conn.createBlob(); + return jdbcConn.createBlob(); } @Override public NClob createNClob() throws SQLException { - return conn.createNClob(); + return jdbcConn.createNClob(); } @Override public SQLXML createSQLXML() throws SQLException { - return conn.createSQLXML(); + return jdbcConn.createSQLXML(); } @Override public boolean isValid(int timeout) throws SQLException { - return conn.isValid(timeout); + return jdbcConn.isValid(timeout); } @Override public void setClientInfo(String name, String value) throws SQLClientInfoException { - conn.setClientInfo(name, value); + jdbcConn.setClientInfo(name, value); } @Override public void setClientInfo(Properties properties) throws SQLClientInfoException { - conn.setClientInfo(properties); + jdbcConn.setClientInfo(properties); } @Override public String getClientInfo(String name) throws SQLException { - return conn.getClientInfo(name); + return jdbcConn.getClientInfo(name); } @Override public Properties getClientInfo() throws SQLException { - return conn.getClientInfo(); + return jdbcConn.getClientInfo(); } @Override public Array createArrayOf(String typeName, Object[] elements) throws SQLException { - return conn.createArrayOf(typeName, elements); + return jdbcConn.createArrayOf(typeName, elements); } @Override public Struct createStruct(String typeName, Object[] attributes) throws SQLException { - return conn.createStruct(typeName, attributes); + return jdbcConn.createStruct(typeName, attributes); } @Override public void setSchema(String schema) throws SQLException { - conn.setSchema(schema); + jdbcConn.setSchema(schema); } @Override public String getSchema() throws SQLException { - return conn.getSchema(); + return jdbcConn.getSchema(); } @Override public void abort(Executor executor) throws SQLException { - conn.abort(executor); + jdbcConn.abort(executor); } @Override public void setNetworkTimeout(Executor executor, int milliseconds) throws SQLException { - conn.setNetworkTimeout(executor, milliseconds); + jdbcConn.setNetworkTimeout(executor, milliseconds); } @Override public int getNetworkTimeout() throws SQLException { - return conn.getNetworkTimeout(); + return jdbcConn.getNetworkTimeout(); } @@ -517,4 +517,8 @@ public class MdbcConnection implements Connection { } } } + + public DBInterface getDBInterface() { + return this.dbi; + } } diff --git a/mdbc-server/src/main/java/org/onap/music/mdbc/StateManager.java b/mdbc-server/src/main/java/org/onap/music/mdbc/StateManager.java index 1f2ad91..22ddee1 100755 --- a/mdbc-server/src/main/java/org/onap/music/mdbc/StateManager.java +++ b/mdbc-server/src/main/java/org/onap/music/mdbc/StateManager.java @@ -165,6 +165,11 @@ public class StateManager { } } + /** + * Opens a connection into database, setting up all necessary triggers, etc + * @param id UUID of a connection + * @param information + */ public void openConnection(String id, Properties information){ if(!mdbcConnections.containsKey(id)){ Connection sqlConnection; diff --git a/mdbc-server/src/main/java/org/onap/music/mdbc/mixins/DBInterface.java b/mdbc-server/src/main/java/org/onap/music/mdbc/mixins/DBInterface.java index 358287f..73622b1 100755 --- a/mdbc-server/src/main/java/org/onap/music/mdbc/mixins/DBInterface.java +++ b/mdbc-server/src/main/java/org/onap/music/mdbc/mixins/DBInterface.java @@ -20,12 +20,15 @@ package org.onap.music.mdbc.mixins; import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Set; import org.onap.music.mdbc.Range; import org.onap.music.mdbc.TableInfo; +import org.onap.music.mdbc.tables.Operation; import org.onap.music.mdbc.tables.StagingTable; /** @@ -108,4 +111,11 @@ public interface DBInterface { List<String> getReservedTblNames(); String getPrimaryKey(String sql, String tableName); + + /** + * Replay a given TxDigest into the local DB + * @param digest + * @throws SQLException if replay cannot occur correctly + */ + public void replayTransaction(HashMap<Range,StagingTable> digest) throws SQLException; } diff --git a/mdbc-server/src/main/java/org/onap/music/mdbc/mixins/MySQLMixin.java b/mdbc-server/src/main/java/org/onap/music/mdbc/mixins/MySQLMixin.java index 383d522..15384ad 100755 --- a/mdbc-server/src/main/java/org/onap/music/mdbc/mixins/MySQLMixin.java +++ b/mdbc-server/src/main/java/org/onap/music/mdbc/mixins/MySQLMixin.java @@ -27,6 +27,7 @@ import java.sql.Statement; import java.sql.Types; import java.util.ArrayList; import java.util.HashMap; +import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Properties; @@ -39,6 +40,7 @@ import org.json.JSONTokener; import org.onap.music.logging.EELFLoggerDelegate; import org.onap.music.mdbc.Range; import org.onap.music.mdbc.TableInfo; +import org.onap.music.mdbc.tables.Operation; import org.onap.music.mdbc.tables.OperationType; import org.onap.music.mdbc.tables.StagingTable; @@ -73,7 +75,7 @@ public class MySQLMixin implements DBInterface { private final MusicInterface mi; private final int connId; private final String dbName; - private final Connection dbConnection; + private final Connection jdbcConn; private final Map<String, TableInfo> tables; private boolean server_tbl_created = false; @@ -81,14 +83,14 @@ public class MySQLMixin implements DBInterface { this.mi = null; this.connId = 0; this.dbName = null; - this.dbConnection = null; + this.jdbcConn = null; this.tables = null; } public MySQLMixin(MusicInterface mi, String url, Connection conn, Properties info) { this.mi = mi; this.connId = generateConnID(conn); this.dbName = getDBName(conn); - this.dbConnection = conn; + this.jdbcConn = conn; this.tables = new HashMap<String, TableInfo>(); } // This is used to generate a unique connId for this connection to the DB. @@ -154,7 +156,7 @@ public class MySQLMixin implements DBInterface { Set<String> set = new TreeSet<String>(); String sql = "SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_TYPE='BASE TABLE'"; try { - Statement stmt = dbConnection.createStatement(); + Statement stmt = jdbcConn.createStatement(); ResultSet rs = stmt.executeQuery(sql); while (rs.next()) { String s = rs.getString("TABLE_NAME"); @@ -267,7 +269,7 @@ mysql> describe tables; try { if (!server_tbl_created) { try { - Statement stmt = dbConnection.createStatement(); + Statement stmt = jdbcConn.createStatement(); stmt.execute(CREATE_TBL_SQL); stmt.close(); logger.info(EELFLoggerDelegate.applicationLogger,"createSQLTriggers: Server side dirty table created."); @@ -473,7 +475,7 @@ NEW.field refers to the new value logger.debug("Executing SQL read:"+ sql); ResultSet rs = null; try { - Statement stmt = dbConnection.createStatement(); + Statement stmt = jdbcConn.createStatement(); rs = stmt.executeQuery(sql); } catch (SQLException e) { logger.error(EELFLoggerDelegate.errorLogger,"executeSQLRead"+e); @@ -489,7 +491,7 @@ NEW.field refers to the new value protected void executeSQLWrite(String sql) throws SQLException { logger.debug(EELFLoggerDelegate.applicationLogger, "Executing SQL write:"+ sql); - Statement stmt = dbConnection.createStatement(); + Statement stmt = jdbcConn.createStatement(); stmt.execute(sql); stmt.close(); } @@ -610,7 +612,7 @@ NEW.field refers to the new value rs.getStatement().close(); if (rows.size() > 0) { sql2 = "DELETE FROM "+TRANS_TBL+" WHERE IX = ?"; - PreparedStatement ps = dbConnection.prepareStatement(sql2); + PreparedStatement ps = jdbcConn.prepareStatement(sql2); logger.debug("Executing: "+sql2); logger.debug(" For ix = "+rows); for (int ix : rows) { @@ -801,4 +803,140 @@ NEW.field refers to the new value } } } + + /** + * Parse the transaction digest into individual events + * @param transaction - base 64 encoded, serialized digest + * @param dbi + */ + public void replayTransaction(HashMap<Range,StagingTable> transaction) throws SQLException { + boolean autocommit = jdbcConn.getAutoCommit(); + jdbcConn.setAutoCommit(false); + Statement jdbcStmt = jdbcConn.createStatement(); + for (Map.Entry<Range,StagingTable> entry: transaction.entrySet()) { + Range r = entry.getKey(); + StagingTable st = entry.getValue(); + ArrayList<Operation> opList = st.getOperationList(); + + for (Operation op: opList) { + try { + replayOperationIntoDB(jdbcStmt, r, op); + } catch (SQLException e) { + //rollback transaction + logger.error("Unable to replay: " + op.getOperationType() + "->" + op.getNewVal() + "." + + "Rolling back the entire digest replay."); + jdbcConn.rollback(); + throw new SQLException(e); + } + } + } + + clearReplayedOperations(jdbcStmt); + jdbcConn.commit(); + jdbcStmt.close(); + + jdbcConn.setAutoCommit(autocommit); + } + + /** + * Replays operation into database, usually from txDigest + * @param stmt + * @param r + * @param op + * @throws SQLException + */ + private void replayOperationIntoDB(Statement jdbcStmt, Range r, Operation op) throws SQLException { + logger.info("Replaying Operation: " + op.getOperationType() + "->" + op.getNewVal()); + JSONObject jsonOp = op.getNewVal(); + JSONObject key = op.getKey(); + + ArrayList<String> cols = new ArrayList<String>(); + ArrayList<Object> vals = new ArrayList<Object>(); + Iterator<String> colIterator = jsonOp.keys(); + while(colIterator.hasNext()) { + String col = colIterator.next(); + //FIXME: should not explicitly refer to cassandramixin + if (col.equals(MusicMixin.MDBC_PRIMARYKEY_NAME)) { + //reserved name + continue; + } + cols.add(col); + vals.add(jsonOp.get(col)); + } + + //build the queries + StringBuilder sql = new StringBuilder(); + String sep = ""; + switch (op.getOperationType()) { + case INSERT: + sql.append(op.getOperationType() + " INTO "); + sql.append(r.table + " (") ; + sep = ""; + for (String col: cols) { + sql.append(sep + col); + sep = ", "; + } + sql.append(") VALUES ("); + sep = ""; + for (Object val: vals) { + sql.append(sep + "\"" + val + "\""); + sep = ", "; + } + sql.append(");"); + break; + case UPDATE: + sql.append(op.getOperationType() + " "); + sql.append(r.table + " SET "); + sep=""; + for (int i=0; i<cols.size(); i++) { + sql.append(sep + cols.get(i) + "=\"" + vals.get(i) +"\""); + sep = ", "; + } + sql.append(" WHERE "); + sql.append(getPrimaryKeyConditional(op.getKey())); + sql.append(";"); + break; + case DELETE: + sql.append(op.getOperationType() + " FROM "); + sql.append(r.table + " WHERE "); + sql.append(getPrimaryKeyConditional(op.getKey())); + sql.append(";"); + break; + case SELECT: + //no update happened, do nothing + return; + default: + logger.error(op.getOperationType() + "not implemented for replay"); + } + logger.info("Replaying operation: " + sql.toString()); + + jdbcStmt.executeQuery(sql.toString()); + } + + /** + * Create an SQL string for AND'ing all of the primary keys + * @param primaryKeys Json of primary keys and their values + * @return string in the form of PK1=Val1 AND PK2=Val2 AND PK3=Val3 + */ + private String getPrimaryKeyConditional(JSONObject primaryKeys) { + StringBuilder keyCondStmt = new StringBuilder(); + String and = ""; + for (String key: primaryKeys.keySet()) { + Object val = primaryKeys.get(key); + keyCondStmt.append(and + key + "=\"" + val + "\""); + and = " AND "; + } + return keyCondStmt.toString(); + } + + /** + * Cleans out the transaction table, removing the replayed operations + * @param jdbcStmt + * @throws SQLException + */ + private void clearReplayedOperations(Statement jdbcStmt) throws SQLException { + logger.info("Clearing replayed operations"); + String sql = "DELETE FROM " + TRANS_TBL + " WHERE CONNECTION_ID = " + this.connId; + jdbcStmt.executeQuery(sql); + } } diff --git a/mdbc-server/src/main/java/org/onap/music/mdbc/tables/MusicTxDigest.java b/mdbc-server/src/main/java/org/onap/music/mdbc/tables/MusicTxDigest.java index cb5d28e..210cb9e 100644 --- a/mdbc-server/src/main/java/org/onap/music/mdbc/tables/MusicTxDigest.java +++ b/mdbc-server/src/main/java/org/onap/music/mdbc/tables/MusicTxDigest.java @@ -20,12 +20,14 @@ package org.onap.music.mdbc.tables; import java.io.IOException; +import java.sql.SQLException; import java.sql.Timestamp; import java.util.ArrayList; import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Properties; import java.util.UUID; import java.util.concurrent.TimeUnit; @@ -36,11 +38,13 @@ import org.onap.music.exceptions.MusicServiceException; import org.onap.music.logging.EELFLoggerDelegate; import org.onap.music.mdbc.DatabasePartition; import org.onap.music.mdbc.MDBCUtils; +import org.onap.music.mdbc.MdbcConnection; import org.onap.music.mdbc.MdbcServerLogic; import org.onap.music.mdbc.Range; import org.onap.music.mdbc.StateManager; import org.onap.music.mdbc.configurations.NodeConfiguration; import org.onap.music.mdbc.mixins.MusicMixin; +import org.onap.music.mdbc.mixins.DBInterface; import org.onap.music.mdbc.mixins.MusicInterface; import com.datastax.driver.core.Row; @@ -56,106 +60,6 @@ public class MusicTxDigest { this.stateManager = stateManager; } - /** - * Parse the transaction digest into individual events - * @param digest - base 64 encoded, serialized digest - */ - public void replayTxDigest(HashMap<Range,StagingTable> digest) { - for (Map.Entry<Range,StagingTable> entry: digest.entrySet()) { - Range r = entry.getKey(); - StagingTable st = entry.getValue(); - ArrayList<Operation> opList = st.getOperationList(); - - for (Operation op: opList) { - replayOperation(r, op); - } - } - } - - /** - * Replays operation into local database - * @param r - * @param op - */ - private void replayOperation(Range r, Operation op) { - logger.info("Operation: " + op.getOperationType() + "->" + op.getNewVal()); - JSONObject jsonOp = op.getNewVal(); - JSONObject key = op.getKey(); - - ArrayList<String> cols = new ArrayList<String>(); - ArrayList<Object> vals = new ArrayList<Object>(); - Iterator<String> colIterator = jsonOp.keys(); - while(colIterator.hasNext()) { - String col = colIterator.next(); - //FIXME: should not explicitly refer to cassandramixin - if (col.equals(MusicMixin.MDBC_PRIMARYKEY_NAME)) { - //reserved name - continue; - } - cols.add(col); - vals.add(jsonOp.get(col)); - } - - //build the queries - StringBuilder sql = new StringBuilder(); - String sep = ""; - switch (op.getOperationType()) { - case INSERT: - sql.append(op.getOperationType() + " INTO "); - sql.append(r.table + " (") ; - sep = ""; - for (String col: cols) { - sql.append(sep + col); - sep = ", "; - } - sql.append(") VALUES ("); - sep = ""; - for (Object val: vals) { - sql.append(sep + "\"" + val + "\""); - sep = ", "; - } - sql.append(");"); - logger.info(sql.toString()); - break; - case UPDATE: - sql.append(op.getOperationType() + " "); - sql.append(r.table + " SET "); - sep=""; - for (int i=0; i<cols.size(); i++) { - sql.append(sep + cols.get(i) + "=\"" + vals.get(i) +"\""); - sep = ", "; - } - sql.append(" WHERE "); - sql.append(getPrimaryKeyConditional(op.getKey())); - sql.append(";"); - logger.info(sql.toString()); - break; - case DELETE: - sql.append(op.getOperationType() + " FROM "); - sql.append(r.table + " WHERE "); - sql.append(getPrimaryKeyConditional(op.getKey())); - sql.append(";"); - logger.info(sql.toString()); - break; - case SELECT: - //no update happened, do nothing - break; - default: - logger.error(op.getOperationType() + "not implemented for replay"); - } - } - - private String getPrimaryKeyConditional(JSONObject primaryKeys) { - StringBuilder keyCondStmt = new StringBuilder(); - String and = ""; - for (String key: primaryKeys.keySet()) { - Object val = primaryKeys.get(key); - keyCondStmt.append(and + key + "=\"" + val + "\""); - and = " AND "; - } - return keyCondStmt.toString(); - } - /** * Runs the body of the background daemon * @param daemonSleepTimeS time, in seconds, between updates @@ -163,6 +67,9 @@ public class MusicTxDigest { */ public void backgroundDaemon(int daemonSleepTimeS) throws InterruptedException { MusicInterface mi = stateManager.getMusicInterface(); + stateManager.openConnection("daemon", new Properties()); + DBInterface dbi = ((MdbcConnection) stateManager.getConnection("daemon")).getDBInterface(); + while (true) { //update logger.info(String.format("[%s] Background MusicTxDigest daemon updating local db", @@ -175,7 +82,7 @@ public class MusicTxDigest { for (UUID partition: partitions) { if (!partition.equals(myPartition.getMusicRangeInformationIndex())){ try { - replayDigestForPartition(mi, partition); + replayDigestForPartition(mi, partition, dbi); } catch (MDBCServiceException e) { logger.error("Unable to update for partition : " + partition + ". " + e.getMessage()); continue; @@ -186,11 +93,24 @@ public class MusicTxDigest { } } - public void replayDigestForPartition(MusicInterface mi, UUID partitionId) throws MDBCServiceException { - List<MusicTxDigestId> redoLogTxIds = mi.getMusicRangeInformation(partitionId).getRedoLog(); - for (MusicTxDigestId txId: redoLogTxIds) { - HashMap<Range, StagingTable> digest = mi.getTxDigest(txId); - replayTxDigest(digest); + /** + * Replay the digest for a given partition + * @param mi + * @param partitionId + * @param dbi + * @throws MDBCServiceException + */ + public void replayDigestForPartition(MusicInterface mi, UUID partitionId, DBInterface dbi) throws MDBCServiceException { + List<MusicTxDigestId> partitionsRedoLogTxIds = mi.getMusicRangeInformation(partitionId).getRedoLog(); + for (MusicTxDigestId txId: partitionsRedoLogTxIds) { + HashMap<Range, StagingTable> transaction = mi.getTxDigest(txId); + try { + dbi.replayTransaction(transaction); + } catch (SQLException e) { + logger.error("Rolling back the entire digest replay. " + partitionId); + return; + } + logger.info("Successfully replayed transaction " + txId); } //todo, keep track of where I am in pointer } |