HBASE-24401 Cell size limit check on append should consider 0 or less value to disable the check (#1742)

Signed-off-by: Guanghao Zhang <zghao@apache.org>

Author: wenbang

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java

@@ -8219,14 +8219,14 @@ private List<Cell> reckonDeltasByStore(HStore store, Operation op, Mutation muta
           break;
         default: throw new UnsupportedOperationException(op.toString());
       }
-      int newCellSize = PrivateCellUtil.estimatedSerializedSizeOf(newCell);
-      if (newCellSize > this.maxCellSize) {
-        String msg = "Cell with size " + newCellSize + " exceeds limit of " + this.maxCellSize
-          + " bytes in region " + this;
-        if (LOG.isDebugEnabled()) {
+      if (this.maxCellSize > 0) {
+        int newCellSize = PrivateCellUtil.estimatedSerializedSizeOf(newCell);
+        if (newCellSize > this.maxCellSize) {
+          String msg = "Cell with size " + newCellSize + " exceeds limit of " + this.maxCellSize
+            + " bytes in region " + this;
           LOG.debug(msg);
+          throw new DoNotRetryIOException(msg);
         }
-        throw new DoNotRetryIOException(msg);
       }
       cellPairs.add(new Pair<>(currentValue, newCell));
       // Add to results to get returned to the Client. If null, cilent does not want results.

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java

@@ -973,9 +973,7 @@ private void checkCellSizeLimit(final HRegion r, final Mutation m) throws IOExce
         int size = PrivateCellUtil.estimatedSerializedSizeOf(cells.current());
         if (size > r.maxCellSize) {
           String msg = "Cell with size " + size + " exceeds limit of " + r.maxCellSize + " bytes";
-          if (LOG.isDebugEnabled()) {
-            LOG.debug(msg);
-          }
+          LOG.debug(msg);
           throw new DoNotRetryIOException(msg);
         }
       }

hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestFromClientSide5.java

@@ -2235,7 +2235,7 @@ public void testFilterAllRecords() throws IOException {
 
   @Test
   public void testCellSizeLimit() throws IOException {
-    final TableName tableName = TableName.valueOf("testCellSizeLimit");
+    final TableName tableName = name.getTableName();
     TableDescriptorBuilder.ModifyableTableDescriptor tableDescriptor =
       new TableDescriptorBuilder.ModifyableTableDescriptor(tableName)
         .setValue(HRegion.HBASE_MAX_CELL_SIZE_KEY, Integer.toString(10 * 1024));
@@ -2272,6 +2272,28 @@ public void testCellSizeLimit() throws IOException {
     }
   }
 
+  @Test
+  public void testCellSizeNoLimit() throws IOException {
+    final TableName tableName = name.getTableName();
+    ColumnFamilyDescriptor familyDescriptor =
+      new ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor(FAMILY);
+    TableDescriptorBuilder.ModifyableTableDescriptor tableDescriptor =
+      new TableDescriptorBuilder.ModifyableTableDescriptor(tableName)
+        .setValue(HRegion.HBASE_MAX_CELL_SIZE_KEY, Integer.toString(0));
+    tableDescriptor.setColumnFamily(familyDescriptor);
+
+    try (Admin admin = TEST_UTIL.getAdmin()) {
+      admin.createTable(tableDescriptor);
+    }
+
+    // Will succeed
+    try (Table ht = TEST_UTIL.getConnection().getTable(tableName)) {
+      ht.put(new Put(ROW).addColumn(FAMILY, QUALIFIER,  new byte[HRegion.DEFAULT_MAX_CELL_SIZE -
+        1024]));
+      ht.append(new Append(ROW).addColumn(FAMILY, QUALIFIER, new byte[1024 + 1]));
+    }
+  }
+
   @Test
   public void testDeleteSpecifiedVersionOfSpecifiedColumn() throws Exception {
     final TableName tableName = name.getTableName();