import java.net.URL;
import java.sql.Array;
import java.sql.Blob;
-import java.sql.CallableStatement;
import java.sql.Clob;
import java.sql.Connection;
import java.sql.Date;
import java.sql.Time;
import java.sql.Timestamp;
import java.util.Calendar;
-
import org.apache.commons.lang3.StringUtils;
-
import org.onap.music.exceptions.MDBCServiceException;
import org.onap.music.logging.EELFLoggerDelegate;
@Override
public void setTimestamp(int parameterIndex, Timestamp x, Calendar cal) throws SQLException {
- ((CallableStatement)stmt).setTimestamp(parameterIndex, x, cal);
+ ((PreparedStatement)stmt).setTimestamp(parameterIndex, x, cal);
}
@Override
public void setNull(int parameterIndex, int sqlType, String typeName) throws SQLException {
- ((CallableStatement)stmt).setNull(parameterIndex, sqlType, typeName);
+ ((PreparedStatement)stmt).setNull(parameterIndex, sqlType, typeName);
}
@Override
public void setURL(int parameterIndex, URL x) throws SQLException {
- ((CallableStatement)stmt).setURL(parameterIndex, x);
+ ((PreparedStatement)stmt).setURL(parameterIndex, x);
}
@Override
public ParameterMetaData getParameterMetaData() throws SQLException {
- return ((CallableStatement)stmt).getParameterMetaData();
+ return ((PreparedStatement)stmt).getParameterMetaData();
}
@Override
public void setRowId(int parameterIndex, RowId x) throws SQLException {
- ((CallableStatement)stmt).setRowId(parameterIndex, x);
+ ((PreparedStatement)stmt).setRowId(parameterIndex, x);
}
@Override
public void setNString(int parameterIndex, String value) throws SQLException {
- ((CallableStatement)stmt).setNString(parameterIndex, value);
+ ((PreparedStatement)stmt).setNString(parameterIndex, value);
}
@Override
public void setNCharacterStream(int parameterIndex, Reader value, long length) throws SQLException {
- ((CallableStatement)stmt).setNCharacterStream(parameterIndex, value, length);
+ ((PreparedStatement)stmt).setNCharacterStream(parameterIndex, value, length);
}
@Override
public void setNClob(int parameterIndex, NClob value) throws SQLException {
- ((CallableStatement)stmt).setNClob(parameterIndex, value);
+ ((PreparedStatement)stmt).setNClob(parameterIndex, value);
}
@Override
public void setClob(int parameterIndex, Reader reader, long length) throws SQLException {
- ((CallableStatement)stmt).setClob(parameterIndex, reader, length);
+ ((PreparedStatement)stmt).setClob(parameterIndex, reader, length);
}
@Override
public void setBlob(int parameterIndex, InputStream inputStream, long length) throws SQLException {
- ((CallableStatement)stmt).setBlob(parameterIndex, inputStream, length);
+ ((PreparedStatement)stmt).setBlob(parameterIndex, inputStream, length);
}
@Override
public void setNClob(int parameterIndex, Reader reader, long length) throws SQLException {
- ((CallableStatement)stmt).setNClob(parameterIndex, reader, length);
+ ((PreparedStatement)stmt).setNClob(parameterIndex, reader, length);
}
@Override
public void setSQLXML(int parameterIndex, SQLXML xmlObject) throws SQLException {
- ((CallableStatement)stmt).setSQLXML(parameterIndex, xmlObject);
+ ((PreparedStatement)stmt).setSQLXML(parameterIndex, xmlObject);
}
@Override
public void setObject(int parameterIndex, Object x, int targetSqlType, int scaleOrLength) throws SQLException {
- ((CallableStatement)stmt).setObject(parameterIndex, x, targetSqlType, scaleOrLength);
+ ((PreparedStatement)stmt).setObject(parameterIndex, x, targetSqlType, scaleOrLength);
}
@Override
public void setAsciiStream(int parameterIndex, InputStream x, long length) throws SQLException {
- ((CallableStatement)stmt).setAsciiStream(parameterIndex, x, length);
+ ((PreparedStatement)stmt).setAsciiStream(parameterIndex, x, length);
}
@Override
public void setBinaryStream(int parameterIndex, InputStream x, long length) throws SQLException {
- ((CallableStatement)stmt).setBinaryStream(parameterIndex, x, length);
+ ((PreparedStatement)stmt).setBinaryStream(parameterIndex, x, length);
}
@Override
public void setCharacterStream(int parameterIndex, Reader reader, long length) throws SQLException {
- ((CallableStatement)stmt).setCharacterStream(parameterIndex, reader, length);
+ ((PreparedStatement)stmt).setCharacterStream(parameterIndex, reader, length);
}
@Override
public void setAsciiStream(int parameterIndex, InputStream x) throws SQLException {
- ((CallableStatement)stmt).setAsciiStream(parameterIndex, x);
+ ((PreparedStatement)stmt).setAsciiStream(parameterIndex, x);
}
@Override
public void setBinaryStream(int parameterIndex, InputStream x) throws SQLException {
- ((CallableStatement)stmt).setBinaryStream(parameterIndex, x);
+ ((PreparedStatement)stmt).setBinaryStream(parameterIndex, x);
}
@Override
public void setCharacterStream(int parameterIndex, Reader reader) throws SQLException {
- ((CallableStatement)stmt).setCharacterStream(parameterIndex, reader);
+ ((PreparedStatement)stmt).setCharacterStream(parameterIndex, reader);
}
@Override
public void setNCharacterStream(int parameterIndex, Reader value) throws SQLException {
- ((CallableStatement)stmt).setNCharacterStream(parameterIndex, value);
+ ((PreparedStatement)stmt).setNCharacterStream(parameterIndex, value);
}
@Override
public void setClob(int parameterIndex, Reader reader) throws SQLException {
- ((CallableStatement)stmt).setClob(parameterIndex, reader);
+ ((PreparedStatement)stmt).setClob(parameterIndex, reader);
}
@Override
public void setBlob(int parameterIndex, InputStream inputStream) throws SQLException {
- ((CallableStatement)stmt).setBlob(parameterIndex, inputStream);
+ ((PreparedStatement)stmt).setBlob(parameterIndex, inputStream);
}
@Override
public void setNClob(int parameterIndex, Reader reader) throws SQLException {
- ((CallableStatement)stmt).setNClob(parameterIndex, reader);
+ ((PreparedStatement)stmt).setNClob(parameterIndex, reader);
}
}
import java.util.List;
import java.util.Map;
import java.util.Set;
-
+import org.onap.music.exceptions.MDBCServiceException;
import org.onap.music.mdbc.Range;
import org.onap.music.mdbc.TableInfo;
import org.onap.music.mdbc.tables.StagingTable;
* Replay a given TxDigest into the local DB
* @param digest
* @throws SQLException if replay cannot occur correctly
+ * @throws MDBCServiceException
*/
- void replayTransaction(StagingTable digest, List<Range> ranges) throws SQLException;
+ void replayTransaction(StagingTable digest, List<Range> ranges) throws SQLException, MDBCServiceException;
void disableForeignKeyChecks() throws SQLException;
void enableForeignKeyChecks() throws SQLException;
- void applyTxDigest(StagingTable txDigest, List<Range> ranges) throws SQLException;
+ void applyTxDigest(StagingTable txDigest, List<Range> ranges) throws SQLException, MDBCServiceException;
}
}
protected void changeIsLatestToMRI(MusicRangeInformationRow row, boolean isLatest, LockResult lock) throws MDBCServiceException{
+
+ if(lock == null)
+ return;
PreparedQueryObject appendQuery = createChangeIsLatestToMriQuery(musicRangeInformationTableName, row.getPartitionIndex(),
musicTxDigestTableName, isLatest);
ReturnType returnType = MusicCore.criticalPut(music_ns, musicRangeInformationTableName, row.getPartitionIndex().toString(),
- appendQuery, lock.getOwnerId(), null);
+ appendQuery,
+ lock.getOwnerId()
+ , null);
if(returnType.getResult().compareTo(ResultType.SUCCESS) != 0 ){
logger.error(EELFLoggerDelegate.errorLogger, "Error when executing change isLatest operation with return type: "+returnType.getMessage());
throw new MDBCServiceException("Error when executing change isLatest operation with return type: "+returnType.getMessage());
}
private void releaseLocks(List<MusicRangeInformationRow> changed, Map<UUID,LockResult> newLocks) throws MDBCServiceException{
+
for(MusicRangeInformationRow r : changed) {
LockResult lock = newLocks.get(r.getPartitionIndex());
+ if(lock == null)
+ continue;
unlockKeyInMusic(musicRangeInformationTableName, r.getPartitionIndex().toString(),
lock.getOwnerId());
newLocks.remove(r.getPartitionIndex());
/**
* Parse the transaction digest into individual events
* @param transaction - base 64 encoded, serialized digest
+ * @throws MDBCServiceException
*/
- public void replayTransaction(StagingTable transaction, List<Range> ranges) throws SQLException {
+ public void replayTransaction(StagingTable transaction, List<Range> ranges) throws SQLException, MDBCServiceException {
boolean autocommit = jdbcConn.getAutoCommit();
jdbcConn.setAutoCommit(false);
Statement jdbcStmt = jdbcConn.createStatement();
if(Range.overlaps(ranges,op.getTable())) {
try {
replayOperationIntoDB(jdbcStmt, op);
- } catch (SQLException e) {
+ } catch (SQLException | MDBCServiceException e) {
//rollback transaction
logger.error("Unable to replay: " + op.getOperationType() + "->" + op.getVal() + "."
+ "Rolling back the entire digest replay.");
}
@Override
- public void applyTxDigest(StagingTable txDigest,List<Range> ranges) throws SQLException {
+ public void applyTxDigest(StagingTable txDigest,List<Range> ranges) throws SQLException, MDBCServiceException {
replayTransaction(txDigest,ranges);
}
/**
- * Replays operation into database, usually from txDigest
- * @param jdbcStmt
- * @param r
- * @param op
- * @throws SQLException
- */
- private void replayOperationIntoDB(Statement jdbcStmt, Operation op) throws SQLException {
- logger.info("Replaying Operation: " + op.getOperationType() + "->" + op.getVal());
- JSONObject jsonOp = op.getVal();
-
- 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(op.getTable() + " (") ;
- 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(op.getTable() + " SET ");
- sep="";
- for (int i=0; i<cols.size(); i++) {
- sql.append(sep + cols.get(i) + "=\"" + vals.get(i) +"\"");
- sep = ", ";
- }
- sql.append(" WHERE ");
- try {
- sql.append(getPrimaryKeyConditional(op.getKey()));
- } catch (MDBCServiceException e) {
- throw new SQLException("Update operatoin doesn't contain the required primary key",e);
- }
- sql.append(";");
- break;
- case DELETE:
- sql.append(op.getOperationType() + " FROM ");
- sql.append(op.getTable() + " WHERE ");
- sql.append(getPrimaryKeyConditional(op.getVal()));
- 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());
- }
+ * Replays operation into database, usually from txDigest
+ * @param jdbcStmt
+ * @param r
+ * @param op
+ * @throws SQLException
+ * @throws MDBCServiceException
+ */
+ private void replayOperationIntoDB(Statement jdbcStmt, Operation op) throws SQLException, MDBCServiceException {
+ logger.info("Replaying Operation: " + op.getOperationType() + "->" + op.getVal());
+ JSONObject jsonOp = op.getVal();
+
+ 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 and replay the queries
+ StringBuilder sql = constructSQL(op, cols, vals);
+ if(sql == null)
+ return;
+
+ try {
+ logger.info("Replaying operation: " + sql.toString());
+ int updated = jdbcStmt.executeUpdate(sql.toString());
+
+ if(updated == 0) {
+ // This applies only for replaying transactions involving Eventually Consistent tables
+ logger.warn("Error Replaying operation: " + sql.toString() + "; Replacing insert/replace/viceversa and replaying ");
+
+ buildAndExecuteSQLInverse(jdbcStmt, op, cols, vals);
+ }
+ } catch (SQLException sqlE) {
+ // This applies only for replaying transactions involving Eventually Consistent tables
+ logger.warn("Error Replaying operation: " + sql.toString() + "; Replacing insert/replace/viceversa and replaying ");
+
+ buildAndExecuteSQLInverse(jdbcStmt, op, cols, vals);
+
+ }
+ }
+ protected void buildAndExecuteSQLInverse(Statement jdbcStmt, Operation op,
+ ArrayList<String> cols, ArrayList<Object> vals) throws SQLException, MDBCServiceException {
+ StringBuilder sqlInverse = constructSQLInverse( op, cols, vals);
+ if(sqlInverse == null)
+ return;
+ logger.info("Replaying operation: " + sqlInverse.toString());
+ jdbcStmt.executeUpdate(sqlInverse.toString());
+ }
+ protected StringBuilder constructSQLInverse(Operation op, ArrayList<String> cols,
+ ArrayList<Object> vals) throws MDBCServiceException {
+ StringBuilder sqlInverse = null;
+ switch (op.getOperationType()) {
+ case INSERT:
+ sqlInverse = constructUpdate(op.getTable() , OperationType.UPDATE, op.getKey(), cols, vals);
+ break;
+ case UPDATE:
+ sqlInverse = constructInsert(op.getTable() , OperationType.INSERT, cols, vals);
+ break;
+ default:
+ break;
+ }
+ return sqlInverse;
+ }
+ protected StringBuilder constructSQL(Operation op, ArrayList<String> cols,
+ ArrayList<Object> vals) throws MDBCServiceException {
+ StringBuilder sql = null;
+ switch (op.getOperationType()) {
+ case INSERT:
+ sql = constructInsert(op.getTable(), op.getOperationType(), cols, vals);
+ break;
+ case UPDATE:
+ sql = constructUpdate(op.getTable(), op.getOperationType(), op.getKey(), cols, vals);
+ break;
+ case DELETE:
+ sql = constructDelete(op.getTable(), op.getOperationType(), op.getKey());
+ break;
+ case SELECT:
+ //no update happened, do nothing
+ break;
+ default:
+ logger.error(op.getOperationType() + "not implemented for replay");
+ }
+ return sql;
+ }
+ private StringBuilder constructDelete(String r, OperationType op, JSONObject key) {
+ StringBuilder sql = new StringBuilder();
+ sql.append(op + " FROM ");
+ sql.append(r + " WHERE ");
+ sql.append(getPrimaryKeyConditional(key));
+ sql.append(";");
+ return sql;
+ }
+ private StringBuilder constructInsert(String r, OperationType op, ArrayList<String> cols,
+ ArrayList<Object> vals) {
+ StringBuilder sql = new StringBuilder();
+ String sep;
+ sql.append(op + " INTO ");
+ sql.append(r + " (") ;
+ 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(");");
+ return sql;
+ }
+ private StringBuilder constructUpdate(String r, OperationType op, JSONObject key, ArrayList<String> cols,
+ ArrayList<Object> vals) {
+ StringBuilder sql = new StringBuilder();
+ String sep;
+ sql.append(op + " ");
+ sql.append(r + " 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(key));
+ sql.append(";");
+
+ return sql;
+ }
/**
* Create an SQL string for AND'ing all of the primary keys
Code Review / music / mdbc.git / commitdiff