fix bug

Author: miyuan-ljr

pom.xml

@@ -45,7 +45,7 @@
     </distributionManagement>
 
     <properties>
-        <hadoop.version>2.5.1</hadoop.version>
+        <hadoop.version>2.7.7</hadoop.version>
         <hbase.version>2.0.6</hbase.version>
         <java.source.version>1.8</java.source.version>
         <java.target.version>1.8</java.target.version>
@@ -103,10 +103,6 @@
                     <artifactId>jettison</artifactId>
                     <groupId>org.codehaus.jettison</groupId>
                 </exclusion>
-                <exclusion>
-                    <artifactId>commons-collections</artifactId>
-                    <groupId>commons-collections</groupId>
-                </exclusion>
             </exclusions>
         </dependency>
         <dependency>

src/main/java/com/alipay/oceanbase/hbase/OHTable.java

@@ -750,6 +750,12 @@ private void doPut(List<Put> puts) throws IOException {
 
             // we need to periodically see if the writebuffer is full instead of waiting until the end of the List
             n++;
+            if (n % putWriteBufferCheck == 0 && currentWriteBufferSize > writeBufferSize) {
+                flushCommits();
+            }
+        }
+        if (autoFlush || currentWriteBufferSize > writeBufferSize) {
+            flushCommits();
         }
     }
 
@@ -1142,6 +1148,106 @@ public long incrementColumnValue(byte[] row, byte[] family, byte[] qualifier, lo
         return incrementColumnValue(row, family, qualifier, amount);
     }
 
+    public void flushCommits() throws IOException {
+
+        try {
+            boolean[] resultSuccess = new boolean[writeBuffer.size()];
+            try {
+                Map<String, Pair<List<Integer>, List<Cell>>> familyMap = new HashMap<String, Pair<List<Integer>, List<Cell>>>();
+                for (int i = 0; i < writeBuffer.size(); i++) {
+                    Put aPut = writeBuffer.get(i);
+                    Map<byte[], List<Cell>> innerFamilyMap = aPut.getFamilyCellMap();
+                    if (innerFamilyMap.size() > 1) {
+                        // Bypass logic: directly construct BatchOperation for puts with family map size > 1
+                        try {
+                            BatchOperation batch = buildBatchOperation(this.tableNameString,
+                                    innerFamilyMap, false, null);
+                            BatchOperationResult results = batch.execute();
+
+                            boolean hasError = results.hasError();
+                            resultSuccess[i] = !hasError;
+                            if (hasError) {
+                                throw results.getFirstException();
+                            }
+                        } catch (Exception e) {
+                            logger.error(LCD.convert("01-00008"), tableNameString, null, autoFlush,
+                                    writeBuffer.size(), e);
+                            throw new IOException("put table " + tableNameString + " error codes "
+                                    + null + "auto flush " + autoFlush
+                                    + " current buffer size " + writeBuffer.size(), e);
+                        }
+                    } else {
+                        // Existing logic for puts with family map size = 1
+                        for (Map.Entry<byte[], List<Cell>> entry : innerFamilyMap.entrySet()) {
+                            String family = Bytes.toString(entry.getKey());
+                            Pair<List<Integer>, List<Cell>> keyValueWithIndex = familyMap
+                                    .get(family);
+                            if (keyValueWithIndex == null) {
+                                keyValueWithIndex = new Pair<List<Integer>, List<Cell>>(
+                                        new ArrayList<Integer>(), new ArrayList<Cell>());
+                                familyMap.put(family, keyValueWithIndex);
+                            }
+                            keyValueWithIndex.getFirst().add(i);
+                            keyValueWithIndex.getSecond().addAll(entry.getValue());
+                        }
+                    }
+                }
+                for (Map.Entry<String, Pair<List<Integer>, List<Cell>>> entry : familyMap
+                        .entrySet()) {
+                    List<Integer> errorCodeList = new ArrayList<Integer>(entry.getValue()
+                            .getSecond().size());
+                    try {
+                        String targetTableName = getTargetTableName(this.tableNameString,
+                                entry.getKey(), configuration);
+
+                        BatchOperation batch = buildBatchOperation(targetTableName, entry
+                                .getValue().getSecond(), false, null);
+                        BatchOperationResult results = batch.execute();
+
+                        errorCodeList = results.getErrorCodeList();
+                        boolean hasError = results.hasError();
+                        for (Integer index : entry.getValue().getFirst()) {
+                            resultSuccess[index] = !hasError;
+                        }
+                        if (hasError) {
+                            throw results.getFirstException();
+                        }
+                    } catch (Exception e) {
+                        logger.error(LCD.convert("01-00008"), tableNameString, errorCodeList,
+                                autoFlush, writeBuffer.size(), e);
+                        throw new IOException("put table " + tableNameString + " error codes "
+                                + errorCodeList + "auto flush " + autoFlush
+                                + " current buffer size " + writeBuffer.size(), e);
+                    }
+
+                }
+
+            } finally {
+                // mutate list so that it is empty for complete success, or contains
+                // only failed records results are returned in the same order as the
+                // requests in list walk the list backwards, so we can remove from list
+                // without impacting the indexes of earlier members
+                for (int i = resultSuccess.length - 1; i >= 0; i--) {
+                    if (resultSuccess[i]) {
+                        // successful Puts are removed from the list here.
+                        writeBuffer.remove(i);
+                    }
+                }
+            }
+        } finally {
+            if (clearBufferOnFail) {
+                writeBuffer.clear();
+                currentWriteBufferSize = 0;
+            } else {
+                // the write buffer was adjusted by processBatchOfPuts
+                currentWriteBufferSize = 0;
+                for (Put aPut : writeBuffer) {
+                    currentWriteBufferSize += aPut.heapSize();
+                }
+            }
+        }
+    }
+
     @Override
     public void close() throws IOException {
         if (cleanupPoolOnClose) {

src/main/java/com/alipay/oceanbase/hbase/util/OHConnectionImpl.java

@@ -56,7 +56,7 @@ public class OHConnectionImpl implements Connection {
 
     private final OHConnectionConfiguration connectionConfig;
 
-    OHConnectionImpl(Configuration conf, final boolean managed, ExecutorService pool,
+    OHConnectionImpl(Configuration conf, ExecutorService pool,
                      final User user) throws IOException {
         this.conf = conf;
         this.batchPool = pool;

src/test/java/com/alipay/oceanbase/hbase/HTableTestBase.java

@@ -105,7 +105,7 @@ PRIMARY KEY (`K`, `Q`, `T`)
         }
 
         // test scan with empty family
-        Scan scan = new Scan();
+        Scan scan = new Scan(toBytes(key));
         ResultScanner scanner = hTable.getScanner(scan);
         for (Result result : scanner) {
             for (Cell keyValue : result.rawCells()) {
@@ -189,10 +189,8 @@ private void testBasic(String family) throws Exception {
         try {
             for (int j = 0; j < 10; j++) {
                 put = new Put((key + "_" + j).getBytes());
-                kv = new KeyValue(toBytes(key), family.getBytes(), column1.getBytes(), timestamp + 2, toBytes(value));
-                put.add(kv);
-                kv = new KeyValue(toBytes(key), family.getBytes(), column2.getBytes(), timestamp + 2, toBytes(value));
-                put.add(kv);
+                put.addColumn(family.getBytes(), column1.getBytes(), timestamp + 2, toBytes(value));
+                put.addColumn(family.getBytes(), column2.getBytes(), timestamp + 2, toBytes(value));
                 hTable.put(put);
             }
 
@@ -230,6 +228,7 @@ private void testBasic(String family) throws Exception {
             // scan.setBatch(1);
 
             scan.setMaxVersions(9);
+            scanner = hTable.getScanner(scan);
             i = 0;
             count = 0;
             for (Result result : scanner) {
@@ -1041,7 +1040,7 @@ public void testRowRangeFilter() throws Exception {
         for (Result result : scanner) {
             for (Cell keyValue : result.rawCells()) {
                 System.out.printf("Rowkey: %s, Column Family: %s, Column Qualifier: %s, Timestamp: %d, Value: %s%n",
-                        Bytes.toString(result.getRow()),
+                        Bytes.toString(CellUtil.cloneRow(keyValue)),
                         Bytes.toString(CellUtil.cloneFamily(keyValue)),
                         Bytes.toString(CellUtil.cloneQualifier(keyValue)),
                         keyValue.getTimestamp(),
@@ -3204,13 +3203,6 @@ public void testGet() throws Exception {
         // | zScanKey2 | column1 | -1729223352450 | value1 |
         // +-----------+---------+----------------+--------+
 
-        // test closestRowBefore
-        get = new Get("scanKey2x2".getBytes());
-        get.addFamily(family.getBytes());
-        get.setClosestRowBefore(true);
-        r = hTable.get(get);
-        assertEquals(key2, Bytes.toString(r.getRow()));
-
         // test exists
         LinkedList<Get> gets = new LinkedList<>();
         Get get1 = new Get(key1.getBytes());
@@ -5291,9 +5283,8 @@ public void testPutColumnFamilyNotExists() throws Exception {
     @Test
     public void testGetColumnFamilyNull() throws Exception {
         Get get = new Get(("key_c_f").getBytes());
-        expectedException.expect(NullPointerException.class);
         get.addFamily(null);
-        expectedException.expect(FeatureNotSupportedException.class);
+        expectedException.expect(IllegalArgumentException.class);
         expectedException.expectMessage("family is empty");
         hTable.get(get);
     }