You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by ji...@apache.org on 2015/04/29 03:43:22 UTC
[1/4] tajo git commit: TAJO-1563: Improve RPC error handling. (jinho)
Repository: tajo
Updated Branches:
refs/heads/index_support 2fe3db4c5 -> e6b91c333
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/NettyClientBase.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/NettyClientBase.java b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/NettyClientBase.java
index 57e436b..190beae 100644
--- a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/NettyClientBase.java
+++ b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/NettyClientBase.java
@@ -18,40 +18,49 @@
package org.apache.tajo.rpc;
+import com.google.protobuf.Descriptors;
+import com.google.protobuf.Message;
+import com.google.protobuf.ServiceException;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.timeout.IdleState;
+import io.netty.handler.timeout.IdleStateEvent;
+import io.netty.util.concurrent.GenericFutureListener;
+import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.tajo.rpc.RpcClientManager.RpcConnectionKey;
+import org.apache.tajo.rpc.RpcProtos.RpcResponse;
import java.io.Closeable;
import java.lang.reflect.Method;
+import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.Collection;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
-public abstract class NettyClientBase implements Closeable {
- private static final Log LOG = LogFactory.getLog(NettyClientBase.class);
- private static final int CONNECTION_TIMEOUT = 60000; // 60 sec
- private static final long PAUSE = 1000; // 1 sec
-
- private final int numRetries;
+public abstract class NettyClientBase<T> implements ProtoDeclaration, Closeable {
+ public final static Log LOG = LogFactory.getLog(NettyClientBase.class);
private Bootstrap bootstrap;
private volatile ChannelFuture channelFuture;
-
- protected final Class<?> protocol;
- protected final AtomicInteger sequence = new AtomicInteger(0);
-
private final RpcConnectionKey key;
+ private final int maxRetries;
+ private boolean enableMonitor;
+
+ private final ConcurrentMap<RpcConnectionKey, ChannelEventListener> channelEventListeners =
+ new ConcurrentHashMap<RpcConnectionKey, ChannelEventListener>();
+ private final ConcurrentMap<Integer, T> requests = new ConcurrentHashMap<Integer, T>();
public NettyClientBase(RpcConnectionKey rpcConnectionKey, int numRetries)
throws ClassNotFoundException, NoSuchMethodException {
this.key = rpcConnectionKey;
- this.protocol = rpcConnectionKey.protocolClass;
- this.numRetries = numRetries;
+ this.maxRetries = numRetries;
}
// should be called from sub class
@@ -59,13 +68,13 @@ public abstract class NettyClientBase implements Closeable {
this.bootstrap = new Bootstrap();
this.bootstrap
.group(RpcChannelFactory.getSharedClientEventloopGroup())
- .channel(NioSocketChannel.class)
- .handler(initializer)
- .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
- .option(ChannelOption.SO_REUSEADDR, true)
- .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, CONNECTION_TIMEOUT)
- .option(ChannelOption.SO_RCVBUF, 1048576 * 10)
- .option(ChannelOption.TCP_NODELAY, true);
+ .channel(NioSocketChannel.class)
+ .handler(initializer)
+ .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
+ .option(ChannelOption.SO_REUSEADDR, true)
+ .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, RpcConstants.DEFAULT_CONNECT_TIMEOUT)
+ .option(ChannelOption.SO_RCVBUF, 1048576 * 10)
+ .option(ChannelOption.TCP_NODELAY, true);
}
public RpcClientManager.RpcConnectionKey getKey() {
@@ -73,23 +82,81 @@ public abstract class NettyClientBase implements Closeable {
}
protected final Class<?> getServiceClass() throws ClassNotFoundException {
- String serviceClassName = protocol.getName() + "$" + protocol.getSimpleName() + "Service";
+ String serviceClassName = getKey().protocolClass.getName() + "$" +
+ getKey().protocolClass.getSimpleName() + "Service";
return Class.forName(serviceClassName);
}
@SuppressWarnings("unchecked")
- protected final <T> T getStub(Method stubMethod, Object rpcChannel) {
+ protected final <I> I getStub(Method stubMethod, Object rpcChannel) {
try {
- return (T) stubMethod.invoke(null, rpcChannel);
+ return (I) stubMethod.invoke(null, rpcChannel);
} catch (Exception e) {
throw new RemoteException(e.getMessage(), e);
}
}
- public abstract <T> T getStub();
+ protected static RpcProtos.RpcRequest buildRequest(int seqId,
+ Descriptors.MethodDescriptor method,
+ Message param) {
+ RpcProtos.RpcRequest.Builder requestBuilder = RpcProtos.RpcRequest.newBuilder()
+ .setId(seqId)
+ .setMethodName(method.getName());
+
+ if (param != null) {
+ requestBuilder.setRequestMessage(param.toByteString());
+ }
+
+ return requestBuilder.build();
+ }
+
+ /**
+ * Repeat invoke rpc request until the connection attempt succeeds or exceeded retries
+ */
+ protected void invoke(final RpcProtos.RpcRequest rpcRequest, final T callback, final int retry) {
+
+ ChannelPromise promise = getChannel().newPromise();
+ promise.addListener(new GenericFutureListener<ChannelFuture>() {
+ @Override
+ public void operationComplete(final ChannelFuture future) throws Exception {
- private InetSocketAddress resolveAddress(InetSocketAddress address) {
+ if (future.isSuccess()) {
+
+ getHandler().registerCallback(rpcRequest.getId(), callback);
+ } else {
+
+ if (!future.channel().isActive() && retry < maxRetries) {
+
+ /* schedule the current request for the retry */
+ LOG.warn(future.cause() + " Try to reconnect :" + getKey().addr);
+
+ final EventLoop loop = future.channel().eventLoop();
+ loop.schedule(new Runnable() {
+ @Override
+ public void run() {
+ doConnect(getKey().addr).addListener(new GenericFutureListener<ChannelFuture>() {
+ @Override
+ public void operationComplete(ChannelFuture future) throws Exception {
+ invoke(rpcRequest, callback, retry + 1);
+ }
+ });
+ }
+ }, RpcConstants.DEFAULT_PAUSE, TimeUnit.MILLISECONDS);
+ } else {
+
+ /* Max retry count has been exceeded or internal failure */
+ getHandler().registerCallback(rpcRequest.getId(), callback);
+ getHandler().exceptionCaught(getChannel().pipeline().lastContext(),
+ new RecoverableException(rpcRequest.getId(), future.cause()));
+ }
+ }
+ }
+ });
+ getChannel().writeAndFlush(rpcRequest, promise);
+ }
+
+ private static InetSocketAddress resolveAddress(InetSocketAddress address) {
if (address.isUnresolved()) {
return RpcUtils.createSocketAddr(address.getHostName(), address.getPort());
}
@@ -100,11 +167,10 @@ public abstract class NettyClientBase implements Closeable {
return this.channelFuture = bootstrap.clone().connect(address);
}
-
- public synchronized void connect() throws ConnectTimeoutException {
+ public synchronized void connect() throws ConnectException {
if (isConnected()) return;
- final AtomicInteger retries = new AtomicInteger();
+ int retries = 0;
InetSocketAddress address = key.addr;
if (address.isUnresolved()) {
address = resolveAddress(address);
@@ -112,22 +178,26 @@ public abstract class NettyClientBase implements Closeable {
/* do not call await() inside handler */
ChannelFuture f = doConnect(address).awaitUninterruptibly();
- retries.incrementAndGet();
- if (!f.isSuccess() && numRetries > 0) {
- doReconnect(address, f, retries);
+ if (!f.isSuccess()) {
+ if (maxRetries > 0) {
+ doReconnect(address, f, ++retries);
+ } else {
+ throw new ConnectException(ExceptionUtils.getMessage(f.cause()));
+ }
}
}
- private void doReconnect(final InetSocketAddress address, ChannelFuture future, AtomicInteger retries)
- throws ConnectTimeoutException {
+ private void doReconnect(final InetSocketAddress address, ChannelFuture future, int retries)
+ throws ConnectException {
for (; ; ) {
- if (numRetries >= retries.getAndIncrement()) {
+ if (maxRetries > retries) {
+ retries++;
- LOG.warn(future.cause().getMessage() + " Try to reconnect");
+ LOG.warn(future.cause() + " Try to reconnect : " + getKey().addr);
try {
- Thread.sleep(PAUSE);
+ Thread.sleep(RpcConstants.DEFAULT_PAUSE);
} catch (InterruptedException e) {
}
@@ -136,19 +206,21 @@ public abstract class NettyClientBase implements Closeable {
break;
}
} else {
- throw new ConnectTimeoutException("Max retry count has been exceeded. attempts=" + numRetries
+ throw new ConnectTimeoutException("Max retry count has been exceeded. attempts=" + retries
+ " caused by: " + future.cause());
}
}
}
+ protected abstract NettyChannelInboundHandler getHandler();
+
public Channel getChannel() {
return channelFuture == null ? null : channelFuture.channel();
}
public boolean isConnected() {
Channel channel = getChannel();
- return channel != null && channel.isOpen() && channel.isActive();
+ return channel != null && channel.isActive();
}
public SocketAddress getRemoteAddress() {
@@ -156,12 +228,165 @@ public abstract class NettyClientBase implements Closeable {
return channel == null ? null : channel.remoteAddress();
}
+ public int getActiveRequests() {
+ return requests.size();
+ }
+
+ public boolean subscribeEvent(RpcConnectionKey key, ChannelEventListener listener) {
+ return channelEventListeners.putIfAbsent(key, listener) == null;
+ }
+
+ public void removeSubscribers() {
+ channelEventListeners.clear();
+ }
+
+ public Collection<ChannelEventListener> getSubscribers() {
+ return channelEventListeners.values();
+ }
+
+ private String getErrorMessage(String message) {
+ return "Exception [" + getKey().protocolClass.getCanonicalName() +
+ "(" + RpcUtils.normalizeInetSocketAddress((InetSocketAddress)
+ getChannel().remoteAddress()) + ")]: " + message;
+ }
+
@Override
public void close() {
+ getHandler().sendExceptions(getClass().getSimpleName() + "terminates all the connections");
+
Channel channel = getChannel();
if (channel != null && channel.isOpen()) {
LOG.debug("Proxy will be disconnected from remote " + channel.remoteAddress());
- channel.close().awaitUninterruptibly();
+ channel.close().syncUninterruptibly();
+ }
+ }
+
+ protected abstract class NettyChannelInboundHandler extends SimpleChannelInboundHandler<RpcResponse> {
+
+ protected void registerCallback(int seqId, T callback) {
+ if (requests.putIfAbsent(seqId, callback) != null) {
+ throw new RemoteException(
+ getErrorMessage("Duplicate Sequence Id " + seqId));
+ }
+ }
+
+ @Override
+ public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
+ MonitorClientHandler handler = ctx.pipeline().get(MonitorClientHandler.class);
+ if (handler != null) {
+ enableMonitor = true;
+ }
+
+ for (ChannelEventListener listener : getSubscribers()) {
+ listener.channelRegistered(ctx);
+ }
+ super.channelRegistered(ctx);
+ }
+
+ @Override
+ public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
+ for (ChannelEventListener listener : getSubscribers()) {
+ listener.channelUnregistered(ctx);
+ }
+ super.channelUnregistered(ctx);
+
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx) throws Exception {
+ super.channelActive(ctx);
+ LOG.debug("Connection established successfully : " + ctx.channel());
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ super.channelInactive(ctx);
+ sendExceptions("Connection lost :" + getKey().addr);
+ }
+
+ @Override
+ protected final void channelRead0(ChannelHandlerContext ctx, RpcResponse response) throws Exception {
+ T callback = requests.remove(response.getId());
+ if (callback == null)
+ LOG.warn("Dangling rpc call");
+ else run(response, callback);
+ }
+
+ /**
+ * A {@link #channelRead0} received a message.
+ * @param response response proto of type {@link RpcResponse}.
+ * @param callback callback of type {@link T}.
+ * @throws Exception
+ */
+ protected abstract void run(RpcResponse response, T callback) throws Exception;
+
+ /**
+ * Calls from exceptionCaught
+ * @param requestId sequence id of request.
+ * @param callback callback of type {@link T}.
+ * @param message the error message to handle
+ */
+ protected abstract void handleException(int requestId, T callback, String message);
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
+ throws Exception {
+
+ Throwable rootCause = ExceptionUtils.getRootCause(cause);
+ LOG.error(getKey().addr + "," + getKey().protocolClass + "," + ExceptionUtils.getMessage(rootCause), rootCause);
+
+ if (cause instanceof RecoverableException) {
+ sendException((RecoverableException) cause);
+ } else {
+ /* unrecoverable fatal error*/
+ sendExceptions(ExceptionUtils.getMessage(rootCause));
+ if (ctx.channel().isOpen()) {
+ ctx.close();
+ }
+ }
+ }
+
+ /**
+ * Send an error to all callback
+ */
+ private void sendExceptions(String message) {
+ for (int requestId : requests.keySet()) {
+ handleException(requestId, requests.remove(requestId), message);
+ }
+ }
+
+ /**
+ * Send an error to callback
+ */
+ private void sendException(RecoverableException e) {
+ T callback = requests.remove(e.getSeqId());
+
+ if (callback != null) {
+ handleException(e.getSeqId(), callback, ExceptionUtils.getRootCauseMessage(e));
+ }
+ }
+
+ /**
+ * Trigger timeout event
+ */
+ @Override
+ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+
+ if (!enableMonitor && evt instanceof IdleStateEvent) {
+ IdleStateEvent e = (IdleStateEvent) evt;
+ /* If all requests is done and event is triggered, idle channel close. */
+ if (e.state() == IdleState.READER_IDLE && requests.isEmpty()) {
+ ctx.close();
+ LOG.info("Idle connection closed successfully :" + ctx.channel());
+ }
+ } else if (evt instanceof MonitorStateEvent) {
+ MonitorStateEvent e = (MonitorStateEvent) evt;
+ if (e.state() == MonitorStateEvent.MonitorState.PING_EXPIRED) {
+ exceptionCaught(ctx, new ServiceException("Server has not respond: " + ctx.channel()));
+ }
+ }
+
+ super.userEventTriggered(ctx, evt);
}
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ProtoChannelInitializer.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ProtoChannelInitializer.java b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ProtoChannelInitializer.java
deleted file mode 100644
index 74eb650..0000000
--- a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ProtoChannelInitializer.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.tajo.rpc;
-
-import com.google.protobuf.MessageLite;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelHandler;
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.ChannelPipeline;
-import io.netty.handler.codec.protobuf.ProtobufDecoder;
-import io.netty.handler.codec.protobuf.ProtobufEncoder;
-import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
-import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
-import io.netty.handler.timeout.IdleStateHandler;
-
-class ProtoChannelInitializer extends ChannelInitializer<Channel> {
- private final MessageLite defaultInstance;
- private final ChannelHandler handler;
- private final int idleTimeSeconds;
-
- public ProtoChannelInitializer(ChannelHandler handler, MessageLite defaultInstance) {
- this(handler, defaultInstance, 0);
- }
-
- public ProtoChannelInitializer(ChannelHandler handler, MessageLite defaultInstance, int idleTimeSeconds) {
- this.handler = handler;
- this.defaultInstance = defaultInstance;
- this.idleTimeSeconds = idleTimeSeconds;
- }
-
- @Override
- protected void initChannel(Channel channel) throws Exception {
- ChannelPipeline pipeline = channel.pipeline();
- pipeline.addLast("frameDecoder", new ProtobufVarint32FrameDecoder());
- pipeline.addLast("protobufDecoder", new ProtobufDecoder(defaultInstance));
- pipeline.addLast("frameEncoder", new ProtobufVarint32LengthFieldPrepender());
- pipeline.addLast("protobufEncoder", new ProtobufEncoder());
- pipeline.addLast("idleStateHandler", new IdleStateHandler(0, 0, idleTimeSeconds)); //zero is disabling
- pipeline.addLast("handler", handler);
- }
-}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ProtoClientChannelInitializer.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ProtoClientChannelInitializer.java b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ProtoClientChannelInitializer.java
new file mode 100644
index 0000000..8787dee
--- /dev/null
+++ b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ProtoClientChannelInitializer.java
@@ -0,0 +1,63 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.rpc;
+
+import com.google.protobuf.MessageLite;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.handler.codec.protobuf.ProtobufDecoder;
+import io.netty.handler.codec.protobuf.ProtobufEncoder;
+import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
+import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
+import io.netty.handler.timeout.IdleStateHandler;
+
+import java.util.concurrent.TimeUnit;
+
+class ProtoClientChannelInitializer extends ChannelInitializer<Channel> {
+ private final MessageLite defaultInstance;
+ private final ChannelHandler handler;
+ private final long timeoutTimeNanos;
+ private final boolean enablePing;
+
+ public ProtoClientChannelInitializer(ChannelHandler handler, MessageLite defaultInstance,
+ long timeoutTimeNanos,
+ boolean enablePing) {
+ this.handler = handler;
+ this.defaultInstance = defaultInstance;
+ this.timeoutTimeNanos = timeoutTimeNanos;
+ this.enablePing = enablePing;
+ }
+
+ @Override
+ protected void initChannel(Channel channel) throws Exception {
+ ChannelPipeline pipeline = channel.pipeline();
+ pipeline.addLast("idleStateHandler",
+ new IdleStateHandler(timeoutTimeNanos, timeoutTimeNanos / 2, 0, TimeUnit.NANOSECONDS));
+
+ if (enablePing) pipeline.addLast("MonitorClientHandler", new MonitorClientHandler());
+
+ pipeline.addLast("frameDecoder", new ProtobufVarint32FrameDecoder());
+ pipeline.addLast("protobufDecoder", new ProtobufDecoder(defaultInstance));
+ pipeline.addLast("frameEncoder", new ProtobufVarint32LengthFieldPrepender());
+ pipeline.addLast("protobufEncoder", new ProtobufEncoder());
+ pipeline.addLast("handler", handler);
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ProtoDeclaration.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ProtoDeclaration.java b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ProtoDeclaration.java
new file mode 100644
index 0000000..91d331e
--- /dev/null
+++ b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ProtoDeclaration.java
@@ -0,0 +1,24 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.rpc;
+
+interface ProtoDeclaration {
+
+ <I> I getStub();
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ProtoServerChannelInitializer.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ProtoServerChannelInitializer.java b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ProtoServerChannelInitializer.java
new file mode 100644
index 0000000..3e2fe0d
--- /dev/null
+++ b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ProtoServerChannelInitializer.java
@@ -0,0 +1,50 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.rpc;
+
+import com.google.protobuf.MessageLite;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.handler.codec.protobuf.ProtobufDecoder;
+import io.netty.handler.codec.protobuf.ProtobufEncoder;
+import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
+import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
+
+class ProtoServerChannelInitializer extends ChannelInitializer<Channel> {
+ private final MessageLite defaultInstance;
+ private final ChannelHandler handler;
+
+ public ProtoServerChannelInitializer(ChannelHandler handler, MessageLite defaultInstance) {
+ this.handler = handler;
+ this.defaultInstance = defaultInstance;
+ }
+
+ @Override
+ protected void initChannel(Channel channel) throws Exception {
+ ChannelPipeline pipeline = channel.pipeline();
+ pipeline.addLast("MonitorServerHandler", new MonitorServerHandler());
+ pipeline.addLast("frameDecoder", new ProtobufVarint32FrameDecoder());
+ pipeline.addLast("protobufDecoder", new ProtobufDecoder(defaultInstance));
+ pipeline.addLast("frameEncoder", new ProtobufVarint32LengthFieldPrepender());
+ pipeline.addLast("protobufEncoder", new ProtobufEncoder());
+ pipeline.addLast("handler", handler);
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/RecoverableException.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/RecoverableException.java b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/RecoverableException.java
new file mode 100644
index 0000000..1d22663
--- /dev/null
+++ b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/RecoverableException.java
@@ -0,0 +1,42 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.rpc;
+
+/* RecoverableException can be handle a failure request */
+public class RecoverableException extends RemoteException {
+ private int seqId;
+
+ public RecoverableException(int seqId, String message) {
+ super(message);
+ this.seqId = seqId;
+ }
+
+ public RecoverableException(int seqId, Throwable t) {
+ this(seqId, t.getMessage(), t);
+ }
+
+ public RecoverableException(int seqId, String message, Throwable t) {
+ super(message, t);
+ this.seqId = seqId;
+ }
+
+ public int getSeqId() {
+ return seqId;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/RpcClientManager.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/RpcClientManager.java b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/RpcClientManager.java
index f05fb97..f8def7f 100644
--- a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/RpcClientManager.java
+++ b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/RpcClientManager.java
@@ -18,26 +18,28 @@
package org.apache.tajo.rpc;
-import io.netty.channel.ConnectTimeoutException;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.internal.logging.CommonsLoggerFactory;
import io.netty.util.internal.logging.InternalLoggerFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import javax.annotation.concurrent.ThreadSafe;
+import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
@ThreadSafe
public class RpcClientManager {
private static final Log LOG = LogFactory.getLog(RpcClientManager.class);
- public static final int RPC_RETRIES = 3;
-
- /* If all requests is done and client is idle state, client will be removed. */
- public static final int RPC_IDLE_TIMEOUT = 43200; // 12 hour
+ private volatile int timeoutSeconds = RpcConstants.DEFAULT_RPC_TIMEOUT_SECONDS;
+ private volatile int retries = RpcConstants.DEFAULT_RPC_RETRIES;
/* entries will be removed by ConnectionCloseFutureListener */
private static final Map<RpcConnectionKey, NettyClientBase>
@@ -57,45 +59,86 @@ public class RpcClientManager {
return instance;
}
- private NettyClientBase makeClient(RpcConnectionKey rpcConnectionKey)
- throws NoSuchMethodException, ClassNotFoundException, ConnectTimeoutException {
+ private <T extends NettyClientBase> T makeClient(RpcConnectionKey rpcConnectionKey,
+ int retries,
+ long timeout,
+ TimeUnit timeUnit,
+ boolean enablePing)
+ throws NoSuchMethodException, ClassNotFoundException, ConnectException {
NettyClientBase client;
if (rpcConnectionKey.asyncMode) {
- client = new AsyncRpcClient(rpcConnectionKey, RPC_RETRIES, RPC_IDLE_TIMEOUT);
+ client = new AsyncRpcClient(rpcConnectionKey, retries, timeout, timeUnit, enablePing);
} else {
- client = new BlockingRpcClient(rpcConnectionKey, RPC_RETRIES, RPC_IDLE_TIMEOUT);
+ client = new BlockingRpcClient(rpcConnectionKey, retries, timeout, timeUnit, enablePing);
}
- return client;
+ return (T) client;
}
/**
* Connect a {@link NettyClientBase} to the remote {@link NettyServerBase}, and returns rpc client by protocol.
* This client will be shared per protocol and address. Client is removed in shared map when a client is closed
- * @param addr
- * @param protocolClass
- * @param asyncMode
- * @return
- * @throws NoSuchMethodException
- * @throws ClassNotFoundException
- * @throws ConnectTimeoutException
*/
- public NettyClientBase getClient(InetSocketAddress addr,
- Class<?> protocolClass, boolean asyncMode)
- throws NoSuchMethodException, ClassNotFoundException, ConnectTimeoutException {
+ public <T extends NettyClientBase> T getClient(InetSocketAddress addr,
+ Class<?> protocolClass, boolean asyncMode)
+ throws NoSuchMethodException, ClassNotFoundException, ConnectException {
RpcConnectionKey key = new RpcConnectionKey(addr, protocolClass, asyncMode);
NettyClientBase client;
synchronized (clients) {
client = clients.get(key);
if (client == null) {
- clients.put(key, client = makeClient(key));
+ clients.put(key, client = makeClient(key, retries, getTimeoutSeconds(), TimeUnit.SECONDS, true));
}
}
if (!client.isConnected()) {
- client.connect();
- client.getChannel().closeFuture().addListener(new ConnectionCloseFutureListener(key));
+
+ final NettyClientBase target = client;
+ client.subscribeEvent(target.getKey(), new ChannelEventListener() {
+ @Override
+ public void channelRegistered(ChannelHandlerContext ctx) {
+ /* Register client to managed map */
+ clients.put(target.getKey(), target);
+ target.getChannel().closeFuture().addListener(new ClientCloseFutureListener(target.getKey()));
+ }
+
+ @Override
+ public void channelUnregistered(ChannelHandlerContext ctx) {
+ // nothing to do
+ }
+ });
}
+
+ client.connect();
+ assert client.isConnected();
+ return (T) client;
+ }
+
+ /**
+ * Connect a {@link NettyClientBase} to the remote {@link NettyServerBase}, and returns rpc client by protocol.
+ * This client does not managed. It should close.
+ */
+ public synchronized <T extends NettyClientBase> T newClient(InetSocketAddress addr,
+ Class<?> protocolClass,
+ boolean asyncMode,
+ int retries,
+ long timeout,
+ TimeUnit timeUnit,
+ boolean enablePing)
+ throws NoSuchMethodException, ClassNotFoundException, ConnectException {
+
+ return newClient(new RpcConnectionKey(addr, protocolClass, asyncMode), retries, timeout, timeUnit, enablePing);
+ }
+
+ public synchronized <T extends NettyClientBase> T newClient(RpcConnectionKey key,
+ int retries,
+ long timeout,
+ TimeUnit timeUnit,
+ boolean enablePing)
+ throws NoSuchMethodException, ClassNotFoundException, ConnectException {
+
+ T client = makeClient(key, retries, timeout, timeUnit, enablePing);
+ client.connect();
assert client.isConnected();
return client;
}
@@ -125,11 +168,6 @@ public class RpcClientManager {
RpcChannelFactory.shutdownGracefully();
}
- protected static NettyClientBase remove(RpcConnectionKey key) {
- LOG.debug("Removing shared rpc client :" + key);
- return clients.remove(key);
- }
-
protected static boolean contains(RpcConnectionKey key) {
return clients.containsKey(key);
}
@@ -148,6 +186,22 @@ public class RpcClientManager {
}
}
+ public int getTimeoutSeconds() {
+ return timeoutSeconds;
+ }
+
+ public void setTimeoutSeconds(int timeoutSeconds) {
+ this.timeoutSeconds = timeoutSeconds;
+ }
+
+ public int getRetries() {
+ return retries;
+ }
+
+ public void setRetries(int retries) {
+ this.retries = retries;
+ }
+
static class RpcConnectionKey {
final InetSocketAddress addr;
final Class<?> protocolClass;
@@ -182,4 +236,17 @@ public class RpcClientManager {
return description.hashCode();
}
}
+
+ static class ClientCloseFutureListener implements GenericFutureListener {
+ private RpcClientManager.RpcConnectionKey key;
+
+ public ClientCloseFutureListener(RpcClientManager.RpcConnectionKey key) {
+ this.key = key;
+ }
+
+ @Override
+ public void operationComplete(Future future) throws Exception {
+ clients.remove(key);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/main/proto/TestProtocol.proto
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/main/proto/TestProtocol.proto b/tajo-rpc/tajo-rpc-protobuf/src/main/proto/TestProtocol.proto
index 824ea38..35cc945 100644
--- a/tajo-rpc/tajo-rpc-protobuf/src/main/proto/TestProtocol.proto
+++ b/tajo-rpc/tajo-rpc-protobuf/src/main/proto/TestProtocol.proto
@@ -28,6 +28,7 @@ service DummyProtocolService {
rpc echo (EchoMessage) returns (EchoMessage);
rpc getError (EchoMessage) returns (EchoMessage);
rpc getNull (EchoMessage) returns (EchoMessage);
- rpc deley (EchoMessage) returns (EchoMessage);
+ rpc delay (EchoMessage) returns (EchoMessage);
+ rpc busy (EchoMessage) returns (EchoMessage);
rpc throwException (EchoMessage) returns (EchoMessage);
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/TestAsyncRpc.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/TestAsyncRpc.java b/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/TestAsyncRpc.java
index 68f170c..f8c6d32 100644
--- a/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/TestAsyncRpc.java
+++ b/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/TestAsyncRpc.java
@@ -39,9 +39,11 @@ import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
+import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -59,17 +61,20 @@ public class TestAsyncRpc {
Interface stub;
DummyProtocolAsyncImpl service;
int retries;
-
+ RpcClientManager.RpcConnectionKey rpcConnectionKey;
+ RpcClientManager manager = RpcClientManager.getInstance();
+
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
@interface SetupRpcConnection {
boolean setupRpcServer() default true;
+
boolean setupRpcClient() default true;
}
-
+
@Rule
public ExternalResource resource = new ExternalResource() {
-
+
private Description description;
@Override
@@ -85,7 +90,7 @@ public class TestAsyncRpc {
if (setupRpcConnection == null || setupRpcConnection.setupRpcServer()) {
setUpRpcServer();
}
-
+
if (setupRpcConnection == null || setupRpcConnection.setupRpcClient()) {
setUpRpcClient();
}
@@ -102,7 +107,7 @@ public class TestAsyncRpc {
fail(e.getMessage());
}
}
-
+
if (setupRpcConnection == null || setupRpcConnection.setupRpcServer()) {
try {
tearDownRpcServer();
@@ -111,25 +116,24 @@ public class TestAsyncRpc {
}
}
}
-
+
};
-
+
public void setUpRpcServer() throws Exception {
service = new DummyProtocolAsyncImpl();
server = new AsyncRpcServer(DummyProtocol.class,
- service, new InetSocketAddress("127.0.0.1", 0), 2);
+ service, new InetSocketAddress("127.0.0.1", 0), 3);
server.start();
}
-
+
public void setUpRpcClient() throws Exception {
retries = 1;
- RpcClientManager.RpcConnectionKey rpcConnectionKey =
- new RpcClientManager.RpcConnectionKey(
- RpcUtils.getConnectAddress(server.getListenAddress()),
- DummyProtocol.class, true);
- client = new AsyncRpcClient(rpcConnectionKey, retries);
- client.connect();
+ rpcConnectionKey = new RpcClientManager.RpcConnectionKey(
+ RpcUtils.getConnectAddress(server.getListenAddress()),
+ DummyProtocol.class, true);
+ client = manager.newClient(rpcConnectionKey, retries, 10, TimeUnit.SECONDS, true);
+ assertTrue(client.isConnected());
stub = client.getStub();
}
@@ -137,16 +141,16 @@ public class TestAsyncRpc {
public static void tearDownClass() throws Exception {
RpcChannelFactory.shutdownGracefully();
}
-
+
public void tearDownRpcServer() throws Exception {
- if(server != null) {
+ if (server != null) {
server.shutdown();
server = null;
}
}
-
+
public void tearDownRpcClient() throws Exception {
- if(client != null) {
+ if (client != null) {
client.close();
client = null;
}
@@ -172,35 +176,37 @@ public class TestAsyncRpc {
});
+ final CountDownLatch barrier = new CountDownLatch(1);
EchoMessage echoMessage = EchoMessage.newBuilder()
.setMessage(MESSAGE).build();
RpcCallback<EchoMessage> callback = new RpcCallback<EchoMessage>() {
@Override
public void run(EchoMessage parameter) {
+ assertNotNull(parameter);
echo = parameter.getMessage();
assertEquals(MESSAGE, echo);
calledMarker = true;
+ barrier.countDown();
}
};
stub.echo(null, echoMessage, callback);
- Thread.sleep(1000);
+
+ assertTrue(barrier.await(1000, TimeUnit.MILLISECONDS));
assertTrue(calledMarker);
}
- private CountDownLatch testNullLatch;
-
@Test
public void testGetNull() throws Exception {
- testNullLatch = new CountDownLatch(1);
+ final CountDownLatch barrier = new CountDownLatch(1);
stub.getNull(null, null, new RpcCallback<EchoMessage>() {
@Override
public void run(EchoMessage parameter) {
assertNull(parameter);
LOG.info("testGetNull retrieved");
- testNullLatch.countDown();
+ barrier.countDown();
}
});
- assertTrue(testNullLatch.await(1000, TimeUnit.MILLISECONDS));
+ assertTrue(barrier.await(1000, TimeUnit.MILLISECONDS));
assertTrue(service.getNullCalled);
}
@@ -209,70 +215,112 @@ public class TestAsyncRpc {
EchoMessage echoMessage = EchoMessage.newBuilder()
.setMessage(MESSAGE).build();
CallFuture<EchoMessage> future = new CallFuture<EchoMessage>();
- stub.deley(null, echoMessage, future);
+ stub.delay(future.getController(), echoMessage, future);
assertFalse(future.isDone());
- assertEquals(future.get(), echoMessage);
+ assertEquals(echoMessage, future.get());
assertTrue(future.isDone());
}
@Test
public void testCallFutureTimeout() throws Exception {
boolean timeout = false;
+ CallFuture<EchoMessage> future = new CallFuture<EchoMessage>();
+ EchoMessage echoMessage = EchoMessage.newBuilder().setMessage(MESSAGE).build();
try {
- EchoMessage echoMessage = EchoMessage.newBuilder()
- .setMessage(MESSAGE).build();
- CallFuture<EchoMessage> future = new CallFuture<EchoMessage>();
- stub.deley(null, echoMessage, future);
-
+ stub.delay(future.getController(), echoMessage, future);
assertFalse(future.isDone());
- future.get(1, TimeUnit.SECONDS);
+ future.get(100, TimeUnit.MILLISECONDS);
} catch (TimeoutException te) {
timeout = true;
}
+ assertFalse(future.getController().failed());
assertTrue(timeout);
}
@Test
- public void testCallFutureDisconnected() throws Exception {
+ public void testThrowException() throws Exception {
EchoMessage echoMessage = EchoMessage.newBuilder()
.setMessage(MESSAGE).build();
CallFuture<EchoMessage> future = new CallFuture<EchoMessage>();
+ stub.throwException(future.getController(), echoMessage, future);
- tearDownRpcServer();
-
- stub.echo(future.getController(), echoMessage, future);
- EchoMessage response = future.get();
+ assertFalse(future.isDone());
+ EchoMessage result = null;
+ try {
+ result = future.get();
+ } catch (ExecutionException e) {
+ }
- assertNull(response);
+ assertEquals(null, result);
+ assertTrue(future.isDone());
assertTrue(future.getController().failed());
- assertTrue(future.getController().errorText() != null);
+ assertNotNull(future.getController().errorText());
}
@Test
- public void testStubDisconnected() throws Exception {
+ public void testThrowException2() throws Exception {
+ EchoMessage echoMessage = EchoMessage.newBuilder()
+ .setMessage(MESSAGE).build();
+ CallFuture<EchoMessage> future = new CallFuture<EchoMessage>();
+ stub.throwException(null, echoMessage, future);
+
+ assertFalse(future.isDone());
+ assertNull(future.get());
+
+ assertTrue(future.isDone());
+ assertFalse(future.getController().failed());
+ assertNull(future.getController().errorText());
+ }
+ @Test(timeout = 60000)
+ public void testServerShutdown1() throws Exception {
EchoMessage echoMessage = EchoMessage.newBuilder()
.setMessage(MESSAGE).build();
CallFuture<EchoMessage> future = new CallFuture<EchoMessage>();
- if (server != null) {
- server.shutdown(true);
- server = null;
+ tearDownRpcServer();
+
+ stub.echo(future.getController(), echoMessage, future);
+
+ EchoMessage result = null;
+ try {
+ result = future.get();
+ } catch (ExecutionException e) {
}
- stub = client.getStub();
+ assertEquals(null, result);
+ assertTrue(future.isDone());
+ assertTrue(future.getController().failed());
+ assertNotNull(future.getController().errorText(), future.getController().errorText());
+ }
+
+ @Test(timeout = 60000)
+ public void testServerShutdown2() throws Exception {
+ EchoMessage echoMessage = EchoMessage.newBuilder()
+ .setMessage(MESSAGE).build();
+ CallFuture<EchoMessage> future = new CallFuture<EchoMessage>();
+
+ tearDownRpcServer();
+
+ Interface stub = client.getStub();
stub.echo(future.getController(), echoMessage, future);
- EchoMessage response = future.get();
- assertNull(response);
+ EchoMessage result = null;
+ try {
+ result = future.get();
+ } catch (ExecutionException e) {
+ }
+
+ assertEquals(null, result);
+ assertTrue(future.isDone());
assertTrue(future.getController().failed());
- assertTrue(future.getController().errorText() != null);
+ assertNotNull(future.getController().errorText(), future.getController().errorText());
}
@Test
- @SetupRpcConnection(setupRpcServer=false,setupRpcClient=false)
- public void testConnectionRetry() throws Exception {
+ @SetupRpcConnection(setupRpcServer = false, setupRpcClient = false)
+ public void testClientRetryOnStartup() throws Exception {
retries = 10;
ServerSocket serverSocket = new ServerSocket(0);
final InetSocketAddress address = new InetSocketAddress("127.0.0.1", serverSocket.getLocalPort());
@@ -300,94 +348,228 @@ public class TestAsyncRpc {
serverThread.start();
RpcClientManager.RpcConnectionKey rpcConnectionKey =
- new RpcClientManager.RpcConnectionKey(address, DummyProtocol.class, true);
- client = new AsyncRpcClient(rpcConnectionKey, retries);
- client.connect();
+ new RpcClientManager.RpcConnectionKey(address, DummyProtocol.class, true);
+ AsyncRpcClient client = manager.newClient(rpcConnectionKey,
+ retries, 0, TimeUnit.MILLISECONDS, false);
assertTrue(client.isConnected());
- stub = client.getStub();
+
+ Interface stub = client.getStub();
stub.echo(future.getController(), echoMessage, future);
assertFalse(future.isDone());
assertEquals(echoMessage, future.get());
assertTrue(future.isDone());
+ client.close();
+ server.shutdown();
}
- @Test
- public void testConnectionFailure() throws Exception {
+
+ @Test(timeout = 60000)
+ @SetupRpcConnection(setupRpcServer = false, setupRpcClient = false)
+ public void testClientRetryFailureOnStartup() throws Exception {
+ retries = 2;
+ ServerSocket serverSocket = new ServerSocket(0);
+ final InetSocketAddress address = new InetSocketAddress("127.0.0.1", serverSocket.getLocalPort());
+ serverSocket.close();
+ service = new DummyProtocolAsyncImpl();
+
+ EchoMessage echoMessage = EchoMessage.newBuilder()
+ .setMessage(MESSAGE).build();
+ CallFuture<EchoMessage> future = new CallFuture<EchoMessage>();
+
+ RpcClientManager.RpcConnectionKey rpcConnectionKey =
+ new RpcClientManager.RpcConnectionKey(address, DummyProtocol.class, true);
+ AsyncRpcClient client = new AsyncRpcClient(rpcConnectionKey, retries);
+ try {
+ client.connect();
+ fail();
+ } catch (ConnectTimeoutException e) {
+ assertFalse(e.getMessage(), client.isConnected());
+ }
+
+ stub = client.getStub();
+ stub.echo(future.getController(), echoMessage, future);
+
+ EchoMessage result = null;
+ try {
+ result = future.get();
+ } catch (ExecutionException e) {
+ }
+
+ assertEquals(null, result);
+ assertTrue(future.isDone());
+ assertTrue(future.getController().failed());
+ assertNotNull(future.getController().errorText(), future.getController().errorText());
+ }
+
+ @Test(timeout = 60000)
+ @SetupRpcConnection(setupRpcServer = false, setupRpcClient = false)
+ public void testUnresolvedAddress() throws Exception {
InetSocketAddress address = new InetSocketAddress("test", 0);
boolean expected = false;
+ AsyncRpcClient client = null;
try {
RpcClientManager.RpcConnectionKey rpcConnectionKey =
new RpcClientManager.RpcConnectionKey(address, DummyProtocol.class, true);
- NettyClientBase client = new AsyncRpcClient(rpcConnectionKey, retries);
+ client = new AsyncRpcClient(rpcConnectionKey, retries);
client.connect();
fail();
- } catch (ConnectTimeoutException e) {
+ } catch (ConnectException e) {
expected = true;
} catch (Throwable throwable) {
- fail();
+ fail(throwable.getMessage());
+ } finally {
+ client.close();
}
assertTrue(expected);
}
- @Test
- @SetupRpcConnection(setupRpcClient=false)
- public void testUnresolvedAddress() throws Exception {
+ @Test(timeout = 60000)
+ @SetupRpcConnection(setupRpcClient = false)
+ public void testUnresolvedAddress2() throws Exception {
String hostAndPort = RpcUtils.normalizeInetSocketAddress(server.getListenAddress());
RpcClientManager.RpcConnectionKey rpcConnectionKey =
- new RpcClientManager.RpcConnectionKey(
- RpcUtils.createUnresolved(hostAndPort), DummyProtocol.class, true);
- client = new AsyncRpcClient(rpcConnectionKey, retries);
+ new RpcClientManager.RpcConnectionKey(
+ RpcUtils.createUnresolved(hostAndPort), DummyProtocol.class, true);
+ AsyncRpcClient client = new AsyncRpcClient(rpcConnectionKey, retries);
client.connect();
- assertTrue(client.isConnected());
- Interface stub = client.getStub();
+ try {
+ assertTrue(client.isConnected());
+ Interface stub = client.getStub();
+ EchoMessage echoMessage = EchoMessage.newBuilder()
+ .setMessage(MESSAGE).build();
+ CallFuture<EchoMessage> future = new CallFuture<EchoMessage>();
+ stub.echo(future.getController(), echoMessage, future);
+
+ assertFalse(future.isDone());
+ assertEquals(echoMessage, future.get());
+ assertTrue(future.isDone());
+ } finally {
+ client.close();
+ }
+ }
+
+ @Test(timeout = 60000)
+ @SetupRpcConnection(setupRpcClient = false)
+ public void testStubRecovery() throws Exception {
+ RpcClientManager.RpcConnectionKey rpcConnectionKey =
+ new RpcClientManager.RpcConnectionKey(server.getListenAddress(), DummyProtocol.class, true);
+ AsyncRpcClient client = manager.newClient(rpcConnectionKey, 2, 0, TimeUnit.MILLISECONDS, false);
+
EchoMessage echoMessage = EchoMessage.newBuilder()
.setMessage(MESSAGE).build();
- CallFuture<EchoMessage> future = new CallFuture<EchoMessage>();
- stub.deley(null, echoMessage, future);
+ int repeat = 5;
- assertFalse(future.isDone());
- assertEquals(future.get(), echoMessage);
- assertTrue(future.isDone());
+ assertTrue(client.isConnected());
+ Interface stub = client.getStub();
+
+ client.close(); // close connection
+ assertFalse(client.isConnected());
+
+ for (int i = 0; i < repeat; i++) {
+ try {
+ CallFuture<EchoMessage> future = new CallFuture<EchoMessage>();
+ stub.echo(future.getController(), echoMessage, future);
+ assertEquals(echoMessage, future.get());
+ assertTrue(future.isDone());
+ assertTrue(client.isConnected());
+ } finally {
+ client.close(); // close connection
+ assertFalse(client.isConnected());
+ }
+ }
}
- @Test
+ @Test(timeout = 60000)
+ @SetupRpcConnection(setupRpcClient = false)
public void testIdleTimeout() throws Exception {
RpcClientManager.RpcConnectionKey rpcConnectionKey =
new RpcClientManager.RpcConnectionKey(server.getListenAddress(), DummyProtocol.class, true);
- AsyncRpcClient client = new AsyncRpcClient(rpcConnectionKey, retries, 1); //1 sec idle timeout
- client.connect();
+ //500 millis idle timeout
+ AsyncRpcClient client = manager.newClient(rpcConnectionKey, retries, 500, TimeUnit.MILLISECONDS, false);
assertTrue(client.isConnected());
- Thread.sleep(2000);
+ Thread.sleep(600); //timeout
assertFalse(client.isConnected());
client.connect(); // try to reconnect
assertTrue(client.isConnected());
- client.close();
+
+ Thread.sleep(600); //timeout
assertFalse(client.isConnected());
+ client.close();
}
- @Test
- public void testIdleTimeoutWithActiveRequest() throws Exception {
+ @Test(timeout = 60000)
+ @SetupRpcConnection(setupRpcClient = false)
+ public void testPingOnIdle() throws Exception {
RpcClientManager.RpcConnectionKey rpcConnectionKey =
new RpcClientManager.RpcConnectionKey(server.getListenAddress(), DummyProtocol.class, true);
- AsyncRpcClient client = new AsyncRpcClient(rpcConnectionKey, retries, 1); //1 sec idle timeout
- client.connect();
+ //500 millis request timeout
+ AsyncRpcClient client = manager.newClient(rpcConnectionKey, retries, 500, TimeUnit.MILLISECONDS, true);
+ assertTrue(client.isConnected());
+
+ Thread.sleep(600);
+ assertTrue(client.isConnected());
+
+ Thread.sleep(600);
+ assertTrue(client.isConnected());
+ client.close();
+ }
+
+ @Test(timeout = 60000)
+ @SetupRpcConnection(setupRpcClient = false)
+ public void testIdleTimeoutWithActiveRequest() throws Exception {
+ RpcClientManager.RpcConnectionKey rpcConnectionKey =
+ new RpcClientManager.RpcConnectionKey(server.getListenAddress(), DummyProtocol.class, true);
+ //500 millis idle timeout
+ AsyncRpcClient client = manager.newClient(rpcConnectionKey, retries, 500, TimeUnit.MILLISECONDS, false);
assertTrue(client.isConnected());
+
Interface stub = client.getStub();
EchoMessage echoMessage = EchoMessage.newBuilder()
.setMessage(MESSAGE).build();
CallFuture<EchoMessage> future = new CallFuture<EchoMessage>();
- stub.deley(null, echoMessage, future); //3 sec delay
+ stub.delay(future.getController(), echoMessage, future); //3 sec delay
+ assertTrue(client.isConnected());
assertFalse(future.isDone());
- assertEquals(future.get(), echoMessage);
+ assertEquals(echoMessage, future.get());
assertTrue(future.isDone());
- Thread.sleep(2000);
+ assertTrue(client.getActiveRequests() == 0);
+ Thread.sleep(600); //timeout
assertFalse(client.isConnected());
}
+
+ @Test(timeout = 60000)
+ @SetupRpcConnection(setupRpcClient = false)
+ public void testRequestTimeoutOnBusy() throws Exception {
+ RpcClientManager.RpcConnectionKey rpcConnectionKey =
+ new RpcClientManager.RpcConnectionKey(server.getListenAddress(), DummyProtocol.class, true);
+
+ //500 millis request timeout
+ AsyncRpcClient client = manager.newClient(rpcConnectionKey, retries, 500, TimeUnit.MILLISECONDS, true);
+ assertTrue(client.isConnected());
+
+ Interface stub = client.getStub();
+ EchoMessage echoMessage = EchoMessage.newBuilder()
+ .setMessage(MESSAGE).build();
+ CallFuture<EchoMessage> future = new CallFuture<EchoMessage>();
+ stub.busy(future.getController(), echoMessage, future); //30 sec delay
+ assertFalse(future.isDone());
+
+ EchoMessage result = null;
+ try {
+ result = future.get();
+ } catch (ExecutionException e) {
+ }
+
+ assertEquals(null, result);
+ assertTrue(future.getController().errorText(), future.getController().failed());
+ assertTrue(client.getActiveRequests() == 0);
+ client.close();
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/TestBlockingRpc.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/TestBlockingRpc.java b/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/TestBlockingRpc.java
index 568eb63..6f7fdd1 100644
--- a/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/TestBlockingRpc.java
+++ b/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/TestBlockingRpc.java
@@ -36,6 +36,7 @@ import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
+import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.util.concurrent.CountDownLatch;
@@ -50,18 +51,20 @@ public class TestBlockingRpc {
private BlockingRpcClient client;
private BlockingInterface stub;
private DummyProtocolBlockingImpl service;
+ RpcClientManager manager = RpcClientManager.getInstance();
private int retries;
-
+
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
@interface SetupRpcConnection {
boolean setupRpcServer() default true;
+
boolean setupRpcClient() default true;
}
-
+
@Rule
public ExternalResource resource = new ExternalResource() {
-
+
private Description description;
@Override
@@ -73,11 +76,11 @@ public class TestBlockingRpc {
@Override
protected void before() throws Throwable {
SetupRpcConnection setupRpcConnection = description.getAnnotation(SetupRpcConnection.class);
-
+
if (setupRpcConnection == null || setupRpcConnection.setupRpcServer()) {
setUpRpcServer();
}
-
+
if (setupRpcConnection == null || setupRpcConnection.setupRpcClient()) {
setUpRpcClient();
}
@@ -86,7 +89,7 @@ public class TestBlockingRpc {
@Override
protected void after() {
SetupRpcConnection setupRpcConnection = description.getAnnotation(SetupRpcConnection.class);
-
+
if (setupRpcConnection == null || setupRpcConnection.setupRpcClient()) {
try {
tearDownRpcClient();
@@ -94,7 +97,7 @@ public class TestBlockingRpc {
fail(e.getMessage());
}
}
-
+
if (setupRpcConnection == null || setupRpcConnection.setupRpcServer()) {
try {
tearDownRpcServer();
@@ -103,25 +106,24 @@ public class TestBlockingRpc {
}
}
}
-
+
};
-
+
public void setUpRpcServer() throws Exception {
service = new DummyProtocolBlockingImpl();
server = new BlockingRpcServer(DummyProtocol.class, service,
new InetSocketAddress("127.0.0.1", 0), 2);
server.start();
}
-
+
public void setUpRpcClient() throws Exception {
retries = 1;
RpcClientManager.RpcConnectionKey rpcConnectionKey =
- new RpcClientManager.RpcConnectionKey(
- RpcUtils.getConnectAddress(server.getListenAddress()),
- DummyProtocol.class, false);
- client = new BlockingRpcClient(rpcConnectionKey, retries);
- client.connect();
+ new RpcClientManager.RpcConnectionKey(
+ RpcUtils.getConnectAddress(server.getListenAddress()),
+ DummyProtocol.class, false);
+ client = manager.newClient(rpcConnectionKey, retries, 10, TimeUnit.SECONDS, true);
assertTrue(client.isConnected());
stub = client.getStub();
}
@@ -130,16 +132,16 @@ public class TestBlockingRpc {
public static void tearDownClass() throws Exception {
RpcChannelFactory.shutdownGracefully();
}
-
+
public void tearDownRpcServer() throws Exception {
- if(server != null) {
+ if (server != null) {
server.shutdown();
server = null;
}
}
-
+
public void tearDownRpcClient() throws Exception {
- if(client != null) {
+ if (client != null) {
client.close();
client = null;
}
@@ -162,7 +164,14 @@ public class TestBlockingRpc {
}
@Test
- @SetupRpcConnection(setupRpcClient=false)
+ public void testGetNull() throws Exception {
+ assertNull(stub.getNull(null, null));
+ assertTrue(service.getNullCalled);
+ }
+
+ @Test
+ @SetupRpcConnection(setupRpcClient = false)
+ @Deprecated // serverCallable will be remove
public void testRpcWithServiceCallable() throws Exception {
RpcClientManager manager = RpcClientManager.getInstance();
final SumRequest request = SumRequest.newBuilder()
@@ -172,15 +181,15 @@ public class TestBlockingRpc {
.setX4(2.0f).build();
SumResponse response =
- new ServerCallable<SumResponse>(manager,
- server.getListenAddress(), DummyProtocol.class, false) {
- @Override
- public SumResponse call(NettyClientBase client) throws Exception {
- BlockingInterface stub2 = client.getStub();
- SumResponse response1 = stub2.sum(null, request);
- return response1;
- }
- }.withRetries();
+ new ServerCallable<SumResponse>(manager,
+ server.getListenAddress(), DummyProtocol.class, false) {
+ @Override
+ public SumResponse call(NettyClientBase client) throws Exception {
+ BlockingInterface stub2 = client.getStub();
+ SumResponse response1 = stub2.sum(null, request);
+ return response1;
+ }
+ }.withRetries();
assertEquals(8.15d, response.getResult(), 1e-15);
@@ -207,10 +216,8 @@ public class TestBlockingRpc {
try {
stub.throwException(null, message);
fail("RpcCall should throw exception");
- } catch (Throwable t) {
- assertTrue(t instanceof TajoServiceException);
- assertEquals("Exception Test", t.getMessage());
- TajoServiceException te = (TajoServiceException)t;
+ } catch (TajoServiceException te) {
+ assertEquals("Exception Test", te.getMessage());
assertEquals("org.apache.tajo.rpc.test.DummyProtocol", te.getProtocol());
assertEquals(server.getListenAddress().getAddress().getHostAddress() + ":" + server.getListenAddress().getPort(),
te.getRemoteAddress());
@@ -222,83 +229,57 @@ public class TestBlockingRpc {
EchoMessage message = EchoMessage.newBuilder()
.setMessage(MESSAGE).build();
+ DefaultRpcController controller = new DefaultRpcController();
try {
- stub.throwException(null, message);
+ stub.throwException(controller, message);
fail("RpcCall should throw exception");
- } catch (Throwable t) {
- assertTrue(t instanceof TajoServiceException);
+ } catch (TajoServiceException t) {
+ assertTrue(controller.failed());
+ assertNotNull(controller.errorText());
}
- EchoMessage message1 = stub.deley(null, message);
+ controller.reset();
+ EchoMessage message1 = stub.delay(controller, message);
assertEquals(message, message1);
+ assertFalse(controller.failed());
+ assertNull(controller.errorText());
}
- @Test
- @SetupRpcConnection(setupRpcServer=false,setupRpcClient=false)
- public void testConnectionRetry() throws Exception {
- retries = 10;
- ServerSocket serverSocket = new ServerSocket(0);
- final InetSocketAddress address = new InetSocketAddress("127.0.0.1", serverSocket.getLocalPort());
- serverSocket.close();
-
+ @Test(timeout = 60000)
+ public void testServerShutdown1() throws Exception {
EchoMessage message = EchoMessage.newBuilder()
.setMessage(MESSAGE).build();
- //lazy startup
- Thread serverThread = new Thread(new Runnable() {
- @Override
- public void run() {
- try {
- Thread.sleep(1000);
- server = new BlockingRpcServer(DummyProtocol.class, new DummyProtocolBlockingImpl(), address, 2);
- } catch (Exception e) {
- fail(e.getMessage());
- }
- server.start();
- }
- });
- serverThread.start();
-
- RpcClientManager.RpcConnectionKey rpcConnectionKey =
- new RpcClientManager.RpcConnectionKey(address, DummyProtocol.class, false);
- client = new BlockingRpcClient(rpcConnectionKey, retries);
- client.connect();
- assertTrue(client.isConnected());
- stub = client.getStub();
-
- EchoMessage response = stub.echo(null, message);
- assertEquals(MESSAGE, response.getMessage());
- }
-
- @Test
- public void testConnectionFailed() throws Exception {
- NettyClientBase client = null;
- boolean expected = false;
+ tearDownRpcServer();
+ boolean expect = false;
try {
- int port = server.getListenAddress().getPort() + 1;
- RpcClientManager.RpcConnectionKey rpcConnectionKey =
- new RpcClientManager.RpcConnectionKey(
- RpcUtils.getConnectAddress(new InetSocketAddress("127.0.0.1", port)),
- DummyProtocol.class, false);
- client = new BlockingRpcClient(rpcConnectionKey, retries);
- client.connect();
- fail();
- } catch (ConnectTimeoutException e) {
- expected = true;
- } catch (Throwable e) {
- fail(e.getMessage());
+ EchoMessage response = stub.echo(null, message);
+ fail(response.getMessage());
+ } catch (TajoServiceException e) {
+ expect = true;
}
- assertTrue(expected);
+ assertTrue(expect);
}
- @Test
- public void testGetNull() throws Exception {
- assertNull(stub.getNull(null, null));
- assertTrue(service.getNullCalled);
+ @Test(timeout = 60000)
+ public void testServerShutdown2() throws Exception {
+ EchoMessage message = EchoMessage.newBuilder()
+ .setMessage(MESSAGE).build();
+
+ tearDownRpcServer();
+ boolean expect = false;
+ try {
+ BlockingInterface stub = client.getStub();
+ EchoMessage response = stub.echo(null, message);
+ fail(response.getMessage());
+ } catch (TajoServiceException e) {
+ expect = true;
+ }
+ assertTrue(expect);
}
@Test
- public void testShutdown() throws Exception {
+ public void testServerShutdown3() throws Exception {
final StringBuilder error = new StringBuilder();
Thread callThread = new Thread() {
public void run() {
@@ -306,11 +287,11 @@ public class TestBlockingRpc {
EchoMessage message = EchoMessage.newBuilder()
.setMessage(MESSAGE)
.build();
- stub.deley(null, message);
+ stub.delay(null, message);
} catch (Exception e) {
error.append(e.getMessage());
}
- synchronized(error) {
+ synchronized (error) {
error.notifyAll();
}
}
@@ -340,67 +321,244 @@ public class TestBlockingRpc {
assertTrue(latch.getCount() == 0);
- synchronized(error) {
+ synchronized (error) {
error.wait(5 * 1000);
}
- if(!error.toString().isEmpty()) {
+ if (!error.toString().isEmpty()) {
fail(error.toString());
}
}
- @Test
- @SetupRpcConnection(setupRpcClient=false)
- public void testUnresolvedAddress() throws Exception {
- String hostAndPort = RpcUtils.normalizeInetSocketAddress(server.getListenAddress());
+ @Test(timeout = 60000)
+ @SetupRpcConnection(setupRpcServer = false, setupRpcClient = false)
+ public void testClientRetryOnStartup() throws Exception {
+ retries = 10;
+ ServerSocket serverSocket = new ServerSocket(0);
+ final InetSocketAddress address = new InetSocketAddress("127.0.0.1", serverSocket.getLocalPort());
+ serverSocket.close();
+
+ EchoMessage message = EchoMessage.newBuilder()
+ .setMessage(MESSAGE).build();
+
+ //lazy startup
+ Thread serverThread = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ Thread.sleep(1000);
+ server = new BlockingRpcServer(DummyProtocol.class, new DummyProtocolBlockingImpl(), address, 2);
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+ server.start();
+ }
+ });
+ serverThread.start();
+
RpcClientManager.RpcConnectionKey rpcConnectionKey =
- new RpcClientManager.RpcConnectionKey(
- RpcUtils.createUnresolved(hostAndPort), DummyProtocol.class, false);
- client = new BlockingRpcClient(rpcConnectionKey, retries);
- client.connect();
+ new RpcClientManager.RpcConnectionKey(address, DummyProtocol.class, false);
+
+ BlockingRpcClient client = manager.newClient(rpcConnectionKey,
+ retries, 0, TimeUnit.MILLISECONDS, false);
assertTrue(client.isConnected());
+
BlockingInterface stub = client.getStub();
+ EchoMessage response = stub.echo(null, message);
+ assertEquals(MESSAGE, response.getMessage());
+ client.close();
+ server.shutdown();
+ }
+
+ @Test(timeout = 60000)
+ @SetupRpcConnection(setupRpcServer = false, setupRpcClient = false)
+ public void testClientRetryFailureOnStartup() throws Exception {
+ retries = 2;
+ ServerSocket serverSocket = new ServerSocket(0);
+ final InetSocketAddress address = new InetSocketAddress("127.0.0.1", serverSocket.getLocalPort());
+ serverSocket.close();
EchoMessage message = EchoMessage.newBuilder()
.setMessage(MESSAGE).build();
- EchoMessage response2 = stub.echo(null, message);
- assertEquals(MESSAGE, response2.getMessage());
+
+ RpcClientManager.RpcConnectionKey rpcConnectionKey =
+ new RpcClientManager.RpcConnectionKey(address, DummyProtocol.class, false);
+ BlockingRpcClient client = new BlockingRpcClient(rpcConnectionKey, retries);
+
+ try {
+ client.connect();
+ fail();
+ } catch (ConnectTimeoutException e) {
+ assertFalse(e.getMessage(), client.isConnected());
+ }
+
+ BlockingInterface stub = client.getStub();
+ try {
+ EchoMessage response = stub.echo(null, message);
+ fail();
+ } catch (TajoServiceException e) {
+ assertFalse(e.getMessage(), client.isConnected());
+ }
+ RpcClientManager.cleanup(client);
+ }
+
+ @Test(timeout = 60000)
+ @SetupRpcConnection(setupRpcServer = false, setupRpcClient = false)
+ public void testUnresolvedAddress() throws Exception {
+ InetSocketAddress address = new InetSocketAddress("test", 0);
+ boolean expected = false;
+ BlockingRpcClient client = null;
+ try {
+ RpcClientManager.RpcConnectionKey rpcConnectionKey =
+ new RpcClientManager.RpcConnectionKey(address, DummyProtocol.class, true);
+ client = new BlockingRpcClient(rpcConnectionKey, retries);
+ client.connect();
+ fail();
+ } catch (ConnectException e) {
+ expected = true;
+ } catch (Throwable throwable) {
+ fail(throwable.getMessage());
+ } finally {
+ client.close();
+ }
+ assertTrue(expected);
}
@Test
+ @SetupRpcConnection(setupRpcClient = false)
+ public void testUnresolvedAddress2() throws Exception {
+ String hostAndPort = RpcUtils.normalizeInetSocketAddress(server.getListenAddress());
+ RpcClientManager.RpcConnectionKey rpcConnectionKey =
+ new RpcClientManager.RpcConnectionKey(
+ RpcUtils.createUnresolved(hostAndPort), DummyProtocol.class, false);
+ BlockingRpcClient client = new BlockingRpcClient(rpcConnectionKey, retries);
+ client.connect();
+ assertTrue(client.isConnected());
+
+ try {
+ BlockingInterface stub = client.getStub();
+ EchoMessage message = EchoMessage.newBuilder()
+ .setMessage(MESSAGE).build();
+ EchoMessage response2 = stub.echo(null, message);
+ assertEquals(MESSAGE, response2.getMessage());
+ } finally {
+ client.close();
+ }
+ }
+
+ @Test(timeout = 60000)
+ @SetupRpcConnection(setupRpcClient = false)
+ public void testStubRecovery() throws Exception {
+ RpcClientManager.RpcConnectionKey rpcConnectionKey =
+ new RpcClientManager.RpcConnectionKey(server.getListenAddress(), DummyProtocol.class, false);
+ BlockingRpcClient client = manager.newClient(rpcConnectionKey, 1, 0, TimeUnit.MILLISECONDS, false);
+
+ EchoMessage echoMessage = EchoMessage.newBuilder()
+ .setMessage(MESSAGE).build();
+ int repeat = 5;
+
+ assertTrue(client.isConnected());
+ BlockingInterface stub = client.getStub();
+
+ client.close(); // close connection
+ assertFalse(client.isConnected());
+
+ for (int i = 0; i < repeat; i++) {
+ try {
+ EchoMessage response = stub.echo(null, echoMessage);
+ assertEquals(MESSAGE, response.getMessage());
+ assertTrue(client.isConnected());
+ } finally {
+ client.close(); // close connection
+ assertFalse(client.isConnected());
+ }
+ }
+ }
+
+
+ @Test(timeout = 60000)
+ @SetupRpcConnection(setupRpcClient = false)
public void testIdleTimeout() throws Exception {
RpcClientManager.RpcConnectionKey rpcConnectionKey =
new RpcClientManager.RpcConnectionKey(server.getListenAddress(), DummyProtocol.class, false);
- BlockingRpcClient client = new BlockingRpcClient(rpcConnectionKey, retries, 1); //1 sec idle timeout
- client.connect();
+ //500 millis idle timeout
+ BlockingRpcClient client = manager.newClient(rpcConnectionKey, retries, 500, TimeUnit.MILLISECONDS, false);
assertTrue(client.isConnected());
- Thread.sleep(2000);
+ Thread.sleep(600); //timeout
assertFalse(client.isConnected());
client.connect(); // try to reconnect
assertTrue(client.isConnected());
- client.close();
+
+ Thread.sleep(600); //timeout
assertFalse(client.isConnected());
+ client.close();
}
- @Test
- public void testIdleTimeoutWithActiveRequest() throws Exception {
+ @Test(timeout = 60000)
+ @SetupRpcConnection(setupRpcClient = false)
+ public void testPingOnIdle() throws Exception {
RpcClientManager.RpcConnectionKey rpcConnectionKey =
new RpcClientManager.RpcConnectionKey(server.getListenAddress(), DummyProtocol.class, false);
- BlockingRpcClient client = new BlockingRpcClient(rpcConnectionKey, retries, 1); //1 sec idle timeout
- client.connect();
+ //500 millis request timeout
+ BlockingRpcClient client = manager.newClient(rpcConnectionKey, retries, 500, TimeUnit.MILLISECONDS, true);
+ assertTrue(client.isConnected());
+
+ Thread.sleep(600);
+ assertTrue(client.isConnected());
+ Thread.sleep(600);
assertTrue(client.isConnected());
+ client.close();
+ }
+
+ @Test(timeout = 60000)
+ @SetupRpcConnection(setupRpcClient = false)
+ public void testIdleTimeoutWithActiveRequest() throws Exception {
+ RpcClientManager.RpcConnectionKey rpcConnectionKey =
+ new RpcClientManager.RpcConnectionKey(server.getListenAddress(), DummyProtocol.class, false);
+ //500 millis idle timeout
+ BlockingRpcClient client = manager.newClient(rpcConnectionKey, retries, 500, TimeUnit.MILLISECONDS, false);
+ assertTrue(client.isConnected());
+
BlockingInterface stub = client.getStub();
EchoMessage echoMessage = EchoMessage.newBuilder()
.setMessage(MESSAGE).build();
- EchoMessage message = stub.deley(null, echoMessage); //3 sec delay
+ EchoMessage message = stub.delay(null, echoMessage); //3 sec delay
assertEquals(message, echoMessage);
+ assertTrue(client.isConnected());
- Thread.sleep(2000);
+ assertTrue(client.getActiveRequests() == 0);
+ Thread.sleep(600); //timeout
assertFalse(client.isConnected());
}
+
+ @Test(timeout = 60000)
+ @SetupRpcConnection(setupRpcClient = false)
+ public void testRequestTimeoutOnBusy() throws Exception {
+ RpcClientManager.RpcConnectionKey rpcConnectionKey =
+ new RpcClientManager.RpcConnectionKey(server.getListenAddress(), DummyProtocol.class, false);
+
+ //500 millis request timeout
+ BlockingRpcClient client = manager.newClient(rpcConnectionKey, retries, 500, TimeUnit.MILLISECONDS, true);
+ assertTrue(client.isConnected());
+
+ BlockingInterface stub = client.getStub();
+ EchoMessage echoMessage = EchoMessage.newBuilder()
+ .setMessage(MESSAGE).build();
+
+ boolean expected = false;
+ try {
+ EchoMessage message = stub.busy(null, echoMessage); //30 sec delay
+ fail();
+ } catch (TajoServiceException e) {
+ expected = true;
+ } finally {
+ client.close();
+ }
+ assertTrue(expected);
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/TestRpcClientManager.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/TestRpcClientManager.java b/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/TestRpcClientManager.java
index 5f86518..71a2b6f 100644
--- a/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/TestRpcClientManager.java
+++ b/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/TestRpcClientManager.java
@@ -28,10 +28,9 @@ import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.junit.Assert.*;
public class TestRpcClientManager {
@@ -39,59 +38,141 @@ public class TestRpcClientManager {
public void testRaceCondition() throws Exception {
final int parallelCount = 50;
final DummyProtocolAsyncImpl service = new DummyProtocolAsyncImpl();
+ ExecutorService executor = Executors.newFixedThreadPool(parallelCount);
+
NettyServerBase server = new AsyncRpcServer(DummyProtocol.class,
service, new InetSocketAddress("127.0.0.1", 0), parallelCount);
server.start();
+ try {
+ final InetSocketAddress address = server.getListenAddress();
+ final RpcClientManager manager = RpcClientManager.getInstance();
- final InetSocketAddress address = server.getListenAddress();
- final RpcClientManager manager = RpcClientManager.getInstance();
-
- ExecutorService executor = Executors.newFixedThreadPool(parallelCount);
- List<Future> tasks = new ArrayList<Future>();
- for (int i = 0; i < parallelCount; i++) {
- tasks.add(executor.submit(new Runnable() {
- @Override
- public void run() {
- NettyClientBase client = null;
- try {
- client = manager.getClient(address, DummyProtocol.class, false);
- } catch (Throwable e) {
- fail(e.getMessage());
+ List<Future> tasks = new ArrayList<Future>();
+ for (int i = 0; i < parallelCount; i++) {
+ tasks.add(executor.submit(new Runnable() {
+ @Override
+ public void run() {
+ NettyClientBase client = null;
+ try {
+ client = manager.getClient(address, DummyProtocol.class, false);
+ } catch (Throwable e) {
+ fail(e.getMessage());
+ }
+ assertTrue(client.isConnected());
}
- assertTrue(client.isConnected());
- }
- })
- );
+ })
+ );
+ }
+
+ for (Future future : tasks) {
+ future.get();
+ }
+
+ NettyClientBase clientBase = manager.getClient(address, DummyProtocol.class, false);
+ RpcClientManager.cleanup(clientBase);
+ } finally {
+ server.shutdown();
+ executor.shutdown();
+ RpcClientManager.close();
}
+ }
+
+ @Test
+ public void testClientCloseEvent() throws Exception {
+ final DummyProtocolAsyncImpl service = new DummyProtocolAsyncImpl();
+ NettyServerBase server = new AsyncRpcServer(DummyProtocol.class,
+ service, new InetSocketAddress("127.0.0.1", 0), 3);
+ server.start();
+ RpcClientManager manager = RpcClientManager.getInstance();
+
+ try {
+
+ NettyClientBase client = manager.getClient(server.getListenAddress(), DummyProtocol.class, true);
+ assertTrue(client.isConnected());
+ assertTrue(client.getChannel().isWritable());
+
+ RpcClientManager.RpcConnectionKey key = client.getKey();
+ assertTrue(RpcClientManager.contains(key));
- for (Future future : tasks) {
- future.get();
+ client.close();
+ assertFalse(RpcClientManager.contains(key));
+ } finally {
+ server.shutdown();
+ RpcClientManager.close();
}
+ }
+
+ @Test
+ public void testClientCloseEventWithReconnect() throws Exception {
+ final DummyProtocolAsyncImpl service = new DummyProtocolAsyncImpl();
+ NettyServerBase server = new AsyncRpcServer(DummyProtocol.class,
+ service, new InetSocketAddress("127.0.0.1", 0), 3);
+ server.start();
+ int repeat = 10;
+ RpcClientManager manager = RpcClientManager.getInstance();
+
+ try {
- NettyClientBase clientBase = manager.getClient(address, DummyProtocol.class, false);
- RpcClientManager.cleanup(clientBase);
- server.shutdown();
- executor.shutdown();
+ NettyClientBase client = manager.getClient(server.getListenAddress(), DummyProtocol.class, true);
+ assertTrue(client.isConnected());
+
+ RpcClientManager.RpcConnectionKey key = client.getKey();
+ assertTrue(RpcClientManager.contains(key));
+
+ client.close();
+ assertFalse(client.isConnected());
+ assertFalse(RpcClientManager.contains(key));
+
+ for (int i = 0; i < repeat; i++) {
+ client.connect();
+ assertTrue(client.isConnected());
+ assertTrue(RpcClientManager.contains(key));
+
+ client.close();
+ assertFalse(client.isConnected());
+ assertFalse(RpcClientManager.contains(key));
+ }
+ } finally {
+ server.shutdown();
+ RpcClientManager.close();
+ }
}
@Test
- public void testCloseFuture() throws Exception {
+ public void testUnManagedClient() throws Exception {
final DummyProtocolAsyncImpl service = new DummyProtocolAsyncImpl();
NettyServerBase server = new AsyncRpcServer(DummyProtocol.class,
service, new InetSocketAddress("127.0.0.1", 0), 3);
server.start();
+ RpcClientManager.RpcConnectionKey key =
+ new RpcClientManager.RpcConnectionKey(server.getListenAddress(), DummyProtocol.class, true);
+ RpcClientManager.close();
+ RpcClientManager manager = RpcClientManager.getInstance();
+
+ try {
+ NettyClientBase client1 = manager.newClient(key, 0, 0, TimeUnit.SECONDS, false);
+ assertTrue(client1.isConnected());
+ assertFalse(RpcClientManager.contains(key));
- final RpcClientManager manager = RpcClientManager.getInstance();
+ NettyClientBase client2 = manager.newClient(key, 0, 0, TimeUnit.SECONDS, false);
+ assertTrue(client2.isConnected());
+ assertFalse(RpcClientManager.contains(key));
- NettyClientBase client = manager.getClient(server.getListenAddress(), DummyProtocol.class, true);
- assertTrue(client.isConnected());
- assertTrue(client.getChannel().isWritable());
+ assertEquals(client1.getRemoteAddress(), client2.getRemoteAddress());
+ assertNotEquals(client1.getChannel(), client2.getChannel());
- RpcClientManager.RpcConnectionKey key = client.getKey();
- assertTrue(RpcClientManager.contains(key));
+ client1.close();
+ assertFalse(client1.isConnected());
+ assertTrue(client2.isConnected());
- client.close();
- assertFalse(RpcClientManager.contains(key));
- server.shutdown();
+ client1.connect();
+ client2.close();
+ assertFalse(client2.isConnected());
+ assertTrue(client1.isConnected());
+
+ RpcClientManager.cleanup(client1, client2);
+ } finally {
+ server.shutdown();
+ }
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/test/impl/DummyProtocolAsyncImpl.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/test/impl/DummyProtocolAsyncImpl.java b/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/test/impl/DummyProtocolAsyncImpl.java
index 0ca7563..abcc057 100644
--- a/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/test/impl/DummyProtocolAsyncImpl.java
+++ b/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/test/impl/DummyProtocolAsyncImpl.java
@@ -68,7 +68,7 @@ public class DummyProtocolAsyncImpl implements Interface {
}
@Override
- public void deley(RpcController controller, EchoMessage request,
+ public void delay(RpcController controller, EchoMessage request,
RpcCallback<EchoMessage> done) {
try {
Thread.sleep(3000);
@@ -79,8 +79,25 @@ public class DummyProtocolAsyncImpl implements Interface {
done.run(request);
}
+ @Override
+ public void busy(RpcController controller, EchoMessage request,
+ RpcCallback<EchoMessage> done) {
+ try {
+ Thread.sleep(30000);
+ } catch (InterruptedException e) {
+ LOG.error(e.getMessage());
+ }
+ done.run(request);
+ }
+
+ @Override
public void throwException(RpcController controller, EchoMessage request,
RpcCallback<EchoMessage> done) {
- done.run(request);
+ if(controller != null) {
+ controller.setFailed("throwException");
+ done.run(request);
+ } else {
+ throw new RuntimeException("throwException");
+ }
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/test/impl/DummyProtocolBlockingImpl.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/test/impl/DummyProtocolBlockingImpl.java b/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/test/impl/DummyProtocolBlockingImpl.java
index 8d4b597..40eb18f 100644
--- a/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/test/impl/DummyProtocolBlockingImpl.java
+++ b/tajo-rpc/tajo-rpc-protobuf/src/test/java/org/apache/tajo/rpc/test/impl/DummyProtocolBlockingImpl.java
@@ -65,7 +65,7 @@ public class DummyProtocolBlockingImpl implements BlockingInterface {
}
@Override
- public EchoMessage deley(RpcController controller, EchoMessage request)
+ public EchoMessage delay(RpcController controller, EchoMessage request)
throws ServiceException {
try {
Thread.sleep(3000);
@@ -76,6 +76,17 @@ public class DummyProtocolBlockingImpl implements BlockingInterface {
return request;
}
+ @Override
+ public EchoMessage busy(RpcController controller, EchoMessage request) {
+ try {
+ Thread.sleep(30000);
+ } catch (InterruptedException e) {
+ LOG.error(e.getMessage());
+ }
+ return request;
+ }
+
+ @Override
public EchoMessage throwException(RpcController controller, EchoMessage request)
throws ServiceException {
throw new ServiceException("Exception Test");
[4/4] tajo git commit: Merge branch 'master' of
https://git-wip-us.apache.org/repos/asf/tajo into index_support
Posted by ji...@apache.org.
Merge branch 'master' of https://git-wip-us.apache.org/repos/asf/tajo into index_support
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/e6b91c33
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/e6b91c33
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/e6b91c33
Branch: refs/heads/index_support
Commit: e6b91c333ae251102f2d7ea6a68d40ee0510028a
Parents: 2fe3db4 9d73267
Author: Jihoon Son <ji...@apache.org>
Authored: Wed Apr 29 10:44:21 2015 +0900
Committer: Jihoon Son <ji...@apache.org>
Committed: Wed Apr 29 10:44:21 2015 +0900
----------------------------------------------------------------------
.travis.yml | 2 +-
CHANGES | 4 +
.../apache/tajo/client/SessionConnection.java | 12 +-
.../java/org/apache/tajo/master/TajoMaster.java | 9 +-
.../tajo/master/rm/TajoResourceTracker.java | 8 +-
.../apache/tajo/querymaster/QueryMaster.java | 13 +-
.../tajo/worker/ExecutionBlockContext.java | 4 +-
.../tajo/worker/TajoResourceAllocator.java | 6 +-
.../java/org/apache/tajo/worker/TajoWorker.java | 7 +
.../java/org/apache/tajo/worker/TaskRunner.java | 14 +-
.../tajo/worker/WorkerHeartbeatService.java | 19 +-
.../tajo/ws/rs/resources/DatabasesResource.java | 8 +-
.../responses/WorkerConnectionInfoResponse.java | 25 +-
.../tajo/ws/rs/responses/WorkerResponse.java | 47 ++-
.../org/apache/tajo/rpc/RpcChannelFactory.java | 6 +-
.../java/org/apache/tajo/rpc/RpcConstants.java | 32 ++
.../org/apache/tajo/rpc/AsyncRpcClient.java | 206 +++-------
.../org/apache/tajo/rpc/AsyncRpcServer.java | 65 +--
.../org/apache/tajo/rpc/BlockingRpcClient.java | 214 ++++------
.../org/apache/tajo/rpc/BlockingRpcServer.java | 64 +--
.../java/org/apache/tajo/rpc/CallFuture.java | 32 +-
.../apache/tajo/rpc/ChannelEventListener.java | 39 ++
.../tajo/rpc/ConnectionCloseFutureListener.java | 35 --
.../apache/tajo/rpc/DefaultRpcController.java | 2 +-
.../apache/tajo/rpc/MonitorClientHandler.java | 101 +++++
.../apache/tajo/rpc/MonitorServerHandler.java | 73 ++++
.../org/apache/tajo/rpc/MonitorStateEvent.java | 49 +++
.../org/apache/tajo/rpc/NettyClientBase.java | 303 ++++++++++++--
.../tajo/rpc/ProtoChannelInitializer.java | 57 ---
.../tajo/rpc/ProtoClientChannelInitializer.java | 63 +++
.../org/apache/tajo/rpc/ProtoDeclaration.java | 24 ++
.../tajo/rpc/ProtoServerChannelInitializer.java | 50 +++
.../apache/tajo/rpc/RecoverableException.java | 42 ++
.../org/apache/tajo/rpc/RpcClientManager.java | 123 ++++--
.../src/main/proto/TestProtocol.proto | 3 +-
.../java/org/apache/tajo/rpc/TestAsyncRpc.java | 350 +++++++++++++----
.../org/apache/tajo/rpc/TestBlockingRpc.java | 392 +++++++++++++------
.../apache/tajo/rpc/TestRpcClientManager.java | 155 ++++++--
.../rpc/test/impl/DummyProtocolAsyncImpl.java | 21 +-
.../test/impl/DummyProtocolBlockingImpl.java | 13 +-
40 files changed, 1882 insertions(+), 810 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/e6b91c33/tajo-client/src/main/java/org/apache/tajo/client/SessionConnection.java
----------------------------------------------------------------------
[2/4] tajo git commit: TAJO-1563: Improve RPC error handling. (jinho)
Posted by ji...@apache.org.
TAJO-1563: Improve RPC error handling. (jinho)
Closes #554
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/2e7d03df
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/2e7d03df
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/2e7d03df
Branch: refs/heads/index_support
Commit: 2e7d03dff9480ecb7a4e4ed09a5578ba618de2ae
Parents: 1971d85
Author: Jinho Kim <jh...@apache.org>
Authored: Tue Apr 28 18:27:05 2015 +0900
Committer: Jinho Kim <jh...@apache.org>
Committed: Tue Apr 28 18:27:05 2015 +0900
----------------------------------------------------------------------
CHANGES | 2 +
.../apache/tajo/client/SessionConnection.java | 12 +-
.../java/org/apache/tajo/master/TajoMaster.java | 9 +-
.../tajo/master/rm/TajoResourceTracker.java | 8 +-
.../apache/tajo/querymaster/QueryMaster.java | 13 +-
.../tajo/worker/ExecutionBlockContext.java | 4 +-
.../tajo/worker/TajoResourceAllocator.java | 6 +-
.../java/org/apache/tajo/worker/TajoWorker.java | 7 +
.../java/org/apache/tajo/worker/TaskRunner.java | 14 +-
.../tajo/worker/WorkerHeartbeatService.java | 19 +-
.../tajo/ws/rs/resources/DatabasesResource.java | 8 +-
.../responses/WorkerConnectionInfoResponse.java | 25 +-
.../tajo/ws/rs/responses/WorkerResponse.java | 47 ++-
.../org/apache/tajo/rpc/RpcChannelFactory.java | 6 +-
.../java/org/apache/tajo/rpc/RpcConstants.java | 32 ++
.../org/apache/tajo/rpc/AsyncRpcClient.java | 206 +++-------
.../org/apache/tajo/rpc/AsyncRpcServer.java | 65 +--
.../org/apache/tajo/rpc/BlockingRpcClient.java | 214 ++++------
.../org/apache/tajo/rpc/BlockingRpcServer.java | 64 +--
.../java/org/apache/tajo/rpc/CallFuture.java | 32 +-
.../apache/tajo/rpc/ChannelEventListener.java | 39 ++
.../tajo/rpc/ConnectionCloseFutureListener.java | 35 --
.../apache/tajo/rpc/DefaultRpcController.java | 2 +-
.../apache/tajo/rpc/MonitorClientHandler.java | 101 +++++
.../apache/tajo/rpc/MonitorServerHandler.java | 73 ++++
.../org/apache/tajo/rpc/MonitorStateEvent.java | 49 +++
.../org/apache/tajo/rpc/NettyClientBase.java | 303 ++++++++++++--
.../tajo/rpc/ProtoChannelInitializer.java | 57 ---
.../tajo/rpc/ProtoClientChannelInitializer.java | 63 +++
.../org/apache/tajo/rpc/ProtoDeclaration.java | 24 ++
.../tajo/rpc/ProtoServerChannelInitializer.java | 50 +++
.../apache/tajo/rpc/RecoverableException.java | 42 ++
.../org/apache/tajo/rpc/RpcClientManager.java | 123 ++++--
.../src/main/proto/TestProtocol.proto | 3 +-
.../java/org/apache/tajo/rpc/TestAsyncRpc.java | 350 +++++++++++++----
.../org/apache/tajo/rpc/TestBlockingRpc.java | 392 +++++++++++++------
.../apache/tajo/rpc/TestRpcClientManager.java | 155 ++++++--
.../rpc/test/impl/DummyProtocolAsyncImpl.java | 21 +-
.../test/impl/DummyProtocolBlockingImpl.java | 13 +-
39 files changed, 1879 insertions(+), 809 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 7b16e11..93fb092 100644
--- a/CHANGES
+++ b/CHANGES
@@ -22,6 +22,8 @@ Release 0.11.0 - unreleased
IMPROVEMENT
+ TAJO-1563: Improve RPC error handling. (jinho)
+
TAJO-1311: Enable Scattered Hash Shuffle for CTAS statement. (jaehwa)
TAJO-1548: Refactoring condition code for CHAR into CatalogUtil.
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-client/src/main/java/org/apache/tajo/client/SessionConnection.java
----------------------------------------------------------------------
diff --git a/tajo-client/src/main/java/org/apache/tajo/client/SessionConnection.java b/tajo-client/src/main/java/org/apache/tajo/client/SessionConnection.java
index d8152f4..be757af 100644
--- a/tajo-client/src/main/java/org/apache/tajo/client/SessionConnection.java
+++ b/tajo-client/src/main/java/org/apache/tajo/client/SessionConnection.java
@@ -31,15 +31,15 @@ import org.apache.tajo.ipc.ClientProtos.SessionUpdateResponse;
import org.apache.tajo.ipc.TajoMasterClientProtocol;
import org.apache.tajo.rpc.NettyClientBase;
import org.apache.tajo.rpc.RpcClientManager;
+import org.apache.tajo.rpc.RpcConstants;
import org.apache.tajo.rpc.ServerCallable;
import org.apache.tajo.service.ServiceTracker;
import org.apache.tajo.util.KeyValueSet;
import org.apache.tajo.util.ProtoUtil;
-import io.netty.channel.ConnectTimeoutException;
-
import java.io.Closeable;
import java.io.IOException;
+import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.util.Collections;
import java.util.HashMap;
@@ -87,6 +87,10 @@ public class SessionConnection implements Closeable {
this.properties = properties;
this.manager = RpcClientManager.getInstance();
+ this.manager.setRetries(properties.getInt(RpcConstants.RPC_CLIENT_RETRY_MAX, RpcConstants.DEFAULT_RPC_RETRIES));
+ this.manager.setTimeoutSeconds(
+ properties.getInt(RpcConstants.RPC_CLIENT_TIMEOUT_SECS, 0)); // disable rpc timeout
+
this.userInfo = UserRoleInfo.getCurrentUser();
this.baseDatabase = baseDatabase != null ? baseDatabase : null;
@@ -98,12 +102,12 @@ public class SessionConnection implements Closeable {
}
public NettyClientBase getTajoMasterConnection(boolean asyncMode) throws NoSuchMethodException,
- ConnectTimeoutException, ClassNotFoundException {
+ ConnectException, ClassNotFoundException {
return manager.getClient(getTajoMasterAddr(), TajoMasterClientProtocol.class, asyncMode);
}
public NettyClientBase getConnection(InetSocketAddress addr, Class protocolClass, boolean asyncMode)
- throws NoSuchMethodException, ConnectTimeoutException, ClassNotFoundException {
+ throws NoSuchMethodException, ConnectException, ClassNotFoundException {
return manager.getClient(addr, protocolClass, asyncMode);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java b/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
index 0a5de58..d6ae49c 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/TajoMaster.java
@@ -44,6 +44,8 @@ import org.apache.tajo.master.rm.WorkerResourceManager;
import org.apache.tajo.metrics.CatalogMetricsGaugeSet;
import org.apache.tajo.metrics.WorkerResourceMetricsGaugeSet;
import org.apache.tajo.rpc.RpcChannelFactory;
+import org.apache.tajo.rpc.RpcClientManager;
+import org.apache.tajo.rpc.RpcConstants;
import org.apache.tajo.rule.EvaluationContext;
import org.apache.tajo.rule.EvaluationFailedException;
import org.apache.tajo.rule.SelfDiagnosisRuleEngine;
@@ -167,8 +169,12 @@ public class TajoMaster extends CompositeService {
try {
RackResolver.init(systemConf);
+ RpcClientManager rpcManager = RpcClientManager.getInstance();
+ rpcManager.setRetries(systemConf.getInt(RpcConstants.RPC_CLIENT_RETRY_MAX, RpcConstants.DEFAULT_RPC_RETRIES));
+ rpcManager.setTimeoutSeconds(
+ systemConf.getInt(RpcConstants.RPC_CLIENT_TIMEOUT_SECS, RpcConstants.DEFAULT_RPC_TIMEOUT_SECONDS));
+
initResourceManager();
- initWebServer();
this.dispatcher = new AsyncDispatcher();
addIfService(dispatcher);
@@ -328,6 +334,7 @@ public class TajoMaster extends CompositeService {
LOG.error(e.getMessage(), e);
}
+ initWebServer();
initSystemMetrics();
haService = ServiceTrackerFactory.get(systemConf);
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-core/src/main/java/org/apache/tajo/master/rm/TajoResourceTracker.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/master/rm/TajoResourceTracker.java b/tajo-core/src/main/java/org/apache/tajo/master/rm/TajoResourceTracker.java
index ba021fc..4f3b66a 100644
--- a/tajo-core/src/main/java/org/apache/tajo/master/rm/TajoResourceTracker.java
+++ b/tajo-core/src/main/java/org/apache/tajo/master/rm/TajoResourceTracker.java
@@ -76,7 +76,7 @@ public class TajoResourceTracker extends AbstractService implements TajoResource
}
@Override
- public void serviceInit(Configuration conf) {
+ public void serviceInit(Configuration conf) throws Exception {
if (!(conf instanceof TajoConf)) {
throw new IllegalArgumentException("Configuration must be a TajoConf instance");
}
@@ -98,17 +98,17 @@ public class TajoResourceTracker extends AbstractService implements TajoResource
systemConf.setVar(TajoConf.ConfVars.RESOURCE_TRACKER_RPC_ADDRESS, NetUtils.normalizeInetSocketAddress(bindAddress));
LOG.info("TajoResourceTracker starts up (" + this.bindAddress + ")");
- super.start();
+ super.serviceInit(conf);
}
@Override
- public void serviceStop() {
+ public void serviceStop() throws Exception {
// server can be null if some exception occurs before the rpc server starts up.
if(server != null) {
server.shutdown();
server = null;
}
- super.stop();
+ super.serviceStop();
}
/** The response builder */
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-core/src/main/java/org/apache/tajo/querymaster/QueryMaster.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/querymaster/QueryMaster.java b/tajo-core/src/main/java/org/apache/tajo/querymaster/QueryMaster.java
index 67dae06..6c5bd22 100644
--- a/tajo-core/src/main/java/org/apache/tajo/querymaster/QueryMaster.java
+++ b/tajo-core/src/main/java/org/apache/tajo/querymaster/QueryMaster.java
@@ -38,10 +38,7 @@ import org.apache.tajo.ipc.QueryCoordinatorProtocol.*;
import org.apache.tajo.ipc.TajoWorkerProtocol;
import org.apache.tajo.master.event.QueryStartEvent;
import org.apache.tajo.master.event.QueryStopEvent;
-import org.apache.tajo.rpc.CallFuture;
-import org.apache.tajo.rpc.NettyClientBase;
-import org.apache.tajo.rpc.NullCallback;
-import org.apache.tajo.rpc.RpcClientManager;
+import org.apache.tajo.rpc.*;
import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos;
import org.apache.tajo.service.ServiceTracker;
import org.apache.tajo.util.NetUtils;
@@ -192,10 +189,8 @@ public class QueryMaster extends CompositeService implements EventHandler {
TajoWorkerProtocol.TajoWorkerProtocolService tajoWorkerProtocolService = rpc.getStub();
tajoWorkerProtocolService.cleanupExecutionBlocks(null, executionBlockListProto, NullCallback.get());
- } catch (RuntimeException e) {
- LOG.warn("Ignoring RuntimeException. " + e.getMessage(), e);
- continue;
} catch (Exception e) {
+ LOG.warn("Ignoring exception. " + e.getMessage(), e);
continue;
}
}
@@ -341,6 +336,7 @@ public class QueryMaster extends CompositeService implements EventHandler {
QueryCoordinatorProtocolService masterClientService = tmClient.getStub();
masterClientService.heartbeat(future.getController(), queryHeartbeat, future);
+ future.get(RpcConstants.DEFAULT_FUTURE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
} catch (Exception e) {
//this function will be closed in new thread.
//When tajo do stop cluster, tajo master maybe throw closed connection exception
@@ -441,9 +437,8 @@ public class QueryMaster extends CompositeService implements EventHandler {
QueryCoordinatorProtocol.class, true);
QueryCoordinatorProtocolService masterClientService = tmClient.getStub();
- CallFuture<TajoHeartbeatResponse> callBack = new CallFuture<TajoHeartbeatResponse>();
TajoHeartbeat queryHeartbeat = buildTajoHeartBeat(eachTask);
- masterClientService.heartbeat(callBack.getController(), queryHeartbeat, callBack);
+ masterClientService.heartbeat(null, queryHeartbeat, NullCallback.get());
} catch (Throwable t) {
t.printStackTrace();
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-core/src/main/java/org/apache/tajo/worker/ExecutionBlockContext.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/ExecutionBlockContext.java b/tajo-core/src/main/java/org/apache/tajo/worker/ExecutionBlockContext.java
index fcf787e..0d26e6c 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/ExecutionBlockContext.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/ExecutionBlockContext.java
@@ -20,7 +20,6 @@ package org.apache.tajo.worker;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
-import io.netty.channel.ConnectTimeoutException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FileSystem;
@@ -45,6 +44,7 @@ import org.apache.tajo.util.NetUtils;
import org.apache.tajo.util.Pair;
import java.io.IOException;
+import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
@@ -154,7 +154,7 @@ public class ExecutionBlockContext {
}
public NettyClientBase getQueryMasterConnection()
- throws NoSuchMethodException, ConnectTimeoutException, ClassNotFoundException {
+ throws NoSuchMethodException, ConnectException, ClassNotFoundException {
return connManager.getClient(qmMasterAddr, QueryMasterProtocol.class, true);
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-core/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java b/tajo-core/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
index 7ba2ebc..05dd1a9 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
@@ -318,7 +318,7 @@ public class TajoResourceAllocator extends AbstractResourceAllocator {
tmClient = RpcClientManager.getInstance().
getClient(serviceTracker.getUmbilicalAddress(), QueryCoordinatorProtocol.class, true);
QueryCoordinatorProtocolService masterClientService = tmClient.getStub();
- masterClientService.allocateWorkerResources(null, request, callBack);
+ masterClientService.allocateWorkerResources(callBack.getController(), request, callBack);
} catch (Throwable e) {
LOG.error(e.getMessage(), e);
}
@@ -335,8 +335,12 @@ public class TajoResourceAllocator extends AbstractResourceAllocator {
} catch (TimeoutException e) {
LOG.info("No available worker resource for " + event.getExecutionBlockId());
continue;
+ } catch (ExecutionException e) {
+ LOG.error(e.getMessage(), e);
+ break;
}
}
+
int numAllocatedContainers = 0;
if(response != null) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java b/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java
index 17af71a..79b83e4 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/TajoWorker.java
@@ -37,6 +37,8 @@ import org.apache.tajo.catalog.FunctionDesc;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.engine.function.FunctionLoader;
import org.apache.tajo.function.FunctionSignature;
+import org.apache.tajo.rpc.RpcClientManager;
+import org.apache.tajo.rpc.RpcConstants;
import org.apache.tajo.service.ServiceTracker;
import org.apache.tajo.service.ServiceTrackerFactory;
import org.apache.tajo.service.TajoMasterInfo;
@@ -152,6 +154,11 @@ public class TajoWorker extends CompositeService {
this.systemConf = (TajoConf)conf;
RackResolver.init(systemConf);
+ RpcClientManager rpcManager = RpcClientManager.getInstance();
+ rpcManager.setRetries(systemConf.getInt(RpcConstants.RPC_CLIENT_RETRY_MAX, RpcConstants.DEFAULT_RPC_RETRIES));
+ rpcManager.setTimeoutSeconds(
+ systemConf.getInt(RpcConstants.RPC_CLIENT_TIMEOUT_SECS, RpcConstants.DEFAULT_RPC_TIMEOUT_SECONDS));
+
serviceTracker = ServiceTrackerFactory.get(systemConf);
this.workerContext = new WorkerContext();
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-core/src/main/java/org/apache/tajo/worker/TaskRunner.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/TaskRunner.java b/tajo-core/src/main/java/org/apache/tajo/worker/TaskRunner.java
index 642c914..6076913 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/TaskRunner.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/TaskRunner.java
@@ -19,7 +19,6 @@
package org.apache.tajo.worker;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
-
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
@@ -38,8 +37,7 @@ import org.apache.tajo.rpc.CallFuture;
import org.apache.tajo.rpc.NettyClientBase;
import org.apache.tajo.rpc.NullCallback;
-import io.netty.channel.ConnectTimeoutException;
-
+import java.net.ConnectException;
import java.util.concurrent.*;
import static org.apache.tajo.ipc.TajoWorkerProtocol.*;
@@ -200,7 +198,7 @@ public class TaskRunner extends AbstractService {
NettyClientBase client;
try {
client = executionBlockContext.getQueryMasterConnection();
- } catch (ConnectTimeoutException ce) {
+ } catch (ConnectException ce) {
// NettyClientBase throws ConnectTimeoutException if connection was failed
stop();
getContext().stopTaskRunner(getId());
@@ -238,17 +236,15 @@ public class TaskRunner extends AbstractService {
if(stopped) {
break;
}
-
- if(callFuture.getController().failed()){
- LOG.error(callFuture.getController().errorText());
- break;
- }
// if there has been no assigning task for a given period,
// TaskRunner will retry to request an assigning task.
if (LOG.isDebugEnabled()) {
LOG.info("Retry assigning task:" + getId());
}
continue;
+ } catch (ExecutionException ee) {
+ LOG.error(ee.getMessage(), ee);
+ break;
}
if (taskRequest != null) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-core/src/main/java/org/apache/tajo/worker/WorkerHeartbeatService.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/worker/WorkerHeartbeatService.java b/tajo-core/src/main/java/org/apache/tajo/worker/WorkerHeartbeatService.java
index f8d5fd9..9afee5a 100644
--- a/tajo-core/src/main/java/org/apache/tajo/worker/WorkerHeartbeatService.java
+++ b/tajo-core/src/main/java/org/apache/tajo/worker/WorkerHeartbeatService.java
@@ -19,13 +19,12 @@
package org.apache.tajo.worker;
import com.google.common.collect.Lists;
-import com.google.protobuf.ServiceException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.service.AbstractService;
import org.apache.tajo.conf.TajoConf;
-import org.apache.tajo.ipc.QueryCoordinatorProtocol.ClusterResourceSummary;
+import org.apache.tajo.ipc.QueryCoordinatorProtocol;
import org.apache.tajo.ipc.QueryCoordinatorProtocol.ServerStatusProto;
import org.apache.tajo.ipc.QueryCoordinatorProtocol.TajoHeartbeatResponse;
import org.apache.tajo.ipc.TajoResourceTrackerProtocol;
@@ -187,17 +186,13 @@ public class WorkerHeartbeatService extends AbstractService {
resourceTracker.heartbeat(callBack.getController(), heartbeatProto, callBack);
TajoHeartbeatResponse response = callBack.get(2, TimeUnit.SECONDS);
- if(response != null) {
- ClusterResourceSummary clusterResourceSummary = response.getClusterResourceSummary();
- if(clusterResourceSummary.getNumWorkers() > 0) {
- context.setNumClusterNodes(clusterResourceSummary.getNumWorkers());
- }
- context.setClusterResource(clusterResourceSummary);
- } else {
- if(callBack.getController().failed()) {
- throw new ServiceException(callBack.getController().errorText());
- }
+
+ QueryCoordinatorProtocol.ClusterResourceSummary clusterResourceSummary = response.getClusterResourceSummary();
+ if(clusterResourceSummary.getNumWorkers() > 0) {
+ context.setNumClusterNodes(clusterResourceSummary.getNumWorkers());
}
+ context.setClusterResource(clusterResourceSummary);
+
} catch (InterruptedException e) {
break;
} catch (TimeoutException te) {
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-core/src/main/java/org/apache/tajo/ws/rs/resources/DatabasesResource.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/ws/rs/resources/DatabasesResource.java b/tajo-core/src/main/java/org/apache/tajo/ws/rs/resources/DatabasesResource.java
index 3868cab..b807198 100644
--- a/tajo-core/src/main/java/org/apache/tajo/ws/rs/resources/DatabasesResource.java
+++ b/tajo-core/src/main/java/org/apache/tajo/ws/rs/resources/DatabasesResource.java
@@ -254,14 +254,18 @@ public class DatabasesResource {
if (selectedDatabase != null) {
List<TablespaceProto> tablespacesList = catalogService.getAllTablespaces();
TablespaceProto selectedTablespace = null;
-
+
for (TablespaceProto tablespace: tablespacesList) {
if (tablespace.hasId() && tablespace.getId() == selectedDatabase.getSpaceId()) {
selectedTablespace = tablespace;
break;
}
}
-
+
+ if(selectedTablespace == null) {
+ return ResourcesUtil.createExceptionResponse(LOG, "Tablespace not found.");
+ }
+
DatabaseInfoResponse databaseInfo = new DatabaseInfoResponse();
databaseInfo.setId(selectedDatabase.getId());
databaseInfo.setName(selectedDatabase.getName());
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-core/src/main/java/org/apache/tajo/ws/rs/responses/WorkerConnectionInfoResponse.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/ws/rs/responses/WorkerConnectionInfoResponse.java b/tajo-core/src/main/java/org/apache/tajo/ws/rs/responses/WorkerConnectionInfoResponse.java
index 3c8cb8b..67e59ac 100644
--- a/tajo-core/src/main/java/org/apache/tajo/ws/rs/responses/WorkerConnectionInfoResponse.java
+++ b/tajo-core/src/main/java/org/apache/tajo/ws/rs/responses/WorkerConnectionInfoResponse.java
@@ -52,16 +52,27 @@ public class WorkerConnectionInfoResponse {
return id;
}
- public void setId(int id) {
- this.id = id;
- }
-
public String getHost() {
return host;
}
- public void setHost(String host) {
- this.host = host;
+ public int getPeerRpcPort() {
+ return peerRpcPort;
+ }
+
+ public int getPullServerPort() {
+ return pullServerPort;
+ }
+
+ public int getQueryMasterPort() {
+ return queryMasterPort;
+ }
+
+ public int getClientPort() {
+ return clientPort;
+ }
+
+ public int getHttpInfoPort() {
+ return httpInfoPort;
}
-
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-core/src/main/java/org/apache/tajo/ws/rs/responses/WorkerResponse.java
----------------------------------------------------------------------
diff --git a/tajo-core/src/main/java/org/apache/tajo/ws/rs/responses/WorkerResponse.java b/tajo-core/src/main/java/org/apache/tajo/ws/rs/responses/WorkerResponse.java
index 84b81f8..ff0399c 100644
--- a/tajo-core/src/main/java/org/apache/tajo/ws/rs/responses/WorkerResponse.java
+++ b/tajo-core/src/main/java/org/apache/tajo/ws/rs/responses/WorkerResponse.java
@@ -69,8 +69,51 @@ public class WorkerResponse {
return connectionInfo;
}
- public void setConnectionInfo(WorkerConnectionInfoResponse connectionInfo) {
- this.connectionInfo = connectionInfo;
+ public int getCpuCoreSlots() {
+ return cpuCoreSlots;
}
+ public int getMemoryMB() {
+ return memoryMB;
+ }
+
+ public float getUsedDiskSlots() {
+ return usedDiskSlots;
+ }
+
+ public int getUsedMemoryMB() {
+ return usedMemoryMB;
+ }
+
+ public int getUsedCpuCoreSlots() {
+ return usedCpuCoreSlots;
+ }
+
+ public long getMaxHeap() {
+ return maxHeap;
+ }
+
+ public long getFreeHeap() {
+ return freeHeap;
+ }
+
+ public long getTotalHeap() {
+ return totalHeap;
+ }
+
+ public int getNumRunningTasks() {
+ return numRunningTasks;
+ }
+
+ public int getNumQueryMasterTasks() {
+ return numQueryMasterTasks;
+ }
+
+ public long getLastHeartbeatTime() {
+ return lastHeartbeatTime;
+ }
+
+ public float getDiskSlots() {
+ return diskSlots;
+ }
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-common/src/main/java/org/apache/tajo/rpc/RpcChannelFactory.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-common/src/main/java/org/apache/tajo/rpc/RpcChannelFactory.java b/tajo-rpc/tajo-rpc-common/src/main/java/org/apache/tajo/rpc/RpcChannelFactory.java
index ed6b634..eb34ca2 100644
--- a/tajo-rpc/tajo-rpc-common/src/main/java/org/apache/tajo/rpc/RpcChannelFactory.java
+++ b/tajo-rpc/tajo-rpc-common/src/main/java/org/apache/tajo/rpc/RpcChannelFactory.java
@@ -19,13 +19,11 @@
package org.apache.tajo.rpc;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import java.util.Map;
import java.util.Queue;
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-common/src/main/java/org/apache/tajo/rpc/RpcConstants.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-common/src/main/java/org/apache/tajo/rpc/RpcConstants.java b/tajo-rpc/tajo-rpc-common/src/main/java/org/apache/tajo/rpc/RpcConstants.java
new file mode 100644
index 0000000..165cd2a
--- /dev/null
+++ b/tajo-rpc/tajo-rpc-common/src/main/java/org/apache/tajo/rpc/RpcConstants.java
@@ -0,0 +1,32 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.rpc;
+
+public class RpcConstants {
+
+ public static final String PING_PACKET = "TAJO";
+ public static final String RPC_CLIENT_RETRY_MAX = "tajo.rpc.client.retry.max";
+ public static final String RPC_CLIENT_TIMEOUT_SECS = "tajo.rpc.client.timeout-secs";
+
+ public static final int DEFAULT_RPC_RETRIES = 3;
+ public static final int DEFAULT_RPC_TIMEOUT_SECONDS = 180;
+ public static final int DEFAULT_CONNECT_TIMEOUT = 60000; // 60 sec
+ public static final int DEFAULT_PAUSE = 1000; // 1 sec
+ public static final int DEFAULT_FUTURE_TIMEOUT_SECONDS = 10;
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/AsyncRpcClient.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/AsyncRpcClient.java b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/AsyncRpcClient.java
index 8f2c2a1..dd7d495 100644
--- a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/AsyncRpcClient.java
+++ b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/AsyncRpcClient.java
@@ -20,120 +20,97 @@ package org.apache.tajo.rpc;
import com.google.protobuf.Descriptors.MethodDescriptor;
import com.google.protobuf.*;
-import io.netty.channel.*;
-import io.netty.handler.timeout.IdleState;
-import io.netty.handler.timeout.IdleStateEvent;
-import io.netty.util.concurrent.GenericFutureListener;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
+import io.netty.channel.ChannelHandler;
import org.apache.tajo.rpc.RpcClientManager.RpcConnectionKey;
-import org.apache.tajo.rpc.RpcProtos.RpcRequest;
import org.apache.tajo.rpc.RpcProtos.RpcResponse;
import java.lang.reflect.Method;
-import java.net.InetSocketAddress;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
-public class AsyncRpcClient extends NettyClientBase {
- private static final Log LOG = LogFactory.getLog(AsyncRpcClient.class);
-
- private final ConcurrentMap<Integer, ResponseCallback> requests =
- new ConcurrentHashMap<Integer, ResponseCallback>();
+public class AsyncRpcClient extends NettyClientBase<AsyncRpcClient.ResponseCallback> {
private final Method stubMethod;
private final ProxyRpcChannel rpcChannel;
- private final ClientChannelInboundHandler inboundHandler;
+ private final NettyChannelInboundHandler handler;
- /**
- * Intentionally make this method package-private, avoiding user directly
- * new an instance through this constructor.
- */
AsyncRpcClient(RpcConnectionKey rpcConnectionKey, int retries)
throws ClassNotFoundException, NoSuchMethodException {
- this(rpcConnectionKey, retries, 0);
+ this(rpcConnectionKey, retries, 0, TimeUnit.NANOSECONDS, false);
}
- AsyncRpcClient(RpcConnectionKey rpcConnectionKey, int retries, int idleTimeSeconds)
+ /**
+ * Intentionally make this method package-private, avoiding user directly
+ * new an instance through this constructor.
+ *
+ * @param rpcConnectionKey
+ * @param retries retry operation number of times
+ * @param timeout disable ping, it trigger timeout event on idle-state.
+ * otherwise it is request timeout on active-state
+ * @param timeUnit TimeUnit
+ * @param enablePing enable to detect remote peer hangs
+ * @throws ClassNotFoundException
+ * @throws NoSuchMethodException
+ */
+ AsyncRpcClient(RpcConnectionKey rpcConnectionKey, int retries, long timeout, TimeUnit timeUnit, boolean enablePing)
throws ClassNotFoundException, NoSuchMethodException {
super(rpcConnectionKey, retries);
- stubMethod = getServiceClass().getMethod("newStub", RpcChannel.class);
- rpcChannel = new ProxyRpcChannel();
- inboundHandler = new ClientChannelInboundHandler();
- init(new ProtoChannelInitializer(inboundHandler, RpcResponse.getDefaultInstance(), idleTimeSeconds));
+
+ this.stubMethod = getServiceClass().getMethod("newStub", RpcChannel.class);
+ this.rpcChannel = new ProxyRpcChannel();
+ this.handler = new ClientChannelInboundHandler();
+ init(new ProtoClientChannelInitializer(handler,
+ RpcResponse.getDefaultInstance(),
+ timeUnit.toNanos(timeout),
+ enablePing));
}
@Override
- public <T> T getStub() {
+ public <I> I getStub() {
return getStub(stubMethod, rpcChannel);
}
- protected void sendExceptions(String message) {
- for(Map.Entry<Integer, ResponseCallback> callbackEntry: requests.entrySet()) {
- ResponseCallback callback = callbackEntry.getValue();
- Integer id = callbackEntry.getKey();
-
- RpcResponse.Builder responseBuilder = RpcResponse.newBuilder()
- .setErrorMessage(message)
- .setId(id);
-
- callback.run(responseBuilder.build());
- }
- }
-
@Override
- public void close() {
- sendExceptions("AsyncRpcClient terminates all the connections");
-
- super.close();
+ protected NettyChannelInboundHandler getHandler() {
+ return handler;
}
private class ProxyRpcChannel implements RpcChannel {
+ private final AtomicInteger sequence = new AtomicInteger(0);
+
public void callMethod(final MethodDescriptor method,
final RpcController controller,
final Message param,
final Message responseType,
- RpcCallback<Message> done) {
+ final RpcCallback<Message> done) {
int nextSeqId = sequence.getAndIncrement();
+ RpcProtos.RpcRequest rpcRequest = buildRequest(nextSeqId, method, param);
- Message rpcRequest = buildRequest(nextSeqId, method, param);
-
- inboundHandler.registerCallback(nextSeqId,
- new ResponseCallback(controller, responseType, done));
-
- ChannelPromise channelPromise = getChannel().newPromise();
- channelPromise.addListener(new GenericFutureListener<ChannelFuture>() {
-
- @Override
- public void operationComplete(ChannelFuture future) throws Exception {
- if (!future.isSuccess()) {
- inboundHandler.exceptionCaught(null, new ServiceException(future.cause()));
- }
- }
- });
- getChannel().writeAndFlush(rpcRequest, channelPromise);
+ invoke(rpcRequest, new ResponseCallback(controller, responseType, done), 0);
}
+ }
- private Message buildRequest(int seqId,
- MethodDescriptor method,
- Message param) {
+ @ChannelHandler.Sharable
+ private class ClientChannelInboundHandler extends NettyChannelInboundHandler {
- RpcRequest.Builder requestBuilder = RpcRequest.newBuilder()
- .setId(seqId)
- .setMethodName(method.getName());
+ @Override
+ protected void run(RpcResponse response, ResponseCallback callback) throws Exception {
+ callback.run(response);
+ }
- if (param != null) {
- requestBuilder.setRequestMessage(param.toByteString());
- }
+ @Override
+ protected void handleException(int requestId, ResponseCallback callback, String message) {
+ RpcResponse.Builder responseBuilder = RpcResponse.newBuilder()
+ .setErrorMessage(message + "")
+ .setId(requestId);
- return requestBuilder.build();
+ callback.run(responseBuilder.build());
}
}
- private class ResponseCallback implements RpcCallback<RpcResponse> {
+ static class ResponseCallback implements RpcCallback<RpcResponse> {
private final RpcController controller;
private final Message responsePrototype;
private final RpcCallback<Message> callback;
@@ -149,88 +126,27 @@ public class AsyncRpcClient extends NettyClientBase {
@Override
public void run(RpcResponse rpcResponse) {
// if hasErrorMessage is true, it means rpc-level errors.
- // it does not call the callback function\
+ // it can be called the callback function with null response.
if (rpcResponse.hasErrorMessage()) {
if (controller != null) {
this.controller.setFailed(rpcResponse.getErrorMessage());
}
callback.run(null);
} else { // if rpc call succeed
- try {
- Message responseMessage;
- if (!rpcResponse.hasResponseMessage()) {
- responseMessage = null;
- } else {
+
+ Message responseMessage = null;
+ if (rpcResponse.hasResponseMessage()) {
+
+ try {
responseMessage = responsePrototype.newBuilderForType().mergeFrom(
rpcResponse.getResponseMessage()).build();
+ } catch (InvalidProtocolBufferException e) {
+ if (controller != null) {
+ this.controller.setFailed(e.getMessage());
+ }
}
-
- callback.run(responseMessage);
-
- } catch (InvalidProtocolBufferException e) {
- throw new RemoteException(getErrorMessage(""), e);
- }
- }
- }
- }
-
- private String getErrorMessage(String message) {
- return "Exception [" + protocol.getCanonicalName() +
- "(" + RpcUtils.normalizeInetSocketAddress((InetSocketAddress)
- getChannel().remoteAddress()) + ")]: " + message;
- }
-
- @ChannelHandler.Sharable
- private class ClientChannelInboundHandler extends SimpleChannelInboundHandler<RpcResponse> {
-
- void registerCallback(int seqId, ResponseCallback callback) {
-
- if (requests.putIfAbsent(seqId, callback) != null) {
- throw new RemoteException(
- getErrorMessage("Duplicate Sequence Id "+ seqId));
- }
- }
-
- @Override
- protected void channelRead0(ChannelHandlerContext ctx, RpcResponse response) throws Exception {
- ResponseCallback callback = requests.remove(response.getId());
-
- if (callback == null) {
- LOG.warn("Dangling rpc call");
- } else {
- callback.run(response);
- }
- }
-
- @Override
- public void channelActive(ChannelHandlerContext ctx) throws Exception {
- super.channelActive(ctx);
- LOG.info("Connection established successfully : " + ctx.channel().remoteAddress());
- }
-
- @Override
- public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
- throws Exception {
- LOG.error(getRemoteAddress() + "," + protocol + "," + cause.getMessage(), cause);
-
- sendExceptions(cause.getMessage());
-
- if(LOG.isDebugEnabled()) {
- LOG.error(cause.getMessage(), cause);
- } else {
- LOG.error("RPC Exception:" + cause.getMessage());
- }
- }
-
- @Override
- public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
- if (evt instanceof IdleStateEvent) {
- IdleStateEvent e = (IdleStateEvent) evt;
- /* If all requests is done and event is triggered, channel will be closed. */
- if (e.state() == IdleState.ALL_IDLE && requests.size() == 0) {
- ctx.close();
- LOG.warn("Idle connection closed successfully :" + ctx.channel().remoteAddress());
}
+ callback.run(responseMessage);
}
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/AsyncRpcServer.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/AsyncRpcServer.java b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/AsyncRpcServer.java
index e4109fe..22f47b0 100644
--- a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/AsyncRpcServer.java
+++ b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/AsyncRpcServer.java
@@ -24,6 +24,7 @@ import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;
import com.google.protobuf.Service;
import io.netty.channel.*;
+import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.tajo.rpc.RpcProtos.RpcRequest;
@@ -52,7 +53,7 @@ public class AsyncRpcServer extends NettyServerBase {
Method method = serviceClass.getMethod("newReflectiveService", interfaceClass);
this.service = (Service) method.invoke(null, instance);
- this.initializer = new ProtoChannelInitializer(new ServerHandler(), RpcRequest.getDefaultInstance());
+ this.initializer = new ProtoServerChannelInitializer(new ServerHandler(), RpcRequest.getDefaultInstance());
super.init(this.initializer, workerNum);
}
@@ -81,53 +82,67 @@ public class AsyncRpcServer extends NettyServerBase {
protected void channelRead0(final ChannelHandlerContext ctx, final RpcRequest request) throws Exception {
String methodName = request.getMethodName();
- MethodDescriptor methodDescriptor = service.getDescriptorForType().findMethodByName(methodName);
+ final MethodDescriptor methodDescriptor = service.getDescriptorForType().findMethodByName(methodName);
if (methodDescriptor == null) {
- throw new RemoteCallException(request.getId(), new NoSuchMethodException(methodName));
+ exceptionCaught(ctx, new RemoteCallException(request.getId(), new NoSuchMethodException(methodName)));
+ return;
}
- Message paramProto = null;
- if (request.hasRequestMessage()) {
- try {
+ try {
+ Message paramProto = null;
+ if (request.hasRequestMessage()) {
paramProto = service.getRequestPrototype(methodDescriptor).newBuilderForType()
.mergeFrom(request.getRequestMessage()).build();
- } catch (Throwable t) {
- throw new RemoteCallException(request.getId(), methodDescriptor, t);
}
- }
- final RpcController controller = new NettyRpcController();
+ final RpcController controller = new NettyRpcController();
- RpcCallback<Message> callback = !request.hasId() ? null : new RpcCallback<Message>() {
+ final RpcCallback<Message> callback = !request.hasId() ? null : new RpcCallback<Message>() {
- public void run(Message returnValue) {
+ public void run(Message returnValue) {
- RpcResponse.Builder builder = RpcResponse.newBuilder().setId(request.getId());
+ RpcResponse.Builder builder = RpcResponse.newBuilder().setId(request.getId());
- if (returnValue != null) {
- builder.setResponseMessage(returnValue.toByteString());
- }
+ if (returnValue != null) {
+ builder.setResponseMessage(returnValue.toByteString());
+ }
- if (controller.failed()) {
- builder.setErrorMessage(controller.errorText());
- }
+ if (controller.failed()) {
+ builder.setErrorMessage(controller.errorText());
+ }
- ctx.writeAndFlush(builder.build());
- }
- };
+ ctx.writeAndFlush(builder.build());
+ }
+ };
- service.callMethod(methodDescriptor, controller, paramProto, callback);
+ service.callMethod(methodDescriptor, controller, paramProto, callback);
+ } catch (RemoteCallException e) {
+ exceptionCaught(ctx, e);
+ } catch (Throwable throwable) {
+ exceptionCaught(ctx, new RemoteCallException(request.getId(), methodDescriptor, throwable));
+ }
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
- throws Exception{
+ throws Exception {
if (cause instanceof RemoteCallException) {
RemoteCallException callException = (RemoteCallException) cause;
ctx.writeAndFlush(callException.getResponse());
+
+ if(LOG.isDebugEnabled()) {
+ Throwable rootCause = ExceptionUtils.getRootCause(cause);
+ LOG.error(ExceptionUtils.getMessage(rootCause), rootCause);
+ }
} else {
- LOG.error(cause.getMessage());
+ /* unhandled exception. */
+ if (ctx.channel().isOpen()) {
+ /* client can be triggered channelInactiveEvent */
+ ctx.close();
+ }
+ Throwable rootCause = ExceptionUtils.getRootCause(cause);
+ LOG.fatal(ExceptionUtils.getMessage(rootCause), rootCause);
}
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/BlockingRpcClient.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/BlockingRpcClient.java b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/BlockingRpcClient.java
index ad536a4..349a0a0 100644
--- a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/BlockingRpcClient.java
+++ b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/BlockingRpcClient.java
@@ -18,71 +18,68 @@
package org.apache.tajo.rpc;
-import com.google.protobuf.BlockingRpcChannel;
+import com.google.protobuf.*;
import com.google.protobuf.Descriptors.MethodDescriptor;
-import com.google.protobuf.Message;
-import com.google.protobuf.RpcController;
-import com.google.protobuf.ServiceException;
-import io.netty.channel.*;
-import io.netty.handler.timeout.IdleState;
-import io.netty.handler.timeout.IdleStateEvent;
-import io.netty.util.concurrent.GenericFutureListener;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
+import io.netty.channel.ChannelHandler;
import org.apache.tajo.rpc.RpcClientManager.RpcConnectionKey;
-import org.apache.tajo.rpc.RpcProtos.RpcRequest;
import org.apache.tajo.rpc.RpcProtos.RpcResponse;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
-import java.util.Map;
import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
-public class BlockingRpcClient extends NettyClientBase {
- private static final Log LOG = LogFactory.getLog(RpcProtos.class);
-
- private final Map<Integer, ProtoCallFuture> requests =
- new ConcurrentHashMap<Integer, ProtoCallFuture>();
+public class BlockingRpcClient extends NettyClientBase<BlockingRpcClient.ProtoCallFuture> {
private final Method stubMethod;
private final ProxyRpcChannel rpcChannel;
- private final ChannelInboundHandlerAdapter inboundHandler;
+ private final NettyChannelInboundHandler handler;
- /**
- * Intentionally make this method package-private, avoiding user directly
- * new an instance through this constructor.
- */
BlockingRpcClient(RpcConnectionKey rpcConnectionKey, int retries)
throws NoSuchMethodException, ClassNotFoundException {
- this(rpcConnectionKey, retries, 0);
+ this(rpcConnectionKey, retries, 0, TimeUnit.NANOSECONDS, false);
}
- BlockingRpcClient(RpcConnectionKey rpcConnectionKey, int retries, int idleTimeSeconds)
- throws ClassNotFoundException, NoSuchMethodException {
+ /**
+ * Intentionally make this method package-private, avoiding user directly
+ * new an instance through this constructor.
+ *
+ * @param rpcConnectionKey
+ * @param retries retry operation number of times
+ * @param timeout disable ping, it trigger timeout event on idle-state.
+ * otherwise it is request timeout on active-state
+ * @param timeUnit TimeUnit
+ * @param enablePing enable to detect remote peer hangs
+ * @throws ClassNotFoundException
+ * @throws NoSuchMethodException
+ */
+ BlockingRpcClient(RpcConnectionKey rpcConnectionKey, int retries, long timeout, TimeUnit timeUnit,
+ boolean enablePing) throws ClassNotFoundException, NoSuchMethodException {
super(rpcConnectionKey, retries);
- stubMethod = getServiceClass().getMethod("newBlockingStub", BlockingRpcChannel.class);
- rpcChannel = new ProxyRpcChannel();
- inboundHandler = new ClientChannelInboundHandler();
- init(new ProtoChannelInitializer(inboundHandler, RpcResponse.getDefaultInstance(), idleTimeSeconds));
+
+ this.stubMethod = getServiceClass().getMethod("newBlockingStub", BlockingRpcChannel.class);
+ this.rpcChannel = new ProxyRpcChannel();
+ this.handler = new ClientChannelInboundHandler();
+ init(new ProtoClientChannelInitializer(handler,
+ RpcResponse.getDefaultInstance(),
+ timeUnit.toNanos(timeout),
+ enablePing));
}
@Override
- public <T> T getStub() {
+ public <I> I getStub() {
return getStub(stubMethod, rpcChannel);
}
@Override
- public void close() {
- for(ProtoCallFuture callback: requests.values()) {
- callback.setFailed("BlockingRpcClient terminates all the connections",
- new ServiceException("BlockingRpcClient terminates all the connections"));
- }
- requests.clear();
- super.close();
+ protected NettyChannelInboundHandler getHandler() {
+ return handler;
}
private class ProxyRpcChannel implements BlockingRpcChannel {
+ private final AtomicInteger sequence = new AtomicInteger(0);
+
@Override
public Message callBlockingMethod(final MethodDescriptor method,
final RpcController controller,
@@ -91,136 +88,66 @@ public class BlockingRpcClient extends NettyClientBase {
throws TajoServiceException {
int nextSeqId = sequence.getAndIncrement();
+ RpcProtos.RpcRequest rpcRequest = buildRequest(nextSeqId, method, param);
+ ProtoCallFuture callFuture = new ProtoCallFuture(controller, responsePrototype);
- Message rpcRequest = buildRequest(nextSeqId, method, param);
-
- ProtoCallFuture callFuture =
- new ProtoCallFuture(controller, responsePrototype);
- requests.put(nextSeqId, callFuture);
-
- ChannelPromise channelPromise = getChannel().newPromise();
- channelPromise.addListener(new GenericFutureListener<ChannelFuture>() {
- @Override
- public void operationComplete(ChannelFuture future) throws Exception {
- if (!future.isSuccess()) {
- inboundHandler.exceptionCaught(null, new ServiceException(future.cause()));
- }
- }
- });
- getChannel().writeAndFlush(rpcRequest, channelPromise);
+ invoke(rpcRequest, callFuture, 0);
try {
- return callFuture.get(60, TimeUnit.SECONDS);
+ return callFuture.get();
} catch (Throwable t) {
if (t instanceof ExecutionException) {
Throwable cause = t.getCause();
if (cause != null && cause instanceof TajoServiceException) {
- throw (TajoServiceException)cause;
+ throw (TajoServiceException) cause;
}
}
throw new TajoServiceException(t.getMessage());
}
}
-
- private Message buildRequest(int seqId,
- MethodDescriptor method,
- Message param) {
- RpcRequest.Builder requestBuilder = RpcRequest.newBuilder()
- .setId(seqId)
- .setMethodName(method.getName());
-
- if (param != null) {
- requestBuilder.setRequestMessage(param.toByteString());
- }
-
- return requestBuilder.build();
- }
- }
-
- private String getErrorMessage(String message) {
- if(getChannel() != null) {
- return protocol.getName() +
- "(" + RpcUtils.normalizeInetSocketAddress((InetSocketAddress)
- getChannel().remoteAddress()) + "): " + message;
- } else {
- return "Exception " + message;
- }
}
private TajoServiceException makeTajoServiceException(RpcResponse response, Throwable cause) {
- if(getChannel() != null) {
- return new TajoServiceException(response.getErrorMessage(), cause, protocol.getName(),
- RpcUtils.normalizeInetSocketAddress((InetSocketAddress)getChannel().remoteAddress()));
+ if (getChannel() != null) {
+ return new TajoServiceException(response.getErrorMessage(), cause, getKey().protocolClass.getName(),
+ RpcUtils.normalizeInetSocketAddress((InetSocketAddress) getChannel().remoteAddress()));
} else {
return new TajoServiceException(response.getErrorMessage());
}
}
@ChannelHandler.Sharable
- private class ClientChannelInboundHandler extends SimpleChannelInboundHandler<RpcResponse> {
+ public class ClientChannelInboundHandler extends NettyChannelInboundHandler {
@Override
- protected void channelRead0(ChannelHandlerContext ctx, RpcResponse rpcResponse) throws Exception {
- ProtoCallFuture callback = requests.remove(rpcResponse.getId());
-
- if (callback == null) {
- LOG.warn("Dangling rpc call");
+ protected void run(RpcResponse rpcResponse, ProtoCallFuture callback) throws Exception {
+ if (rpcResponse.hasErrorMessage()) {
+ callback.setFailed(rpcResponse.getErrorMessage(),
+ makeTajoServiceException(rpcResponse, new ServiceException(rpcResponse.getErrorTrace())));
} else {
- if (rpcResponse.hasErrorMessage()) {
- callback.setFailed(rpcResponse.getErrorMessage(),
- makeTajoServiceException(rpcResponse, new ServiceException(rpcResponse.getErrorTrace())));
- } else {
- Message responseMessage;
-
- if (!rpcResponse.hasResponseMessage()) {
- responseMessage = null;
- } else {
+ Message responseMessage = null;
+
+ if (rpcResponse.hasResponseMessage()) {
+ try {
responseMessage = callback.returnType.newBuilderForType().mergeFrom(rpcResponse.getResponseMessage())
.build();
+ } catch (InvalidProtocolBufferException e) {
+ callback.setFailed(e.getMessage(), e);
}
-
- callback.setResponse(responseMessage);
}
+ callback.setResponse(responseMessage);
}
}
@Override
- public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
- throws Exception {
- /* Current requests will be failed */
- for(ProtoCallFuture callback: requests.values()) {
- callback.setFailed(cause.getMessage(), cause);
- }
- requests.clear();
-
- if(LOG.isDebugEnabled()) {
- LOG.error("" + cause.getMessage(), cause);
- } else {
- LOG.error("RPC Exception:" + cause.getMessage());
- }
- }
-
- @Override
- public void channelActive(ChannelHandlerContext ctx) throws Exception {
- super.channelActive(ctx);
- LOG.info("Connection established successfully : " + ctx.channel().remoteAddress());
- }
-
- @Override
- public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
- if (evt instanceof IdleStateEvent) {
- IdleStateEvent e = (IdleStateEvent) evt;
- /* If all requests is done and event is triggered, channel will be closed. */
- if (e.state() == IdleState.ALL_IDLE && requests.size() == 0) {
- ctx.close();
- LOG.warn("Idle connection closed successfully :" + ctx.channel().remoteAddress());
- }
- }
+ protected void handleException(int requestId, ProtoCallFuture callback, String message) {
+ callback.setFailed(message + "", new TajoServiceException(message));
}
}
- static class ProtoCallFuture implements Future<Message> {
+ static class ProtoCallFuture implements Future<Message> {
private Semaphore sem = new Semaphore(0);
+ private boolean done = false;
private Message response = null;
private Message returnType;
@@ -240,8 +167,9 @@ public class BlockingRpcClient extends NettyClientBase {
@Override
public Message get() throws InterruptedException, ExecutionException {
- sem.acquire();
- if(ee != null) {
+ if(!isDone()) sem.acquire();
+
+ if (ee != null) {
throw ee;
}
return response;
@@ -250,14 +178,16 @@ public class BlockingRpcClient extends NettyClientBase {
@Override
public Message get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
- if(sem.tryAcquire(timeout, unit)) {
- if (ee != null) {
- throw ee;
+ if(!isDone()) {
+ if (!sem.tryAcquire(timeout, unit)) {
+ throw new TimeoutException();
}
- return response;
- } else {
- throw new TimeoutException();
}
+
+ if (ee != null) {
+ throw ee;
+ }
+ return response;
}
@Override
@@ -267,19 +197,21 @@ public class BlockingRpcClient extends NettyClientBase {
@Override
public boolean isDone() {
- return sem.availablePermits() > 0;
+ return done;
}
public void setResponse(Message response) {
this.response = response;
+ done = true;
sem.release();
}
public void setFailed(String errorText, Throwable t) {
- if(controller != null) {
+ if (controller != null) {
this.controller.setFailed(errorText);
}
ee = new ExecutionException(errorText, t);
+ done = true;
sem.release();
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/BlockingRpcServer.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/BlockingRpcServer.java b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/BlockingRpcServer.java
index bb31367..93c28e3 100644
--- a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/BlockingRpcServer.java
+++ b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/BlockingRpcServer.java
@@ -23,10 +23,10 @@ import com.google.protobuf.Descriptors.MethodDescriptor;
import com.google.protobuf.Message;
import com.google.protobuf.RpcController;
import io.netty.channel.*;
+import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.tajo.rpc.RpcProtos.RpcRequest;
-import org.apache.tajo.rpc.RpcProtos.RpcResponse;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
@@ -53,7 +53,7 @@ public class BlockingRpcServer extends NettyServerBase {
"newReflectiveBlockingService", interfaceClass);
this.service = (BlockingService) method.invoke(null, instance);
- this.initializer = new ProtoChannelInitializer(new ServerHandler(), RpcRequest.getDefaultInstance());
+ this.initializer = new ProtoServerChannelInitializer(new ServerHandler(), RpcRequest.getDefaultInstance());
super.init(this.initializer, workerNum);
}
@@ -80,43 +80,40 @@ public class BlockingRpcServer extends NettyServerBase {
}
@Override
- protected void channelRead0(ChannelHandlerContext ctx, RpcRequest request) throws Exception {
+ protected void channelRead0(final ChannelHandlerContext ctx, final RpcRequest request) throws Exception {
String methodName = request.getMethodName();
- MethodDescriptor methodDescriptor = service.getDescriptorForType().findMethodByName(methodName);
-
+ final MethodDescriptor methodDescriptor = service.getDescriptorForType().findMethodByName(methodName);
if (methodDescriptor == null) {
- throw new RemoteCallException(request.getId(), new NoSuchMethodException(methodName));
+ exceptionCaught(ctx, new RemoteCallException(request.getId(), new NoSuchMethodException(methodName)));
+ return;
}
- Message paramProto = null;
- if (request.hasRequestMessage()) {
- try {
+
+ try {
+ Message paramProto = null;
+ if (request.hasRequestMessage()) {
paramProto = service.getRequestPrototype(methodDescriptor).newBuilderForType()
.mergeFrom(request.getRequestMessage()).build();
-
- } catch (Throwable t) {
- throw new RemoteCallException(request.getId(), methodDescriptor, t);
}
- }
- Message returnValue;
- RpcController controller = new NettyRpcController();
- try {
- returnValue = service.callBlockingMethod(methodDescriptor, controller, paramProto);
- } catch (Throwable t) {
- throw new RemoteCallException(request.getId(), methodDescriptor, t);
- }
+ RpcController controller = new NettyRpcController();
+ Message returnValue = service.callBlockingMethod(methodDescriptor, controller, paramProto);
- RpcResponse.Builder builder = RpcResponse.newBuilder().setId(request.getId());
+ RpcProtos.RpcResponse.Builder builder = RpcProtos.RpcResponse.newBuilder().setId(request.getId());
- if (returnValue != null) {
- builder.setResponseMessage(returnValue.toByteString());
- }
+ if (returnValue != null) {
+ builder.setResponseMessage(returnValue.toByteString());
+ }
- if (controller.failed()) {
- builder.setErrorMessage(controller.errorText());
+ if (controller.failed()) {
+ builder.setErrorMessage(controller.errorText());
+ }
+ ctx.writeAndFlush(builder.build());
+ } catch (RemoteCallException e) {
+ exceptionCaught(ctx, e);
+ } catch (Throwable throwable) {
+ exceptionCaught(ctx, new RemoteCallException(request.getId(), methodDescriptor, throwable));
}
- ctx.writeAndFlush(builder.build());
}
@Override
@@ -124,6 +121,19 @@ public class BlockingRpcServer extends NettyServerBase {
if (cause instanceof RemoteCallException) {
RemoteCallException callException = (RemoteCallException) cause;
ctx.writeAndFlush(callException.getResponse());
+
+ if(LOG.isDebugEnabled()) {
+ Throwable rootCause = ExceptionUtils.getRootCause(cause);
+ LOG.debug(ExceptionUtils.getMessage(rootCause), rootCause);
+ }
+ } else {
+ /* unhandled exception. */
+ if (ctx.channel().isOpen()) {
+ /* client can be triggered channelInactiveEvent */
+ ctx.close();
+ }
+ Throwable rootCause = ExceptionUtils.getRootCause(cause);
+ LOG.fatal(ExceptionUtils.getMessage(rootCause), rootCause);
}
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/CallFuture.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/CallFuture.java b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/CallFuture.java
index c4c3256..dee0ff4 100644
--- a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/CallFuture.java
+++ b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/CallFuture.java
@@ -20,11 +20,9 @@ package org.apache.tajo.rpc;
import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;
+import com.google.protobuf.ServiceException;
-import java.util.concurrent.Future;
-import java.util.concurrent.Semaphore;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
+import java.util.concurrent.*;
public class CallFuture<T> implements RpcCallback<T>, Future<T> {
@@ -66,19 +64,29 @@ public class CallFuture<T> implements RpcCallback<T>, Future<T> {
}
@Override
- public T get() throws InterruptedException {
- sem.acquire();
+ public T get() throws InterruptedException, ExecutionException {
+ if (!isDone())
+ sem.acquire();
+ throwIfFailed();
return response;
}
@Override
- public T get(long timeout, TimeUnit unit)
- throws InterruptedException, TimeoutException {
- if (sem.tryAcquire(timeout, unit)) {
- return response;
- } else {
- throw new TimeoutException();
+ public T get(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException, ExecutionException {
+ if (!isDone()) {
+ if (!sem.tryAcquire(timeout, unit)) {
+ throw new TimeoutException();
+ }
+ }
+
+ throwIfFailed();
+ return response;
+ }
+
+ private void throwIfFailed() throws ExecutionException {
+ if (controller.failed()) {
+ throw new ExecutionException(new ServiceException(controller.errorText()));
}
}
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ChannelEventListener.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ChannelEventListener.java b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ChannelEventListener.java
new file mode 100644
index 0000000..e0f2c0d
--- /dev/null
+++ b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ChannelEventListener.java
@@ -0,0 +1,39 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.rpc;
+
+import io.netty.channel.ChannelHandlerContext;
+
+/**
+ * Event listener for netty code. Users can subscribe events by using this interface.
+ */
+public interface ChannelEventListener {
+
+ /**
+ * Performs actions before channel open.
+ * @param ctx
+ */
+ void channelRegistered(ChannelHandlerContext ctx);
+
+ /**
+ * Performs actions after channel close
+ * @param ctx
+ */
+ void channelUnregistered(ChannelHandlerContext ctx);
+}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ConnectionCloseFutureListener.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ConnectionCloseFutureListener.java b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ConnectionCloseFutureListener.java
deleted file mode 100644
index 29c9772..0000000
--- a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/ConnectionCloseFutureListener.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.tajo.rpc;
-
-import io.netty.util.concurrent.Future;
-import io.netty.util.concurrent.GenericFutureListener;
-
-public class ConnectionCloseFutureListener implements GenericFutureListener {
- private RpcClientManager.RpcConnectionKey key;
-
- public ConnectionCloseFutureListener(RpcClientManager.RpcConnectionKey key) {
- this.key = key;
- }
-
- @Override
- public void operationComplete(Future future) throws Exception {
- RpcClientManager.remove(key);
- }
-}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/DefaultRpcController.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/DefaultRpcController.java b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/DefaultRpcController.java
index 4ba19a5..47125e4 100644
--- a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/DefaultRpcController.java
+++ b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/DefaultRpcController.java
@@ -28,7 +28,7 @@ public class DefaultRpcController implements RpcController {
@Override
public void reset() {
- errorText = "";
+ errorText = null;
error = false;
canceled = false;
}
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/MonitorClientHandler.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/MonitorClientHandler.java b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/MonitorClientHandler.java
new file mode 100644
index 0000000..441c78a
--- /dev/null
+++ b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/MonitorClientHandler.java
@@ -0,0 +1,101 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.rpc;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.timeout.IdleState;
+import io.netty.handler.timeout.IdleStateEvent;
+import io.netty.handler.timeout.IdleStateHandler;
+import io.netty.util.ReferenceCountUtil;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import java.nio.charset.Charset;
+
+/**
+ * MonitorClientHandler is a packet sender for detecting server hangs
+ * Triggers an {@link MonitorStateEvent} when a remote peer has not respond.
+ */
+
+public class MonitorClientHandler extends ChannelInboundHandlerAdapter {
+ private static final Log LOG = LogFactory.getLog(MonitorClientHandler.class);
+
+ private ByteBuf ping;
+ private boolean enableMonitor;
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx) throws Exception {
+ // Initialize the message.
+ ping = ctx.alloc().buffer(RpcConstants.PING_PACKET.length())
+ .writeBytes(RpcConstants.PING_PACKET.getBytes(Charset.defaultCharset()));
+ IdleStateHandler handler = ctx.pipeline().get(IdleStateHandler.class);
+ if(handler != null && handler.getWriterIdleTimeInMillis() > 0) {
+ enableMonitor = true;
+ }
+ super.channelActive(ctx);
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ ping.release();
+ super.channelInactive(ctx);
+ }
+
+ private boolean isPing(Object msg) {
+ if(msg instanceof ByteBuf){
+ return ByteBufUtil.equals(ping.duplicate(), ((ByteBuf)msg).duplicate());
+ }
+ return false;
+ }
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+ if(enableMonitor && isPing(msg)){
+ //ignore ping response
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("received ping " + ctx.channel());
+ }
+ ReferenceCountUtil.release(msg);
+ } else {
+ super.channelRead(ctx, msg);
+ }
+ }
+
+ @Override
+ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+ if (enableMonitor && evt instanceof IdleStateEvent) {
+ IdleStateEvent e = (IdleStateEvent) evt;
+ if (e.state() == IdleState.READER_IDLE && !e.isFirst()) {
+ /* trigger expired event */
+ LOG.info("Server has not respond " + ctx.channel());
+ ctx.fireUserEventTriggered(MonitorStateEvent.MONITOR_EXPIRED_STATE_EVENT);
+ } else if (e.state() == IdleState.WRITER_IDLE) {
+ /* send ping packet to remote server */
+ if(LOG.isDebugEnabled()){
+ LOG.debug("sending ping request " + ctx.channel());
+ }
+ ctx.writeAndFlush(ping.duplicate().retain());
+ }
+ }
+ super.userEventTriggered(ctx, evt);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/MonitorServerHandler.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/MonitorServerHandler.java b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/MonitorServerHandler.java
new file mode 100644
index 0000000..e4333f4
--- /dev/null
+++ b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/MonitorServerHandler.java
@@ -0,0 +1,73 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.rpc;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import java.nio.charset.Charset;
+
+/**
+ * MonitorServerHandler is a packet receiver for detecting server hangs
+ * Reply response when a remote peer sent a ping packet.
+ */
+
+public class MonitorServerHandler extends ChannelInboundHandlerAdapter {
+ private static final Log LOG = LogFactory.getLog(MonitorServerHandler.class);
+
+ private ByteBuf ping;
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx) throws Exception {
+ // Initialize the message.
+ ping = ctx.alloc().directBuffer(4).writeBytes(RpcConstants.PING_PACKET.getBytes(Charset.defaultCharset()));
+ super.channelActive(ctx);
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ ping.release();
+ super.channelInactive(ctx);
+ }
+
+ private boolean isPing(Object msg) {
+ if (msg instanceof ByteBuf) {
+ return ByteBufUtil.equals(ping.duplicate(), ((ByteBuf) msg).duplicate());
+ }
+
+ return false;
+ }
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+ if (isPing(msg)) {
+ /* reply to client */
+ if(LOG.isDebugEnabled()){
+ LOG.debug("reply to " + ctx.channel());
+ }
+ ctx.writeAndFlush(msg);
+ } else {
+ super.channelRead(ctx, msg);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tajo/blob/2e7d03df/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/MonitorStateEvent.java
----------------------------------------------------------------------
diff --git a/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/MonitorStateEvent.java b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/MonitorStateEvent.java
new file mode 100644
index 0000000..294be3b
--- /dev/null
+++ b/tajo-rpc/tajo-rpc-protobuf/src/main/java/org/apache/tajo/rpc/MonitorStateEvent.java
@@ -0,0 +1,49 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tajo.rpc;
+
+/**
+ * A user event triggered by {@link org.apache.tajo.rpc.MonitorStateEvent} when a ping packet is not receive.
+ */
+public final class MonitorStateEvent {
+ /**
+ * An {@link Enum} that represents the monitor state of a remote peer.
+ */
+ public enum MonitorState {
+ /**
+ * No ping was received.
+ */
+ PING_EXPIRED
+ }
+
+ public static final MonitorStateEvent MONITOR_EXPIRED_STATE_EVENT = new MonitorStateEvent(MonitorState.PING_EXPIRED);
+
+ private final MonitorState state;
+
+ private MonitorStateEvent(MonitorState state) {
+ this.state = state;
+ }
+
+ /**
+ * Returns the monitor state.
+ */
+ public MonitorState state() {
+ return state;
+ }
+}
\ No newline at end of file
[3/4] tajo git commit: TAJO-1587: Upgrade java version to 1.7 for
Travis CI. (jihoon)
Posted by ji...@apache.org.
TAJO-1587: Upgrade java version to 1.7 for Travis CI. (jihoon)
Closes #557
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/9d73267c
Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/9d73267c
Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/9d73267c
Branch: refs/heads/index_support
Commit: 9d73267cb308e1d4c2df60e1099322938f0559a5
Parents: 2e7d03d
Author: Jihoon Son <ji...@apache.org>
Authored: Wed Apr 29 10:36:37 2015 +0900
Committer: Jihoon Son <ji...@apache.org>
Committed: Wed Apr 29 10:37:04 2015 +0900
----------------------------------------------------------------------
.travis.yml | 2 +-
CHANGES | 2 ++
2 files changed, 3 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tajo/blob/9d73267c/.travis.yml
----------------------------------------------------------------------
diff --git a/.travis.yml b/.travis.yml
index 4b3459a..8fc9b94 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -19,7 +19,7 @@ git:
depth: 150
jdk:
- - openjdk6
+ - openjdk7
env: PATH=$PATH:$HOME/local/bin
http://git-wip-us.apache.org/repos/asf/tajo/blob/9d73267c/CHANGES
----------------------------------------------------------------------
diff --git a/CHANGES b/CHANGES
index 93fb092..c7a8edd 100644
--- a/CHANGES
+++ b/CHANGES
@@ -212,6 +212,8 @@ Release 0.11.0 - unreleased
TASKS
+ TAJO-1587: Upgrade java version to 1.7 for Travis CI. (jihoon)
+
TAJO-1559: Fix data model description (tinyint, smallint).
(Contributed by Dongjoon Hyun, Committed by jihoon)