add TsoBatchUsedUp region error handler

Signed-off-by: iosmanthus <myosmanthustree@gmail.com>

src/main/java/org/tikv/common/operation/RegionErrorHandler.java

@@ -31,6 +31,7 @@ import org.tikv.common.region.RegionErrorReceiver;
 import org.tikv.common.region.RegionManager;
 import org.tikv.common.region.TiRegion;
 import org.tikv.common.util.BackOffFunction;
+import org.tikv.common.util.BackOffFunction.BackOffFuncType;
 import org.tikv.common.util.BackOffer;
 import org.tikv.kvproto.Errorpb;
 import org.tikv.kvproto.Metapb;
@@ -168,6 +169,12 @@ public class RegionErrorHandler<RespT> implements ErrorHandler<RespT> {
       regionManager.clearRegionCache();
       throw new StatusRuntimeException(Status.UNKNOWN.withDescription(error.toString()));
     }
+    // The tso cache is used up in TiKV servers, we should backoff and wait its cache is renewed.
+    else if (error.getMessage().contains("TsoBatchUsedUp")) {
+      logger.warn(String.format("tso batch used up for region [%s]", recv.getRegion()));
+      backOffer.doBackOff(BackOffFuncType.BoTsoBatchUsedUp, new GrpcException(error.getMessage()));
+      return true;
+    }
 
     logger.warn(String.format("Unknown error %s for region [%s]", error, recv.getRegion()));
     // For other errors, we only drop cache here.

src/main/java/org/tikv/common/util/BackOffFunction.java

@@ -81,6 +81,7 @@ public class BackOffFunction {
     BoServerBusy,
     BoTxnNotFound,
     BoCheckTimeout,
-    BoCheckHealth
+    BoCheckHealth,
+    BoTsoBatchUsedUp
   }
 }

src/main/java/org/tikv/common/util/ConcreteBackOffer.java

@@ -174,6 +174,13 @@ public class ConcreteBackOffer implements BackOffer {
       case BoCheckHealth:
         backOffFunction = BackOffFunction.create(100, 600, BackOffStrategy.EqualJitter);
         break;
+      case BoTsoBatchUsedUp:
+        backOffFunction =
+            BackOffFunction.create(
+                TiConfiguration.getInt(TIKV_BO_REGION_MISS_BASE_IN_MS),
+                500,
+                BackOffStrategy.NoJitter);
+        break;
     }
     return backOffFunction;
   }

src/test/java/org/tikv/common/TsoBatchUsedUpTest.java

@@ -0,0 +1,29 @@
+package org.tikv.common;
+
+import com.google.protobuf.ByteString;
+import org.junit.Assert;
+import org.junit.Test;
+import org.tikv.kvproto.Errorpb.Error;
+import org.tikv.raw.RawKVClient;
+
+public class TsoBatchUsedUpTest extends MockThreeStoresTest {
+  RawKVClient createClient() {
+    return session.createRawClient();
+  }
+
+  @Test
+  public void testTsoBatchUsedUp() {
+    ByteString key = ByteString.copyFromUtf8("tso");
+    servers.get(0).putError("tso", () -> Error.newBuilder().setMessage("TsoBatchUsedUp"));
+    try (RawKVClient client = createClient()) {
+      try {
+        client.put(key, ByteString.EMPTY);
+        Assert.fail();
+      } catch (Exception ignore) {
+      }
+      pdServers.get(0).addGetRegionListener(request -> null);
+      // Will not clean region cache
+      Assert.assertNotNull(session.getRegionManager().getRegionByKey(key));
+    }
+  }
+}