[Fix] table direct load rpc timeout

src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/ObAddr.java

@@ -42,6 +42,15 @@ public VER getVersion() {
         return version;
     }
 
+    public String getIPString() {
+        String ipStr = new String();
+        if (version == VER.IPV4) {
+            ipStr = String.format("%d.%d.%d.%d", (ip[0] >> 24) & 0XFF, (ip[0] >> 16) & 0XFF,
+                (ip[0] >> 8) & 0XFF, (ip[0]) & 0XFF);
+        }
+        return ipStr;
+    }
+
     public int getPort() {
         return port;
     }
@@ -50,6 +59,14 @@ public void setPort(int port) {
         this.port = port;
     }
 
+    public String toString() {
+        String str = new String();
+        if (version == VER.IPV4) {
+            str = String.format("%s:%d", getIPString(), port);
+        }
+        return str;
+    }
+
     @Override
     public byte[] encode() {
         int needBytes = (int) getEncodedSize();

src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/direct_load/ObTableDirectLoadRequest.java

@@ -25,6 +25,7 @@
 import com.alipay.oceanbase.rpc.util.ObByteBuf;
 import com.alipay.oceanbase.rpc.util.ObBytesString;
 import com.alipay.oceanbase.rpc.util.Serialization;
+import com.alipay.oceanbase.rpc.util.TraceUtil;
 
 import io.netty.buffer.ByteBuf;
 
@@ -43,6 +44,10 @@ public ObTableDirectLoadRequest() {
         setVersion(2);
     }
 
+    public void setTimeout(long timeout) {
+        this.timeout = timeout;
+    }
+
     public Header getHeader() {
         return header;
     }
@@ -71,6 +76,11 @@ public void setArgContent(ObBytesString argContent) {
         this.argContent = argContent;
     }
 
+    public String toString() {
+        return String.format("%s {header:%s, argContent:%d}", TraceUtil.formatTraceMessage(this),
+            header, argContent.length());
+    }
+
     /*
      * Get pcode.
      */
@@ -138,6 +148,10 @@ public void setOperationType(ObTableDirectLoadOperationType operationType) {
             this.operationType = operationType;
         }
 
+        public String toString() {
+            return String.format("{addr:%s, operationType:%s}", addr, operationType);
+        }
+
         /**
          * Encode.
          */

src/main/java/com/alipay/oceanbase/rpc/table/ObTableDirectLoad.java

@@ -42,6 +42,8 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+
+import java.util.Properties;
 import java.util.Timer;
 import java.util.TimerTask;
 
@@ -92,6 +94,12 @@ public ObTableDirectLoad(ObTable table, String tableName, ObDirectLoadParameter
         initProperties();
     }
 
+    @Override
+    public void setProperties(Properties properties) {
+        this.properties = properties;
+        initProperties();
+    }
+
     public ObTable getTable() {
         return table;
     }
@@ -338,14 +346,15 @@ private void execute(ObTableDirectLoadOperationType operationType, ObSimplePaylo
         }
     }
 
-    private ObPayload rpcCall(ObPayload request) throws Exception {
+    private ObPayload rpcCall(ObTableDirectLoadRequest request) throws Exception {
         int tries = 0;
         while (true) {
             try {
+                request.setTimeout(table.getObTableExecuteTimeout());
                 return table.execute(request);
             } catch (Exception e) {
-                logger.warn(String.format("table execute failed, table:%s, tries:%d", this, tries),
-                    e);
+                logger.warn(String.format("table execute failed, table:%s, tries:%d, request:%s",
+                    this, tries, request), e);
                 if (tries < runtimeRetryTimes) {
                     Thread.sleep(runtimeRetryInterval);
                 } else {