[to #773] implement UpdateServiceGCSafePoint (#723)

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

@@ -104,6 +104,7 @@
 import org.tikv.kvproto.Pdpb.Timestamp;
 import org.tikv.kvproto.Pdpb.TsoRequest;
 import org.tikv.kvproto.Pdpb.TsoResponse;
+import org.tikv.kvproto.Pdpb.UpdateServiceGCSafePointRequest;
 
 public class PDClient extends AbstractGRPCClient<PDBlockingStub, PDFutureStub>
     implements ReadOnlyPDClient {
@@ -383,6 +384,17 @@ private Supplier<GetAllStoresRequest> buildGetAllStoresReq() {
     return () -> GetAllStoresRequest.newBuilder().setHeader(header).build();
   }
 
+  private Supplier<UpdateServiceGCSafePointRequest> buildUpdateServiceGCSafePointRequest(
+      ByteString serviceId, long ttl, long safePoint) {
+    return () ->
+        UpdateServiceGCSafePointRequest.newBuilder()
+            .setHeader(header)
+            .setSafePoint(safePoint)
+            .setServiceId(serviceId)
+            .setTTL(ttl)
+            .build();
+  }
+
   private <T> PDErrorHandler<GetStoreResponse> buildPDErrorHandler() {
     return new PDErrorHandler<>(
         r -> r.getHeader().hasError() ? buildFromPdpbError(r.getHeader().getError()) : null, this);
@@ -419,6 +431,20 @@ public TiConfiguration.ReplicaRead getReplicaRead() {
     return conf.getReplicaRead();
   }
 
+  @Override
+  public Long updateServiceGCSafePoint(
+      String serviceId, long ttl, long safePoint, BackOffer backOffer) {
+    return callWithRetry(
+            backOffer,
+            PDGrpc.getUpdateServiceGCSafePointMethod(),
+            buildUpdateServiceGCSafePointRequest(
+                ByteString.copyFromUtf8(serviceId), ttl, safePoint),
+            new PDErrorHandler<>(
+                r -> r.getHeader().hasError() ? buildFromPdpbError(r.getHeader().getError()) : null,
+                this))
+        .getMinSafePoint();
+  }
+
   @Override
   public void close() throws InterruptedException {
     etcdClient.close();

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

@@ -72,4 +72,16 @@ List<Pdpb.Region> scanRegions(
   Long getClusterId();
 
   RequestKeyCodec getCodec();
+
+  /**
+   * Update ServiceGCSafePoint
+   *
+   * @param serviceId ServiceId
+   * @param ttl TTL in seconds
+   * @param safePoint The TiTimestamp you want to set. Set to start_ts.getPrevious() is a good
+   *     practice
+   * @return the MinSafePoint of all services. If this value is greater than safePoint, it means
+   *     update failed
+   */
+  Long updateServiceGCSafePoint(String serviceId, long ttl, long safePoint, BackOffer backOffer);
 }