add configration parameter for RawKV timeout (#246)

README.md

@@ -132,6 +132,9 @@ The following includes ThreadPool related parameters, which can be passed in thr
 - the thread pool size of deleteRange on client side
 - default: 20
 
+#### tikv.rawkv.default_backoff_in_ms
+- RawKV default backoff in milliseconds
+- default: 20000 (20 seconds)
 
 ## Metrics
 

src/main/java/org/tikv/common/ConfigUtils.java

@@ -15,6 +15,7 @@
 
 package org.tikv.common;
 
+import org.tikv.common.util.BackOffer;
 import org.tikv.kvproto.Kvrpcpb;
 
 public class ConfigUtils {
@@ -54,6 +55,8 @@ public class ConfigUtils {
   public static final String TIKV_HEALTH_CHECK_PERIOD_DURATION =
       "tikv.health_check_period_duration";
 
+  public static final String TIKV_RAWKV_DEFAULT_BACKOFF_IN_MS = "tikv.rawkv.default_backoff_in_ms";
+
   public static final String DEF_PD_ADDRESSES = "127.0.0.1:2379";
   public static final String DEF_TIMEOUT = "200ms";
   public static final String DEF_FORWARD_TIMEOUT = "300ms";
@@ -86,6 +89,8 @@ public class ConfigUtils {
   public static final String DEF_TIKV_NETWORK_MAPPING_NAME = "";
   public static final boolean DEF_GRPC_FORWARD_ENABLE = true;
 
+  public static final int DEF_TIKV_RAWKV_DEFAULT_BACKOFF_IN_MS = BackOffer.RAWKV_MAX_BACKOFF;
+
   public static final String NORMAL_COMMAND_PRIORITY = "NORMAL";
   public static final String LOW_COMMAND_PRIORITY = "LOW";
   public static final String HIGH_COMMAND_PRIORITY = "HIGH";

src/main/java/org/tikv/common/TiConfiguration.java

@@ -81,6 +81,7 @@ private static void loadFromDefaultProperties() {
     setIfMissing(TIKV_ENABLE_GRPC_FORWARD, DEF_GRPC_FORWARD_ENABLE);
     setIfMissing(TIKV_GRPC_HEALTH_CHECK_TIMEOUT, DEF_CHECK_HEALTH_TIMEOUT);
     setIfMissing(TIKV_HEALTH_CHECK_PERIOD_DURATION, DEF_HEALTH_CHECK_PERIOD_DURATION);
+    setIfMissing(TIKV_RAWKV_DEFAULT_BACKOFF_IN_MS, DEF_TIKV_RAWKV_DEFAULT_BACKOFF_IN_MS);
   }
 
   public static void listAll() {
@@ -270,6 +271,8 @@ private static ReplicaRead getReplicaRead(String key) {
   private final String networkMappingName = get(TIKV_NETWORK_MAPPING_NAME);
   private HostMapping hostMapping = null;
 
+  private int rawKVDefaultBackoffInMS = getInt(TIKV_RAWKV_DEFAULT_BACKOFF_IN_MS);
+
   public enum KVMode {
     TXN,
     RAW
@@ -575,4 +578,12 @@ public long getGrpcHealthCheckTimeout() {
   public long getHealthCheckPeriodDuration() {
     return this.healthCheckPeriodDuration;
   }
+
+  public int getRawKVDefaultBackoffInMS() {
+    return rawKVDefaultBackoffInMS;
+  }
+
+  public void setRawKVDefaultBackoffInMS(int rawKVDefaultBackoffInMS) {
+    this.rawKVDefaultBackoffInMS = rawKVDefaultBackoffInMS;
+  }
 }

src/main/java/org/tikv/common/TiSession.java

@@ -412,7 +412,7 @@ public void splitRegionAndScatter(
     long endMS = System.currentTimeMillis();
     logger.info("splitRegionAndScatter cost {} seconds", (endMS - startMS) / 1000);
   }
-  
+
   private List<Metapb.Region> splitRegion(List<ByteString> splitKeys, BackOffer backOffer) {
     List<Metapb.Region> regions = new ArrayList<>();
 

src/main/java/org/tikv/raw/RawKVClient.java

@@ -235,7 +235,7 @@ private void batchPut(Map<ByteString, ByteString> kvPairs, long ttl, boolean ato
     String label = "client_raw_batch_put";
     Histogram.Timer requestTimer = RAW_REQUEST_LATENCY.labels(label).startTimer();
     try {
-      doSendBatchPut(ConcreteBackOffer.newRawKVBackOff(), kvPairs, ttl, atomic);
+      doSendBatchPut(defaultBackOff(), kvPairs, ttl, atomic);
       RAW_REQUEST_SUCCESS.labels(label).inc();
     } catch (Exception e) {
       RAW_REQUEST_FAILURE.labels(label).inc();
@@ -866,6 +866,6 @@ private Iterator<KvPair> rawScanIterator(
   }
 
   private BackOffer defaultBackOff() {
-    return ConcreteBackOffer.newRawKVBackOff();
+    return ConcreteBackOffer.newCustomBackOff(conf.getRawKVDefaultBackoffInMS());
   }
 }