import org.onap.music.mdbc.query.SQLOperation;
import org.onap.music.mdbc.query.SQLOperationType;
import org.onap.music.mdbc.tables.MusicRangeInformationRow;
+import org.onap.music.mdbc.tables.MusicTxDigestId;
import org.onap.music.mdbc.tables.StagingTable;
import org.onap.music.mdbc.tables.TxCommitProgress;
try {
logger.debug(EELFLoggerDelegate.applicationLogger, " commit ");
// transaction was committed -- add all the updates into the REDO-Log in MUSIC
- mi.commitLog(partition, statemanager.getEventualRanges(), transactionDigest, id, progressKeeper);
+ MusicTxDigestId digestCreated = mi.commitLog(partition, statemanager.getEventualRanges(), transactionDigest, id, progressKeeper);
+ statemanager.getOwnAndCheck().updateAlreadyApplied(mi, dbi, partition.getSnapshot(), partition.getMRIIndex(), digestCreated);
} catch (MDBCServiceException e) {
//If the commit fail, then a new commitId should be used
logger.error(EELFLoggerDelegate.errorLogger, "Commit to music failed", AppMessages.UNKNOWNERROR, ErrorTypes.UNKNOWN, ErrorSeverity.FATAL);
MDBCUtils.writeLocksOnly = (writeLocksOnly==null) ? Configuration.WRITE_LOCK_ONLY_DEFAULT : Boolean.parseBoolean(writeLocksOnly);
initMusic();
- initSqlDatabase();
- initTxDaemonThread();
+ Map<Range, Pair<MriReference, MusicTxDigestId>> alreadyApplied = initSqlDatabase();
+
String t = info.getProperty(Configuration.KEY_OWNERSHIP_TIMEOUT);
long timeout = (t == null) ? Configuration.DEFAULT_OWNERSHIP_TIMEOUT : Integer.parseInt(t);
- alreadyApplied = new ConcurrentHashMap<>();
ownAndCheck = new OwnershipAndCheckpoint(alreadyApplied, timeout);
+
+ initTxDaemonThread();
}
protected String cleanSqlUrl(String url){
this.mdbcConnections = new HashMap<>();
}
- protected void initSqlDatabase() throws MDBCServiceException {
+ /**
+ * Do everything necessary to initialize the sql database
+ * @return the current checkpoint location of this database, if restarting
+ * @throws MDBCServiceException
+ */
+ protected Map<Range, Pair<MriReference, MusicTxDigestId>> initSqlDatabase() throws MDBCServiceException {
if(!this.sqlDBUrl.toLowerCase().startsWith("jdbc:postgresql")) {
try {
Connection sqlConnection = DriverManager.getConnection(this.sqlDBUrl, this.info);
}
}
- // Verify the tables in MUSIC match the tables in the database
- // and create triggers on any tables that need them
+ Map<Range, Pair<MriReference, MusicTxDigestId>> alreadyAppliedToDb = null;
try {
MdbcConnection mdbcConn = (MdbcConnection) openConnection("init");
mdbcConn.initDatabase();
+ alreadyAppliedToDb = mdbcConn.getDBInterface().getCheckpointLocations();
closeConnection("init");
} catch (QueryException e) {
- logger.error("Error syncrhonizing tables");
+ logger.error("Error initializing sql database tables");
logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(), AppMessages.QUERYERROR, ErrorTypes.QUERYERROR, ErrorSeverity.CRITICAL);
}
+
+ if (alreadyAppliedToDb==null) {
+ alreadyAppliedToDb = new ConcurrentHashMap<>();
+ }
+ return alreadyAppliedToDb;
}
/**
import org.onap.music.exceptions.MDBCServiceException;
import org.onap.music.mdbc.Range;
import org.onap.music.mdbc.TableInfo;
+import org.onap.music.mdbc.tables.MriReference;
+import org.onap.music.mdbc.tables.MusicTxDigestId;
import org.onap.music.mdbc.tables.StagingTable;
/**
* @param r
* @param playbackPointer
*/
- public void updateCheckpointLocations(Range r, Pair<UUID, Integer> playbackPointer);
+ public void updateCheckpointLocations(Range r, Pair<MriReference, MusicTxDigestId> playbackPointer);
+ /**
+ * Get current locations of this database's already applied locations
+ * @return
+ */
+ public Map<Range, Pair<MriReference, MusicTxDigestId>> getCheckpointLocations();
/**
* Initialize the SQL database by creating any tables necessary
* Commits the corresponding REDO-log into MUSIC
* Transaction is committed -- add all the updates into the REDO-Log in MUSIC
*
+ * This officially commits the transaction globally
+ *
+ *
+ *
* @param partition information related to ownership of partitions, used to verify ownership when commiting the Tx
* @param eventualRanges
* @param transactionDigest digest of the transaction that is being committed into the Redo log in music.
* @param txId id associated with the log being send
* @param progressKeeper data structure that is used to handle to detect failures, and know what to do
+ * @return digest that was created for this transaction commit
* @throws MDBCServiceException
*/
- void commitLog(DatabasePartition partition, Set<Range> eventualRanges, StagingTable transactionDigest, String txId,TxCommitProgress progressKeeper) throws MDBCServiceException;
+ public MusicTxDigestId commitLog(DatabasePartition partition, Set<Range> eventualRanges, StagingTable transactionDigest, String txId,TxCommitProgress progressKeeper) throws MDBCServiceException;
/**
* @param playbackPointer
* @throws MDBCServiceException
*/
- public void updateCheckpointLocations(Range r, Pair<UUID, Integer> playbackPointer) throws MDBCServiceException;
+ public void updateCheckpointLocations(Range r, Pair<MriReference, MusicTxDigestId> playbackPointer);
}
addTxDigest(digestId, serializedTransactionDigest);
}
- /**
- * Writes the transaction information to metric's txDigest and musicRangeInformation table
- * This officially commits the transaction globally
- */
@Override
- public void commitLog(DatabasePartition partition,Set<Range> eventualRanges, StagingTable transactionDigest,
+ public MusicTxDigestId commitLog(DatabasePartition partition,Set<Range> eventualRanges, StagingTable transactionDigest,
String txId ,TxCommitProgress progressKeeper) throws MDBCServiceException {
// first deal with commit for eventually consistent tables
if(partition==null){
logger.warn("Trying tcommit log with null partition");
- return;
+ return null;
}
Set<Range> snapshot = partition.getSnapshot();
if(snapshot==null || snapshot.isEmpty()){
logger.warn("Trying to commit log with empty ranges");
- return;
+ return null;
}
//Add creation type of transaction digest
if(transactionDigest == null || transactionDigest.isEmpty()) {
- return;
+ return null;
}
UUID mriIndex = partition.getMRIIndex();
progressKeeper.setRecordId(txId, digestId);
}
- Set<Range> ranges = partition.getSnapshot();
-
- Map<Range, Pair<MriReference, MusicTxDigestId>> alreadyApplied = stateManager.getOwnAndCheck().getAlreadyApplied();
- for(Range r : ranges) {
- alreadyApplied.put(r, Pair.of(new MriReference(mriIndex), digestId));
- }
+ return digestId;
}
private void filterAndAddEventualTxDigest(Set<Range> eventualRanges,
}
public static void createMusicMdbcCheckpointTable(String namespace, String checkpointTable) throws MDBCServiceException {
- String priKey = "txid";
- StringBuilder fields = new StringBuilder();
- fields.append("txid uuid, ");
- fields.append("compressed boolean, ");
- fields.append("transactiondigest blob ");//notice lack of ','
String cql =
- String.format("CREATE TABLE IF NOT EXISTS %s.%s (mdbcnode UUID, mridigest UUID, digestindex int, PRIMARY KEY (mdbcnode));",
+ String.format("CREATE TABLE IF NOT EXISTS %s.%s (mdbcnode text, range text, mridigest UUID,"
+ + "digestid UUID, PRIMARY KEY (mdbcnode, range));",
namespace, checkpointTable);
try {
executeMusicWriteQuery(namespace,checkpointTable,cql);
}
@Override
- public void updateCheckpointLocations(Range r, Pair<UUID, Integer> playbackPointer) throws MDBCServiceException {
- String cql = String.format("INSERT INTO %s.%s (mdbcnode, mridigest, digestindex) VALUES ("
- + this.myId + ", " + playbackPointer.getLeft() + ", " + playbackPointer.getRight() + ");",
- music_ns, this.musicMdbcCheckpointsTableName);
+ public void updateCheckpointLocations(Range r, Pair<MriReference, MusicTxDigestId> playbackPointer) {
+ String cql = String.format("INSERT INTO %s.%s (mdbcnode, range, mridigest, digestid) VALUES ('%s', '%s', %s, %s);",
+ music_ns, this.musicMdbcCheckpointsTableName, this.stateManager.getMdbcServerName(), r.getTable(),
+ playbackPointer.getLeft().getIndex(), playbackPointer.getRight().transactionId);
PreparedQueryObject pQueryObject = new PreparedQueryObject();
pQueryObject.appendQueryString(cql);
try {
} catch (MusicServiceException e) {
logger.warn(EELFLoggerDelegate.applicationLogger, "Unable to update the checkpoint location", e);
} catch (MusicQueryException e) {
- throw new MDBCServiceException(e);
+ logger.warn(EELFLoggerDelegate.applicationLogger, "Unable to update the checkpoint location with query", e);
}
}
import java.util.Set;
import java.util.TreeSet;
import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.lang3.tuple.Pair;
import org.json.JSONObject;
import org.onap.music.exceptions.MDBCServiceException;
import org.onap.music.mdbc.Range;
import org.onap.music.mdbc.TableInfo;
import org.onap.music.mdbc.query.SQLOperation;
+import org.onap.music.mdbc.tables.MriReference;
+import org.onap.music.mdbc.tables.MusicTxDigestId;
import org.onap.music.mdbc.tables.Operation;
import org.onap.music.mdbc.tables.StagingTable;
import net.sf.jsqlparser.JSQLParserException;
+ "CONNECTION_ID INT, PRIMARY KEY (IX));";
private static final String CKPT_TBL = "MDBC_CHECKPOINT";
private static final String CREATE_CKPT_SQL =
- "CREATE TABLE IF NOT EXISTS " + CKPT_TBL + " (RANGENAME VARCHAR(64) PRIMARY KEY, MRIROW VARCHAR(36), DIGESTINDEX INT);";
+ "CREATE TABLE IF NOT EXISTS " + CKPT_TBL + " (RANGENAME VARCHAR(64) PRIMARY KEY, MRIROW VARCHAR(36), DIGESTID VARCHAR(36));";
private final MusicInterface mi;
private final int connId;
logger.debug(EELFLoggerDelegate.applicationLogger, "getSQLTableSet returning: " + set);
Set<Range> rangeSet = new HashSet<>();
for (String table : set) {
- if (getReservedTblNames().contains(table)) {
+ if (!getReservedTblNames().contains(table)) {
// Don't create triggers for the table the triggers write into!!!
rangeSet.add(new Range(table));
}
}
@Override
- public void updateCheckpointLocations(Range r, Pair<UUID, Integer> playbackPointer) {
- String query = "UPDATE " + CKPT_TBL + " SET MRIROW=?, DIGESTINDEX=? where RANGENAME=?;";
+ public void updateCheckpointLocations(Range r, Pair<MriReference, MusicTxDigestId> playbackPointer) {
+ String query = "UPDATE " + CKPT_TBL + " SET MRIROW=?, DIGESTID=? where RANGENAME=?;";
try {
PreparedStatement stmt = jdbcConn.prepareStatement(query);
- stmt.setString(1, playbackPointer.getLeft().toString());
- stmt.setInt(2, playbackPointer.getRight());
+ stmt.setString(1, playbackPointer.getLeft().getIndex().toString());
+ stmt.setString(2, playbackPointer.getRight().transactionId.toString());
stmt.setString(3, r.getTable());
stmt.execute();
stmt.close();
}
}
+ @Override
+ public Map<Range, Pair<MriReference, MusicTxDigestId>> getCheckpointLocations() {
+ Map<Range, Pair<MriReference, MusicTxDigestId>> alreadyApplied = new ConcurrentHashMap<>();
+ try {
+ Statement stmt = jdbcConn.createStatement();
+ ResultSet rs = stmt.executeQuery("SELECT * FROM " + CKPT_TBL + ";");
+ while (rs.next()) {
+ Range r = new Range(rs.getString("RANGENAME"));
+ String mrirow = rs.getString("MRIROW");
+ String txId = rs.getString("DIGESTID");
+ if (mrirow!=null) {
+ logger.info(EELFLoggerDelegate.applicationLogger,
+ "Previously checkpointed: " + r.getTable() + " at (" + mrirow + ", " + txId + ")");
+ alreadyApplied.put(r, Pair.of(new MriReference(mrirow), new MusicTxDigestId(mrirow, txId, -1)));
+ }
+ }
+ stmt.close();
+ } catch (SQLException e) {
+ logger.error(EELFLoggerDelegate.errorLogger, "Unable to get replay checkpoint location", e);
+ }
+
+ return alreadyApplied;
+ }
+
@Override
public void initTables() {
try {
import org.onap.music.mdbc.Range;
import org.onap.music.mdbc.TableInfo;
import org.onap.music.mdbc.mixins.MySQLMixin.StagingTableUpdateRunnable;
+import org.onap.music.mdbc.tables.MriReference;
+import org.onap.music.mdbc.tables.MusicTxDigestId;
import org.onap.music.mdbc.tables.Operation;
import org.onap.music.mdbc.query.SQLOperation;
import org.onap.music.mdbc.tables.StagingTable;
}
@Override
- public void updateCheckpointLocations(Range r, Pair<UUID, Integer> playbackPointer) {
+ public void updateCheckpointLocations(Range r, Pair<MriReference, MusicTxDigestId> playbackPointer) {
+ throw new org.apache.commons.lang.NotImplementedException();
+ }
+
+ @Override
+ public Map<Range, Pair<MriReference, MusicTxDigestId>> getCheckpointLocations() {
throw new org.apache.commons.lang.NotImplementedException();
}
return;
}
applyTxDigest(dbi, txDigest);
- for (Range r : ranges) {
- MusicRangeInformationRow row = node.getRow();
- alreadyApplied.put(r, Pair.of(new MriReference(row.getPartitionIndex()), digestId));
-
- updateCheckpointLocations(mi, dbi, r, row.getPartitionIndex(), digestId);
- }
+ MusicRangeInformationRow row = node.getRow();
+ updateAlreadyApplied(mi, dbi, ranges, row.getPartitionIndex(), digestId);
}
/**
}
MriReference appliedMriRef = applied.getLeft();
MusicTxDigestId appliedDigest = applied.getRight();
- int appliedIndex = node.getRow().getRedoLog().indexOf(appliedDigest);
+ appliedDigest.index = node.getRow().getRedoLog().indexOf(appliedDigest);
if (appliedMriRef==null || appliedMriRef.getTimestamp() < node.getTimestamp()
|| (appliedMriRef.getTimestamp() == node.getTimestamp()
- && appliedIndex < index)) {
+ && appliedDigest.index < index)) {
return false;
}
}
* @param mi
* @param di
* @param r
- * @param partitionIndex
+ * @param mriRef
* @param index
* @throws MDBCServiceException
*/
- private void updateCheckpointLocations(MusicInterface mi, DBInterface dbi, Range r, UUID partitionIndex, MusicTxDigestId txdigest) throws MDBCServiceException {
- dbi.updateCheckpointLocations(r, Pair.of(partitionIndex, txdigest.index));
- mi.updateCheckpointLocations(r, Pair.of(partitionIndex, txdigest.index));
+ private void updateCheckpointLocations(MusicInterface mi, DBInterface dbi, Range r, MriReference mriRef, MusicTxDigestId txdigest) {
+ dbi.updateCheckpointLocations(r, Pair.of(mriRef, txdigest));
+ mi.updateCheckpointLocations(r, Pair.of(mriRef, txdigest));
}
/**
public Map<Range, Pair<MriReference, MusicTxDigestId>> getAlreadyApplied() {
return this.alreadyApplied;
- }
+ }
+ public void updateAlreadyApplied(MusicInterface mi, DBInterface dbi, Set<Range> ranges, UUID mriIndex, MusicTxDigestId digestId) {
+ for (Range r: ranges) {
+ updateAlreadyApplied(mi, dbi, r, mriIndex, digestId);
+ }
+ }
+
+ public void updateAlreadyApplied(MusicInterface mi, DBInterface dbi, Range r, UUID mriIndex, MusicTxDigestId digestId) {
+ MriReference mriRef = new MriReference(mriIndex);
+ alreadyApplied.put(r, Pair.of(mriRef, digestId));
+ updateCheckpointLocations(mi, dbi, r, mriRef, digestId);
+ }
+
}
this.index= index;
}
- public long getTimestamp() { return index.timestamp();}
+ public MriReference(String mrirow) {
+ index = UUID.fromString(mrirow);
+ }
+ public long getTimestamp() {
+ return index.timestamp();
+ }
+
+ public UUID getIndex() {
+ return this.index;
+ }
+
public String toString() {
return index.toString();
}
public final class MusicTxDigestId {
public final UUID mriId;
public final UUID transactionId;
- public final int index;
+ public int index;
public MusicTxDigestId(UUID mriRowId, UUID digestId, int index) {
this.mriId=mriRowId;
this.transactionId= digestId;
this.index=index;
}
+
+ public MusicTxDigestId(String mriRowId, String digestId, int index) {
+ this.mriId = UUID.fromString(mriRowId);
+ this.transactionId = UUID.fromString(digestId);
+ this.index = index;
+ }
public MusicTxDigestId(UUID digestId, int index) {
this.mriId = null;
}
}
- public static void cleanDatabase(DBType type){
+ public static void stopDatabase(DBType type){
switch(type) {
case MySQL:
stopMySql();
package org.onap.music.mdbc.mixins;
+import java.util.HashSet;
+import java.util.Map;
import java.util.Properties;
+import java.util.Set;
+import org.apache.commons.lang3.tuple.Pair;
import org.junit.*;
-
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
+import java.sql.Statement;
+import org.onap.music.mdbc.MDBCUtils;
import org.onap.music.mdbc.MdbcTestUtils;
import org.onap.music.mdbc.MdbcTestUtils.DBType;
+import org.onap.music.mdbc.Range;
import org.onap.music.mdbc.mixins.MySQLMixin;
-
+import org.onap.music.mdbc.tables.MriReference;
+import org.onap.music.mdbc.tables.MusicTxDigestId;
import ch.vorburger.mariadb4j.DB;
public class MySQLMixinTest {
@AfterClass
public static void close() throws Exception {
- MdbcTestUtils.cleanDatabase(DBType.MySQL);
+ MdbcTestUtils.stopDatabase(DBType.MySQL);
}
@Before
this.mysqlMixin = new MySQLMixin(null, null, conn, info);
}
- @Test
- public void testGetDataBaseName() throws SQLException {
- Assert.assertEquals(MdbcTestUtils.getMariaDBDBName().toUpperCase(), mysqlMixin.getDatabaseName());
- }
+
+ @After
+ public void afterEachTest() throws SQLException {
+ clearTables();
+ }
+
+ @Test
+ public void testGetDataBaseName() throws SQLException {
+ assertEquals(MdbcTestUtils.getMariaDBDBName(), mysqlMixin.getDatabaseName());
+ }
+
+ @Test
+ public void testGetTableSet() throws SQLException {
+ Set<Range> rangesAdded = new HashSet<>();
+ rangesAdded.add(new Range("TEST.RANGER"));
+ rangesAdded.add(new Range("TEST.RANGES"));
+ Statement st = conn.createStatement();
+ for (Range r: rangesAdded) {
+ st.execute("CREATE TABLE " + r + " (name VARCHAR(20));");
+ }
+ st.close();
+ Set<Range> ranges = mysqlMixin.getSQLRangeSet();
+
+ assertTrue(ranges.containsAll(rangesAdded));
+ assertTrue(rangesAdded.containsAll(ranges));
+ }
+
+ @Test
+ public void testCkpt() throws SQLException {
+ createTables();
+
+ Range r1 = new Range(MdbcTestUtils.mariaDBDatabaseName + ".RANGER");
+ MriReference mri1 = new MriReference(MDBCUtils.generateUniqueKey());
+ MusicTxDigestId i1 = new MusicTxDigestId(mri1.index, MDBCUtils.generateUniqueKey(), 1);
+ Pair<MriReference, MusicTxDigestId> digestId = Pair.of(mri1, i1);
+ mysqlMixin.updateCheckpointLocations(r1, digestId);
+
+ Range r2 = new Range(MdbcTestUtils.mariaDBDatabaseName + ".RANGES");
+ MriReference mri2 = new MriReference(MDBCUtils.generateUniqueKey());
+ MusicTxDigestId i2 = new MusicTxDigestId(mri2.index, MDBCUtils.generateUniqueKey(), 2);
+ Pair<MriReference, MusicTxDigestId> p2 = Pair.of(mri2, i2);
+ mysqlMixin.updateCheckpointLocations(r2, p2);
+
+ Map<Range, Pair<MriReference, MusicTxDigestId>> ckptmap = mysqlMixin.getCheckpointLocations();
+ assertTrue(ckptmap.containsKey(r1));
+ assertEquals(mri1.getIndex(), ckptmap.get(r1).getLeft().getIndex());
+ assertEquals(i1.transactionId, ckptmap.get(r1).getRight().transactionId);
+
+ assertTrue(ckptmap.containsKey(r2));
+ assertEquals(mri2.getIndex(), ckptmap.get(r2).getLeft().getIndex());
+ assertEquals(i2.transactionId, ckptmap.get(r2).getRight().transactionId);
+ }
+
+ private void createTables() throws SQLException {
+ Statement st = conn.createStatement();
+ st.execute("CREATE TABLE RANGER (name VARCHAR(20));");
+ st.execute("CREATE TABLE RANGES (name VARCHAR(20));");
+ st.close();
+ //need to re-initiate the tables
+ this.mysqlMixin.initTables();
+ }
+
+
+ private void clearTables() throws SQLException {
+ Set<Range> ranges = mysqlMixin.getSQLRangeSet();
+ Statement st = conn.createStatement();
+ for (Range r: ranges) {
+ try {
+ st.execute("DROP TABLE " + r + ";");
+ } catch (SQLException e) {
+ System.out.println("Trouble dropping: " + r);
+ e.printStackTrace();
+ }
+ }
+ st.close();
+ }
}
public static void close(){
//TODO: shutdown cassandra
mixin=null;
- MdbcTestUtils.cleanDatabase(DBType.POSTGRES);
+ MdbcTestUtils.stopDatabase(DBType.POSTGRES);
MdbcTestUtils.stopCassandra();
}
public static void close() throws MusicServiceException, MusicQueryException, ManagedProcessException {
//TODO: shutdown cassandra
musicMixin=null;
- MdbcTestUtils.cleanDatabase(DBType.MySQL);
+ MdbcTestUtils.stopDatabase(DBType.MySQL);
MdbcTestUtils.stopCassandra();
}
Code Review / music / mdbc.git / commitdiff