public boolean addVersion ( String cluster , String user , String appId , String version , long timestamp ) throws IOException { boolean updated = false ; // check if the version already exists byte [ ] rowKey = getRowKey ( cluster , user , appId ) ; byte [ ] versionCol = Bytes . toBytes ( version ) ; int attempts = 0 ; // retry up to this many times for checkAndPut failures int maxAttempts = 3 ; boolean checkForUpdate = true ; while ( checkForUpdate && attempts < maxAttempts ) { attempts ++ ; // values for conditional update Put p = null ; byte [ ] expectedValue = null ; Get get = new Get ( rowKey ) ; get . addColumn ( Constants . INFO_FAM_BYTES , versionCol ) ; Table versionsTable = null ; try { versionsTable = hbaseConnection . getTable ( TableName . valueOf ( Constants . HISTORY_APP_VERSION_TABLE ) ) ; Result r = versionsTable . get ( get ) ; if ( r != null && ! r . isEmpty ( ) ) { byte [ ] storedValue = r . getValue ( Constants . INFO_FAM_BYTES , versionCol ) ; long storedTS = Bytes . toLong ( storedValue ) ; if ( timestamp < storedTS ) { // update the stored timestamp to our earlier value p = new Put ( rowKey ) ; p . addColumn ( Constants . INFO_FAM_BYTES , versionCol , Bytes . toBytes ( timestamp ) ) ; expectedValue = storedValue ; } else { // version exists and exceeds our value, no update necessary checkForUpdate = false ; } } else { // no stored value p = new Put ( rowKey ) ; p . addColumn ( Constants . INFO_FAM_BYTES , versionCol , Bytes . toBytes ( timestamp ) ) ; } if ( p != null ) { // we have an updated value to add updated = versionsTable . checkAndPut ( rowKey , Constants . INFO_FAM_BYTES , versionCol , expectedValue , p ) ; checkForUpdate = ! updated ; if ( ! updated ) { LOG . warn ( "Update of cluster=" + cluster + ", user=" + user + ", app=" + appId + ", version=" + version + " to timestamp " + timestamp + " failed because currently set value changed!" + " (attempt " + attempts + " of " + maxAttempts + ")" ) ; } } } finally { if ( versionsTable != null ) { versionsTable . close ( ) ; } } } return updated ; }
Adds an entry for the given version if it does not already exist . If the given timestamp is earlier than the currently stored timestamp for the version it will be updated .