Update HBaseResourceStore.java

Avoid resource inconsistent caused by hbase rpc timeout
saveORwaste · Oct 12, 2018 · 562e88d · 562e88d
1 parent 8581322
commit 562e88d
Showing 1 changed file with 14 additions and 6 deletions.
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseResourceStore.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseResourceStore.java
@@ -43,6 +43,7 @@
 import org.apache.hadoop.hbase.client.ResultScanner;
 import org.apache.hadoop.hbase.client.Scan;
 import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.client.RetriesExhaustedException;
 import org.apache.hadoop.hbase.filter.CompareFilter;
 import org.apache.hadoop.hbase.filter.Filter;
 import org.apache.hadoop.hbase.filter.FilterList;
@@ -317,13 +318,20 @@ protected long checkAndPutResourceImpl(String resPath, byte[] content, long oldT
             byte[] bOldTS = oldTS == 0 ? null : Bytes.toBytes(oldTS);
             Put put = buildPut(resPath, newTS, row, content, table);
 
-            boolean ok = table.checkAndPut(row, B_FAMILY, B_COLUMN_TS, bOldTS, put);
-            logger.trace("Update row " + resPath + " from oldTs: " + oldTS + ", to newTs: " + newTS
-                    + ", operation result: " + ok);
-            if (!ok) {
+            try {
+                boolean ok = table.checkAndPut(row, B_FAMILY, B_COLUMN_TS, bOldTS, put);
+                logger.trace("Update row " + resPath + " from oldTs: " + oldTS + ", to newTs: " + newTS + ", operation result: " + ok);
+                if (!ok) {
+                    long real = getResourceTimestampImpl(resPath);
+                    throw new WriteConflictException(
+                            "Overwriting conflict " + resPath + ", expect old TS " + oldTS + ", but it is " + real);
+                }
+            } catch (RetriesExhaustedException e){
                 long real = getResourceTimestampImpl(resPath);
-                throw new WriteConflictException(
-                        "Overwriting conflict " + resPath + ", expect old TS " + oldTS + ", but it is " + real);
+                // rpc timeout but resource has been already updated
+                if(newTS != real){
+                    throw e;
+                }
             }
 
             return newTS;