提交 246332b0 编写于 作者: C CalvinKirs

fix bug

上级 a63c2c6c
......@@ -62,6 +62,9 @@ public class ProtoStuffUtils {
public static <T> T deserialize(byte[] bytes, Class<T> clazz) {
Schema<T> schema = getSchema(clazz);
T obj = schema.newMessage();
if(null==obj){
return null;
}
ProtostuffIOUtil.mergeFrom(bytes, obj, schema);
return obj;
}
......
......@@ -17,6 +17,7 @@
package org.apache.dolphinscheduler.rpc.future;
import org.apache.dolphinscheduler.rpc.common.RpcRequest;
import org.apache.dolphinscheduler.rpc.common.RpcResponse;
import java.util.concurrent.CountDownLatch;
......@@ -34,6 +35,12 @@ public class RpcFuture implements Future<Object> {
private RpcResponse response;
private RpcRequest request;
public RpcFuture(RpcRequest rpcRequest) {
this.request = rpcRequest;
}
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
return false;
......@@ -51,13 +58,23 @@ public class RpcFuture implements Future<Object> {
@Override
public RpcResponse get() throws InterruptedException, ExecutionException {
latch.await(5, TimeUnit.SECONDS);
boolean success = latch.await(5, TimeUnit.SECONDS);
if (!success) {
throw new RuntimeException("Timeout exception. Request id: " + this.request.getRequestId()
+ ". Request class name: " + this.request.getClassName()
+ ". Request method: " + this.request.getMethodName());
}
return response;
}
@Override
public RpcResponse get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
latch.await(timeout, unit);
boolean success = latch.await(timeout, unit);
if (!success) {
throw new RuntimeException("Timeout exception. Request id: " + this.request.getRequestId()
+ ". Request class name: " + this.request.getClassName()
+ ". Request method: " + this.request.getMethodName());
}
return response;
}
......
......@@ -18,16 +18,16 @@
package org.apache.dolphinscheduler.rpc.remote;
import org.apache.dolphinscheduler.remote.config.NettyClientConfig;
import org.apache.dolphinscheduler.rpc.codec.NettyDecoder;
import org.apache.dolphinscheduler.rpc.codec.NettyEncoder;
import org.apache.dolphinscheduler.remote.utils.Constants;
import org.apache.dolphinscheduler.remote.utils.Host;
import org.apache.dolphinscheduler.remote.utils.NettyUtils;
import org.apache.dolphinscheduler.rpc.client.RpcRequestCache;
import org.apache.dolphinscheduler.rpc.client.RpcRequestTable;
import org.apache.dolphinscheduler.rpc.codec.NettyDecoder;
import org.apache.dolphinscheduler.rpc.codec.NettyEncoder;
import org.apache.dolphinscheduler.rpc.common.RpcRequest;
import org.apache.dolphinscheduler.rpc.common.RpcResponse;
import org.apache.dolphinscheduler.rpc.future.RpcFuture;
import org.apache.dolphinscheduler.remote.utils.Constants;
import org.apache.dolphinscheduler.remote.utils.Host;
import org.apache.dolphinscheduler.remote.utils.NettyUtils;
import java.net.InetSocketAddress;
import java.util.concurrent.ConcurrentHashMap;
......@@ -200,7 +200,7 @@ public class NettyClient {
rpcRequestCache.setServiceName(serviceName);
RpcFuture future = null;
if (!async) {
future = new RpcFuture();
future = new RpcFuture(request);
rpcRequestCache.setRpcFuture(future);
}
RpcRequestTable.put(request.getRequestId(), rpcRequestCache);
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册