http://git-wip-us.apache.org/repos/asf/flink/blob/f48f5340/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/Client.java ---------------------------------------------------------------------- diff --git a/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/Client.java b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/Client.java new file mode 100644 index 0000000..e6d59de --- /dev/null +++ b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/Client.java @@ -0,0 +1,537 @@ +/* + * 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.flink.queryablestate.network; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.queryablestate.network.messages.MessageBody; +import org.apache.flink.queryablestate.network.messages.MessageSerializer; +import org.apache.flink.runtime.concurrent.FutureUtils; +import org.apache.flink.runtime.io.network.netty.NettyBufferPool; +import org.apache.flink.runtime.query.KvStateServerAddress; +import org.apache.flink.runtime.query.netty.KvStateRequestStats; +import org.apache.flink.util.Preconditions; + +import org.apache.flink.shaded.guava18.com.google.common.util.concurrent.ThreadFactoryBuilder; +import org.apache.flink.shaded.netty4.io.netty.bootstrap.Bootstrap; +import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf; +import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBufAllocator; +import org.apache.flink.shaded.netty4.io.netty.channel.Channel; +import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFuture; +import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFutureListener; +import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInitializer; +import org.apache.flink.shaded.netty4.io.netty.channel.ChannelOption; +import org.apache.flink.shaded.netty4.io.netty.channel.EventLoopGroup; +import org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoopGroup; +import org.apache.flink.shaded.netty4.io.netty.channel.socket.SocketChannel; +import org.apache.flink.shaded.netty4.io.netty.channel.socket.nio.NioSocketChannel; +import org.apache.flink.shaded.netty4.io.netty.handler.codec.LengthFieldBasedFrameDecoder; +import org.apache.flink.shaded.netty4.io.netty.handler.stream.ChunkedWriteHandler; + +import java.nio.channels.ClosedChannelException; +import java.util.ArrayDeque; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; + +/** + * The base class for every client in the queryable state module. + * It is using pure netty to send and receive messages of type {@link MessageBody}. + * + * @param <REQ> the type of request the client will send. + * @param <RESP> the type of response the client expects to receive. + */ +@Internal +public class Client<REQ extends MessageBody, RESP extends MessageBody> { + + /** The name of the client. Used for logging and stack traces.*/ + private final String clientName; + + /** Netty's Bootstrap. */ + private final Bootstrap bootstrap; + + /** The serializer to be used for (de-)serializing messages. */ + private final MessageSerializer<REQ, RESP> messageSerializer; + + /** Statistics tracker. */ + private final KvStateRequestStats stats; + + /** Established connections. */ + private final Map<KvStateServerAddress, EstablishedConnection> establishedConnections = new ConcurrentHashMap<>(); + + /** Pending connections. */ + private final Map<KvStateServerAddress, PendingConnection> pendingConnections = new ConcurrentHashMap<>(); + + /** Atomic shut down flag. */ + private final AtomicBoolean shutDown = new AtomicBoolean(); + + /** + * Creates a client with the specified number of event loop threads. + * + * @param clientName the name of the client. + * @param numEventLoopThreads number of event loop threads (minimum 1). + * @param serializer the serializer used to (de-)serialize messages. + * @param stats the statistics collector. + */ + public Client( + final String clientName, + final int numEventLoopThreads, + final MessageSerializer<REQ, RESP> serializer, + final KvStateRequestStats stats) { + + Preconditions.checkArgument(numEventLoopThreads >= 1, + "Non-positive number of event loop threads."); + + this.clientName = Preconditions.checkNotNull(clientName); + this.messageSerializer = Preconditions.checkNotNull(serializer); + this.stats = Preconditions.checkNotNull(stats); + + final ThreadFactory threadFactory = new ThreadFactoryBuilder() + .setDaemon(true) + .setNameFormat("Flink " + clientName + " Event Loop Thread %d") + .build(); + + final EventLoopGroup nioGroup = new NioEventLoopGroup(numEventLoopThreads, threadFactory); + final ByteBufAllocator bufferPool = new NettyBufferPool(numEventLoopThreads); + + this.bootstrap = new Bootstrap() + .group(nioGroup) + .channel(NioSocketChannel.class) + .option(ChannelOption.ALLOCATOR, bufferPool) + .handler(new ChannelInitializer<SocketChannel>() { + @Override + protected void initChannel(SocketChannel channel) throws Exception { + channel.pipeline() + .addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4)) + .addLast(new ChunkedWriteHandler()); + } + }); + } + + public String getClientName() { + return clientName; + } + + public CompletableFuture<RESP> sendRequest(final KvStateServerAddress serverAddress, final REQ request) { + if (shutDown.get()) { + return FutureUtils.getFailedFuture(new IllegalStateException("Shut down")); + } + + EstablishedConnection connection = establishedConnections.get(serverAddress); + if (connection != null) { + return connection.sendRequest(request); + } else { + PendingConnection pendingConnection = pendingConnections.get(serverAddress); + if (pendingConnection != null) { + // There was a race, use the existing pending connection. + return pendingConnection.sendRequest(request); + } else { + // We try to connect to the server. + PendingConnection pending = new PendingConnection(serverAddress, messageSerializer); + PendingConnection previous = pendingConnections.putIfAbsent(serverAddress, pending); + + if (previous == null) { + // OK, we are responsible to connect. + bootstrap.connect(serverAddress.getHost(), serverAddress.getPort()).addListener(pending); + return pending.sendRequest(request); + } else { + // There was a race, use the existing pending connection. + return previous.sendRequest(request); + } + } + } + } + + /** + * Shuts down the client and closes all connections. + * + * <p>After a call to this method, all returned futures will be failed. + */ + public void shutdown() { + if (shutDown.compareAndSet(false, true)) { + for (Map.Entry<KvStateServerAddress, EstablishedConnection> conn : establishedConnections.entrySet()) { + if (establishedConnections.remove(conn.getKey(), conn.getValue())) { + conn.getValue().close(); + } + } + + for (Map.Entry<KvStateServerAddress, PendingConnection> conn : pendingConnections.entrySet()) { + if (pendingConnections.remove(conn.getKey()) != null) { + conn.getValue().close(); + } + } + + if (bootstrap != null) { + EventLoopGroup group = bootstrap.group(); + if (group != null) { + group.shutdownGracefully(0L, 10L, TimeUnit.SECONDS); + } + } + } + } + + /** + * A pending connection that is in the process of connecting. + */ + private class PendingConnection implements ChannelFutureListener { + + /** Lock to guard the connect call, channel hand in, etc. */ + private final Object connectLock = new Object(); + + /** Address of the server we are connecting to. */ + private final KvStateServerAddress serverAddress; + + private final MessageSerializer<REQ, RESP> serializer; + + /** Queue of requests while connecting. */ + private final ArrayDeque<PendingRequest> queuedRequests = new ArrayDeque<>(); + + /** The established connection after the connect succeeds. */ + private EstablishedConnection established; + + /** Closed flag. */ + private boolean closed; + + /** Failure cause if something goes wrong. */ + private Throwable failureCause; + + /** + * Creates a pending connection to the given server. + * + * @param serverAddress Address of the server to connect to. + */ + private PendingConnection( + final KvStateServerAddress serverAddress, + final MessageSerializer<REQ, RESP> serializer) { + this.serverAddress = serverAddress; + this.serializer = serializer; + } + + @Override + public void operationComplete(ChannelFuture future) throws Exception { + if (future.isSuccess()) { + handInChannel(future.channel()); + } else { + close(future.cause()); + } + } + + /** + * Returns a future holding the serialized request result. + * + * <p>If the channel has been established, forward the call to the + * established channel, otherwise queue it for when the channel is + * handed in. + * + * @param request the request to be sent. + * @return Future holding the serialized result + */ + public CompletableFuture<RESP> sendRequest(REQ request) { + synchronized (connectLock) { + if (failureCause != null) { + return FutureUtils.getFailedFuture(failureCause); + } else if (closed) { + return FutureUtils.getFailedFuture(new ClosedChannelException()); + } else { + if (established != null) { + return established.sendRequest(request); + } else { + // Queue this and handle when connected + final PendingRequest pending = new PendingRequest(request); + queuedRequests.add(pending); + return pending; + } + } + } + } + + /** + * Hands in a channel after a successful connection. + * + * @param channel Channel to hand in + */ + private void handInChannel(Channel channel) { + synchronized (connectLock) { + if (closed || failureCause != null) { + // Close the channel and we are done. Any queued requests + // are removed on the close/failure call and after that no + // new ones can be enqueued. + channel.close(); + } else { + established = new EstablishedConnection(serverAddress, serializer, channel); + + while (!queuedRequests.isEmpty()) { + final PendingRequest pending = queuedRequests.poll(); + + established.sendRequest(pending.request) + .thenAccept(resp -> pending.complete(resp)) + .exceptionally(throwable -> { + pending.completeExceptionally(throwable); + return null; + }); + } + + // Publish the channel for the general public + establishedConnections.put(serverAddress, established); + pendingConnections.remove(serverAddress); + + // Check shut down for possible race with shut down. We + // don't want any lingering connections after shut down, + // which can happen if we don't check this here. + if (shutDown.get()) { + if (establishedConnections.remove(serverAddress, established)) { + established.close(); + } + } + } + } + } + + /** + * Close the connecting channel with a ClosedChannelException. + */ + private void close() { + close(new ClosedChannelException()); + } + + /** + * Close the connecting channel with an Exception (can be {@code null}) + * or forward to the established channel. + */ + private void close(Throwable cause) { + synchronized (connectLock) { + if (!closed) { + if (failureCause == null) { + failureCause = cause; + } + + if (established != null) { + established.close(); + } else { + PendingRequest pending; + while ((pending = queuedRequests.poll()) != null) { + pending.completeExceptionally(cause); + } + } + closed = true; + } + } + } + + @Override + public String toString() { + synchronized (connectLock) { + return "PendingConnection{" + + "serverAddress=" + serverAddress + + ", queuedRequests=" + queuedRequests.size() + + ", established=" + (established != null) + + ", closed=" + closed + + '}'; + } + } + + /** + * A pending request queued while the channel is connecting. + */ + private final class PendingRequest extends CompletableFuture<RESP> { + + private final REQ request; + + private PendingRequest(REQ request) { + this.request = request; + } + } + } + + /** + * An established connection that wraps the actual channel instance and is + * registered at the {@link ClientHandler} for callbacks. + */ + private class EstablishedConnection implements ClientHandlerCallback<RESP> { + + /** Address of the server we are connected to. */ + private final KvStateServerAddress serverAddress; + + /** The actual TCP channel. */ + private final Channel channel; + + /** Pending requests keyed by request ID. */ + private final ConcurrentHashMap<Long, TimestampedCompletableFuture> pendingRequests = new ConcurrentHashMap<>(); + + /** Current request number used to assign unique request IDs. */ + private final AtomicLong requestCount = new AtomicLong(); + + /** Reference to a failure that was reported by the channel. */ + private final AtomicReference<Throwable> failureCause = new AtomicReference<>(); + + /** + * Creates an established connection with the given channel. + * + * @param serverAddress Address of the server connected to + * @param channel The actual TCP channel + */ + EstablishedConnection( + final KvStateServerAddress serverAddress, + final MessageSerializer<REQ, RESP> serializer, + final Channel channel) { + + this.serverAddress = Preconditions.checkNotNull(serverAddress); + this.channel = Preconditions.checkNotNull(channel); + + // Add the client handler with the callback + channel.pipeline().addLast( + getClientName() + " Handler", + new ClientHandler<>(clientName, serializer, this) + ); + + stats.reportActiveConnection(); + } + + /** + * Close the channel with a ClosedChannelException. + */ + void close() { + close(new ClosedChannelException()); + } + + /** + * Close the channel with a cause. + * + * @param cause The cause to close the channel with. + * @return Channel close future + */ + private boolean close(Throwable cause) { + if (failureCause.compareAndSet(null, cause)) { + channel.close(); + stats.reportInactiveConnection(); + + for (long requestId : pendingRequests.keySet()) { + TimestampedCompletableFuture pending = pendingRequests.remove(requestId); + if (pending != null && pending.completeExceptionally(cause)) { + stats.reportFailedRequest(); + } + } + return true; + } + return false; + } + + /** + * Returns a future holding the serialized request result. + * @param request the request to be sent. + * @return Future holding the serialized result + */ + CompletableFuture<RESP> sendRequest(REQ request) { + TimestampedCompletableFuture requestPromiseTs = + new TimestampedCompletableFuture(System.nanoTime()); + try { + final long requestId = requestCount.getAndIncrement(); + pendingRequests.put(requestId, requestPromiseTs); + + stats.reportRequest(); + + ByteBuf buf = MessageSerializer.serializeRequest(channel.alloc(), requestId, request); + + channel.writeAndFlush(buf).addListener((ChannelFutureListener) future -> { + if (!future.isSuccess()) { + // Fail promise if not failed to write + TimestampedCompletableFuture pending = pendingRequests.remove(requestId); + if (pending != null && pending.completeExceptionally(future.cause())) { + stats.reportFailedRequest(); + } + } + }); + + // Check failure for possible race. We don't want any lingering + // promises after a failure, which can happen if we don't check + // this here. Note that close is treated as a failure as well. + Throwable failure = failureCause.get(); + if (failure != null) { + // Remove from pending requests to guard against concurrent + // removal and to make sure that we only count it once as failed. + TimestampedCompletableFuture pending = pendingRequests.remove(requestId); + if (pending != null && pending.completeExceptionally(failure)) { + stats.reportFailedRequest(); + } + } + } catch (Throwable t) { + requestPromiseTs.completeExceptionally(t); + } + + return requestPromiseTs; + } + + @Override + public void onRequestResult(long requestId, RESP response) { + TimestampedCompletableFuture pending = pendingRequests.remove(requestId); + if (pending != null && pending.complete(response)) { + long durationMillis = (System.nanoTime() - pending.getTimestamp()) / 1_000_000L; + stats.reportSuccessfulRequest(durationMillis); + } + } + + @Override + public void onRequestFailure(long requestId, Throwable cause) { + TimestampedCompletableFuture pending = pendingRequests.remove(requestId); + if (pending != null && pending.completeExceptionally(cause)) { + stats.reportFailedRequest(); + } + } + + @Override + public void onFailure(Throwable cause) { + if (close(cause)) { + // Remove from established channels, otherwise future + // requests will be handled by this failed channel. + establishedConnections.remove(serverAddress, this); + } + } + + @Override + public String toString() { + return "EstablishedConnection{" + + "serverAddress=" + serverAddress + + ", channel=" + channel + + ", pendingRequests=" + pendingRequests.size() + + ", requestCount=" + requestCount + + ", failureCause=" + failureCause + + '}'; + } + + /** + * Pair of promise and a timestamp. + */ + private class TimestampedCompletableFuture extends CompletableFuture<RESP> { + + private final long timestampInNanos; + + TimestampedCompletableFuture(long timestampInNanos) { + this.timestampInNanos = timestampInNanos; + } + + public long getTimestamp() { + return timestampInNanos; + } + } + } +}
http://git-wip-us.apache.org/repos/asf/flink/blob/f48f5340/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/ClientHandler.java ---------------------------------------------------------------------- diff --git a/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/ClientHandler.java b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/ClientHandler.java new file mode 100644 index 0000000..fc9b1d4 --- /dev/null +++ b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/ClientHandler.java @@ -0,0 +1,122 @@ +/* + * 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.flink.queryablestate.network; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.queryablestate.network.messages.MessageBody; +import org.apache.flink.queryablestate.network.messages.MessageSerializer; +import org.apache.flink.queryablestate.network.messages.MessageType; +import org.apache.flink.queryablestate.network.messages.RequestFailure; +import org.apache.flink.util.Preconditions; + +import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf; +import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext; +import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandlerAdapter; +import org.apache.flink.shaded.netty4.io.netty.util.ReferenceCountUtil; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.nio.channels.ClosedChannelException; + +/** + * The handler used by a {@link Client} to handling incoming messages. + * + * @param <REQ> the type of request the client will send. + * @param <RESP> the type of response the client expects to receive. + */ +@Internal +public class ClientHandler<REQ extends MessageBody, RESP extends MessageBody> extends ChannelInboundHandlerAdapter { + + private static final Logger LOG = LoggerFactory.getLogger(ClientHandler.class); + + private final String clientName; + + private final MessageSerializer<REQ, RESP> serializer; + + private final ClientHandlerCallback<RESP> callback; + + /** + * Creates a handler with the callback. + * + * @param clientName the name of the client. + * @param serializer the serializer used to (de-)serialize messages. + * @param callback Callback for responses. + */ + public ClientHandler( + final String clientName, + final MessageSerializer<REQ, RESP> serializer, + final ClientHandlerCallback<RESP> callback) { + + this.clientName = Preconditions.checkNotNull(clientName); + this.serializer = Preconditions.checkNotNull(serializer); + this.callback = Preconditions.checkNotNull(callback); + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + try { + ByteBuf buf = (ByteBuf) msg; + MessageType msgType = MessageSerializer.deserializeHeader(buf); + + if (msgType == MessageType.REQUEST_RESULT) { + long requestId = MessageSerializer.getRequestId(buf); + RESP result = serializer.deserializeResponse(buf); + callback.onRequestResult(requestId, result); + } else if (msgType == MessageType.REQUEST_FAILURE) { + RequestFailure failure = MessageSerializer.deserializeRequestFailure(buf); + callback.onRequestFailure(failure.getRequestId(), failure.getCause()); + } else if (msgType == MessageType.SERVER_FAILURE) { + throw MessageSerializer.deserializeServerFailure(buf); + } else { + throw new IllegalStateException("Unexpected response type '" + msgType + "'"); + } + } catch (Throwable t1) { + try { + callback.onFailure(t1); + } catch (Throwable t2) { + LOG.error("Failed to notify callback about failure", t2); + } + } finally { + ReferenceCountUtil.release(msg); + } + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { + try { + callback.onFailure(cause); + } catch (Throwable t) { + LOG.error("Failed to notify callback about failure", t); + } + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + // Only the client is expected to close the channel. Otherwise it + // indicates a failure. Note that this will be invoked in both cases + // though. If the callback closed the channel, the callback must be + // ignored. + try { + callback.onFailure(new ClosedChannelException()); + } catch (Throwable t) { + LOG.error("Failed to notify callback about failure", t); + } + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/f48f5340/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/ClientHandlerCallback.java ---------------------------------------------------------------------- diff --git a/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/ClientHandlerCallback.java b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/ClientHandlerCallback.java new file mode 100644 index 0000000..00ce1ed --- /dev/null +++ b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/ClientHandlerCallback.java @@ -0,0 +1,56 @@ +/* + * 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.flink.queryablestate.network; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.queryablestate.network.messages.MessageBody; + +/** + * Callback for {@link ClientHandler}. + */ +@Internal +public interface ClientHandlerCallback<RESP extends MessageBody> { + + /** + * Called on a successful request. + * + * @param requestId ID of the request + * @param response The received response + */ + void onRequestResult(long requestId, RESP response); + + /** + * Called on a failed request. + * + * @param requestId ID of the request + * @param cause Cause of the request failure + */ + void onRequestFailure(long requestId, Throwable cause); + + /** + * Called on any failure, which is not related to a specific request. + * + * <p>This can be for example a caught Exception in the channel pipeline + * or an unexpected channel close. + * + * @param cause Cause of the failure + */ + void onFailure(Throwable cause); + +} http://git-wip-us.apache.org/repos/asf/flink/blob/f48f5340/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/MessageBody.java ---------------------------------------------------------------------- diff --git a/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/MessageBody.java b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/MessageBody.java new file mode 100644 index 0000000..f26c267 --- /dev/null +++ b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/MessageBody.java @@ -0,0 +1,38 @@ +/* + * 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.flink.queryablestate.network.messages; + +import org.apache.flink.annotation.Internal; + +/** + * The base class for every message exchanged during the communication between + * {@link org.apache.flink.queryablestate.network.Client client} and + * {@link org.apache.flink.queryablestate.network.AbstractServerBase server}. + * + * <p>Every such message should also have a {@link MessageDeserializer}. + */ +@Internal +public abstract class MessageBody { + + /** + * Serializes the message into a byte array. + * @return A byte array with the serialized content of the message. + */ + public abstract byte[] serialize(); +} http://git-wip-us.apache.org/repos/asf/flink/blob/f48f5340/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/MessageDeserializer.java ---------------------------------------------------------------------- diff --git a/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/MessageDeserializer.java b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/MessageDeserializer.java new file mode 100644 index 0000000..436fb82 --- /dev/null +++ b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/MessageDeserializer.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.flink.queryablestate.network.messages; + +import org.apache.flink.annotation.Internal; + +import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf; + +/** + * A utility used to deserialize a {@link MessageBody message}. + * @param <M> The type of the message to be deserialized. + * It has to extend {@link MessageBody} + */ +@Internal +public interface MessageDeserializer<M extends MessageBody> { + + /** + * Deserializes a message contained in a byte buffer. + * @param buf the buffer containing the message. + * @return The deserialized message. + */ + M deserializeMessage(ByteBuf buf); +} http://git-wip-us.apache.org/repos/asf/flink/blob/f48f5340/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/MessageSerializer.java ---------------------------------------------------------------------- diff --git a/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/MessageSerializer.java b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/MessageSerializer.java index 32bca64..c0a0d32 100644 --- a/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/MessageSerializer.java +++ b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/MessageSerializer.java @@ -18,11 +18,7 @@ package org.apache.flink.queryablestate.network.messages; -import org.apache.flink.queryablestate.messages.KvStateRequest; -import org.apache.flink.queryablestate.messages.KvStateRequestFailure; -import org.apache.flink.queryablestate.messages.KvStateRequestResult; -import org.apache.flink.runtime.query.KvStateID; -import org.apache.flink.util.AbstractID; +import org.apache.flink.annotation.Internal; import org.apache.flink.util.Preconditions; import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf; @@ -37,8 +33,8 @@ import java.io.ObjectOutputStream; /** * Serialization and deserialization of messages exchanged between - * {@link org.apache.flink.queryablestate.client.KvStateClient client} and - * {@link org.apache.flink.queryablestate.server.KvStateServerImpl server}. + * {@link org.apache.flink.queryablestate.network.Client client} and + * {@link org.apache.flink.queryablestate.network.AbstractServerBase server}. * * <p>The binary messages have the following format: * @@ -52,8 +48,12 @@ import java.io.ObjectOutputStream; * </pre> * * <p>The concrete content of a message depends on the {@link MessageType}. + * + * @param <REQ> Type of the requests of the protocol. + * @param <RESP> Type of the responses of the protocol. */ -public final class MessageSerializer { +@Internal +public final class MessageSerializer<REQ extends MessageBody, RESP extends MessageBody> { /** The serialization version ID. */ private static final int VERSION = 0x79a1b710; @@ -64,78 +64,58 @@ public final class MessageSerializer { /** Byte length of the request id. */ private static final int REQUEST_ID_SIZE = Long.BYTES; + /** The constructor of the {@link MessageBody client requests}. Used for deserialization. */ + private final MessageDeserializer<REQ> requestDeserializer; + + /** The constructor of the {@link MessageBody server responses}. Used for deserialization. */ + private final MessageDeserializer<RESP> responseDeserializer; + + public MessageSerializer(MessageDeserializer<REQ> requestDeser, MessageDeserializer<RESP> responseDeser) { + requestDeserializer = Preconditions.checkNotNull(requestDeser); + responseDeserializer = Preconditions.checkNotNull(responseDeser); + } + // ------------------------------------------------------------------------ // Serialization // ------------------------------------------------------------------------ /** - * Allocates a buffer and serializes the KvState request into it. + * Serializes the request sent to the + * {@link org.apache.flink.queryablestate.network.AbstractServerBase}. * - * @param alloc ByteBuf allocator for the buffer to - * serialize message into - * @param requestId ID for this request - * @param kvStateId ID of the requested KvState instance - * @param serializedKeyAndNamespace Serialized key and namespace to request - * from the KvState instance. - * @return Serialized KvState request message + * @param alloc The {@link ByteBufAllocator} used to allocate the buffer to serialize the message into. + * @param requestId The id of the request to which the message refers to. + * @param request The request to be serialized. + * @return A {@link ByteBuf} containing the serialized message. */ - public static ByteBuf serializeKvStateRequest( - ByteBufAllocator alloc, - long requestId, - KvStateID kvStateId, - byte[] serializedKeyAndNamespace) { - - // Header + request ID + KvState ID + Serialized namespace - int frameLength = HEADER_LENGTH + REQUEST_ID_SIZE + AbstractID.SIZE + (Integer.BYTES + serializedKeyAndNamespace.length); - ByteBuf buf = alloc.ioBuffer(frameLength + 4); // +4 for frame length - - buf.writeInt(frameLength); - - writeHeader(buf, MessageType.REQUEST); - - buf.writeLong(requestId); - buf.writeLong(kvStateId.getLowerPart()); - buf.writeLong(kvStateId.getUpperPart()); - buf.writeInt(serializedKeyAndNamespace.length); - buf.writeBytes(serializedKeyAndNamespace); - - return buf; + public static <REQ extends MessageBody> ByteBuf serializeRequest( + final ByteBufAllocator alloc, + final long requestId, + final REQ request) { + Preconditions.checkNotNull(request); + return writePayload(alloc, requestId, MessageType.REQUEST, request.serialize()); } /** - * Allocates a buffer and serializes the KvState request result into it. + * Serializes the response sent to the + * {@link org.apache.flink.queryablestate.network.Client}. * - * @param alloc ByteBuf allocator for the buffer to serialize message into - * @param requestId ID for this request - * @param serializedResult Serialized Result - * @return Serialized KvState request result message + * @param alloc The {@link ByteBufAllocator} used to allocate the buffer to serialize the message into. + * @param requestId The id of the request to which the message refers to. + * @param response The response to be serialized. + * @return A {@link ByteBuf} containing the serialized message. */ - public static ByteBuf serializeKvStateRequestResult( - ByteBufAllocator alloc, - long requestId, - byte[] serializedResult) { - - Preconditions.checkNotNull(serializedResult, "Serialized result"); - - // Header + request ID + serialized result - int frameLength = HEADER_LENGTH + REQUEST_ID_SIZE + 4 + serializedResult.length; - - // TODO: 10/5/17 there was a bug all this time? - ByteBuf buf = alloc.ioBuffer(frameLength + 4); - - buf.writeInt(frameLength); - writeHeader(buf, MessageType.REQUEST_RESULT); - buf.writeLong(requestId); - - buf.writeInt(serializedResult.length); - buf.writeBytes(serializedResult); - - return buf; + public static <RESP extends MessageBody> ByteBuf serializeResponse( + final ByteBufAllocator alloc, + final long requestId, + final RESP response) { + Preconditions.checkNotNull(response); + return writePayload(alloc, requestId, MessageType.REQUEST_RESULT, response.serialize()); } /** * Serializes the exception containing the failure message sent to the - * {@link org.apache.flink.queryablestate.client.KvStateClient} in case of + * {@link org.apache.flink.queryablestate.network.Client} in case of * protocol related errors. * * @param alloc The {@link ByteBufAllocator} used to allocate the buffer to serialize the message into. @@ -143,7 +123,7 @@ public final class MessageSerializer { * @param cause The exception thrown at the server. * @return A {@link ByteBuf} containing the serialized message. */ - public static ByteBuf serializeKvStateRequestFailure( + public static ByteBuf serializeRequestFailure( final ByteBufAllocator alloc, final long requestId, final Throwable cause) throws IOException { @@ -168,7 +148,7 @@ public final class MessageSerializer { /** * Serializes the failure message sent to the - * {@link org.apache.flink.queryablestate.client.KvStateClient} in case of + * {@link org.apache.flink.queryablestate.network.Client} in case of * server related errors. * * @param alloc The {@link ByteBufAllocator} used to allocate the buffer to serialize the message into. @@ -207,6 +187,31 @@ public final class MessageSerializer { buf.writeInt(messageType.ordinal()); } + /** + * Helper for serializing the messages. + * + * @param alloc The {@link ByteBufAllocator} used to allocate the buffer to serialize the message into. + * @param requestId The id of the request to which the message refers to. + * @param messageType The {@link MessageType type of the message}. + * @param payload The serialized version of the message. + * @return A {@link ByteBuf} containing the serialized message. + */ + private static ByteBuf writePayload( + final ByteBufAllocator alloc, + final long requestId, + final MessageType messageType, + final byte[] payload) { + + final int frameLength = HEADER_LENGTH + REQUEST_ID_SIZE + payload.length; + final ByteBuf buf = alloc.ioBuffer(frameLength + Integer.BYTES); + + buf.writeInt(frameLength); + writeHeader(buf, messageType); + buf.writeLong(requestId); + buf.writeBytes(payload); + return buf; + } + // ------------------------------------------------------------------------ // Deserialization // ------------------------------------------------------------------------ @@ -230,71 +235,54 @@ public final class MessageSerializer { // fetching the message type int msgType = buf.readInt(); MessageType[] values = MessageType.values(); - Preconditions.checkState(msgType >= 0 && msgType <= values.length, + Preconditions.checkState(msgType >= 0 && msgType < values.length, "Illegal message type with index " + msgType + '.'); return values[msgType]; } /** - * Deserializes the KvState request message. - * - * <p><strong>Important</strong>: the returned buffer is sliced from the - * incoming ByteBuf stream and retained. Therefore, it needs to be recycled - * by the consumer. - * - * @param buf Buffer to deserialize (expected to be positioned after header) - * @return Deserialized KvStateRequest + * De-serializes the header and returns the {@link MessageType}. + * <pre> + * <b>The buffer is expected to be at the request id position.</b> + * </pre> + * @param buf The {@link ByteBuf} containing the serialized request id. + * @return The request id. */ - public static KvStateRequest deserializeKvStateRequest(ByteBuf buf) { - long requestId = buf.readLong(); - KvStateID kvStateId = new KvStateID(buf.readLong(), buf.readLong()); - - // Serialized key and namespace - int length = buf.readInt(); - - if (length < 0) { - throw new IllegalArgumentException("Negative length for serialized key and namespace. " + - "This indicates a serialization error."); - } - - // Copy the buffer in order to be able to safely recycle the ByteBuf - byte[] serializedKeyAndNamespace = new byte[length]; - if (length > 0) { - buf.readBytes(serializedKeyAndNamespace); - } - - return new KvStateRequest(requestId, kvStateId, serializedKeyAndNamespace); + public static long getRequestId(final ByteBuf buf) { + return buf.readLong(); } /** - * Deserializes the KvState request result. - * - * @param buf Buffer to deserialize (expected to be positioned after header) - * @return Deserialized KvStateRequestResult + * De-serializes the request sent to the + * {@link org.apache.flink.queryablestate.network.AbstractServerBase}. + * <pre> + * <b>The buffer is expected to be at the request position.</b> + * </pre> + * @param buf The {@link ByteBuf} containing the serialized request. + * @return The request. */ - public static KvStateRequestResult deserializeKvStateRequestResult(ByteBuf buf) { - long requestId = buf.readLong(); - - // Serialized KvState - int length = buf.readInt(); - - if (length < 0) { - throw new IllegalArgumentException("Negative length for serialized result. " + - "This indicates a serialization error."); - } - - byte[] serializedValue = new byte[length]; - - if (length > 0) { - buf.readBytes(serializedValue); - } + public REQ deserializeRequest(final ByteBuf buf) { + Preconditions.checkNotNull(buf); + return requestDeserializer.deserializeMessage(buf); + } - return new KvStateRequestResult(requestId, serializedValue); + /** + * De-serializes the response sent to the + * {@link org.apache.flink.queryablestate.network.Client}. + * <pre> + * <b>The buffer is expected to be at the response position.</b> + * </pre> + * @param buf The {@link ByteBuf} containing the serialized response. + * @return The response. + */ + public RESP deserializeResponse(final ByteBuf buf) { + Preconditions.checkNotNull(buf); + return responseDeserializer.deserializeMessage(buf); } /** - * De-serializes the {@link KvStateRequestFailure} sent to the - * {@link org.apache.flink.queryablestate.client.KvStateClient} in case of + * De-serializes the {@link RequestFailure} sent to the + * {@link org.apache.flink.queryablestate.network.Client} in case of * protocol related errors. * <pre> * <b>The buffer is expected to be at the correct position.</b> @@ -302,7 +290,7 @@ public final class MessageSerializer { * @param buf The {@link ByteBuf} containing the serialized failure message. * @return The failure message. */ - public static KvStateRequestFailure deserializeKvStateRequestFailure(final ByteBuf buf) throws IOException, ClassNotFoundException { + public static RequestFailure deserializeRequestFailure(final ByteBuf buf) throws IOException, ClassNotFoundException { long requestId = buf.readLong(); Throwable cause; @@ -310,12 +298,12 @@ public final class MessageSerializer { ObjectInputStream in = new ObjectInputStream(bis)) { cause = (Throwable) in.readObject(); } - return new KvStateRequestFailure(requestId, cause); + return new RequestFailure(requestId, cause); } /** * De-serializes the failure message sent to the - * {@link org.apache.flink.queryablestate.client.KvStateClient} in case of + * {@link org.apache.flink.queryablestate.network.Client} in case of * server related errors. * <pre> * <b>The buffer is expected to be at the correct position.</b> http://git-wip-us.apache.org/repos/asf/flink/blob/f48f5340/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/MessageType.java ---------------------------------------------------------------------- diff --git a/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/MessageType.java b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/MessageType.java index 4e4435d..562ce93 100644 --- a/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/MessageType.java +++ b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/MessageType.java @@ -18,11 +18,14 @@ package org.apache.flink.queryablestate.network.messages; +import org.apache.flink.annotation.Internal; + /** * Expected message types during the communication between - * {@link org.apache.flink.queryablestate.client.KvStateClient state client} and - * {@link org.apache.flink.queryablestate.server.KvStateServerImpl state server}. + * {@link org.apache.flink.queryablestate.network.Client client} and + * {@link org.apache.flink.queryablestate.network.AbstractServerBase server}. */ +@Internal public enum MessageType { /** The message is a request. */ http://git-wip-us.apache.org/repos/asf/flink/blob/f48f5340/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/RequestFailure.java ---------------------------------------------------------------------- diff --git a/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/RequestFailure.java b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/RequestFailure.java new file mode 100644 index 0000000..106199f --- /dev/null +++ b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/network/messages/RequestFailure.java @@ -0,0 +1,71 @@ +/* + * 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.flink.queryablestate.network.messages; + +import org.apache.flink.annotation.Internal; + +/** + * A message indicating a protocol-related error. + */ +@Internal +public class RequestFailure { + + /** ID of the request responding to. */ + private final long requestId; + + /** Failure cause. Not allowed to be a user type. */ + private final Throwable cause; + + /** + * Creates a failure response to a {@link MessageBody}. + * + * @param requestId ID for the request responding to + * @param cause Failure cause (not allowed to be a user type) + */ + public RequestFailure(long requestId, Throwable cause) { + this.requestId = requestId; + this.cause = cause; + } + + /** + * Returns the request ID responding to. + * + * @return Request ID responding to + */ + public long getRequestId() { + return requestId; + } + + /** + * Returns the failure cause. + * + * @return Failure cause + */ + public Throwable getCause() { + return cause; + } + + @Override + public String toString() { + return "RequestFailure{" + + "requestId=" + requestId + + ", cause=" + cause + + '}'; + } +} http://git-wip-us.apache.org/repos/asf/flink/blob/f48f5340/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/server/ChunkedByteBuf.java ---------------------------------------------------------------------- diff --git a/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/server/ChunkedByteBuf.java b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/server/ChunkedByteBuf.java deleted file mode 100644 index f10969e..0000000 --- a/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/server/ChunkedByteBuf.java +++ /dev/null @@ -1,98 +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.flink.queryablestate.server; - -import org.apache.flink.util.Preconditions; - -import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf; -import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext; -import org.apache.flink.shaded.netty4.io.netty.handler.stream.ChunkedInput; -import org.apache.flink.shaded.netty4.io.netty.handler.stream.ChunkedWriteHandler; - -/** - * A {@link ByteBuf} instance to be consumed in chunks by {@link ChunkedWriteHandler}, - * respecting the high and low watermarks. - * - * @see <a href="http://normanmaurer.me/presentations/2014-facebook-eng-netty/slides.html#10.0";>Low/High Watermarks</a> - */ -public class ChunkedByteBuf implements ChunkedInput<ByteBuf> { - - /** The buffer to chunk. */ - private final ByteBuf buf; - - /** Size of chunks. */ - private final int chunkSize; - - /** Closed flag. */ - private boolean isClosed; - - /** End of input flag. */ - private boolean isEndOfInput; - - public ChunkedByteBuf(ByteBuf buf, int chunkSize) { - this.buf = Preconditions.checkNotNull(buf, "Buffer"); - Preconditions.checkArgument(chunkSize > 0, "Non-positive chunk size"); - this.chunkSize = chunkSize; - } - - @Override - public boolean isEndOfInput() throws Exception { - return isClosed || isEndOfInput; - } - - @Override - public void close() throws Exception { - if (!isClosed) { - // If we did not consume the whole buffer yet, we have to release - // it here. Otherwise, it's the responsibility of the consumer. - if (!isEndOfInput) { - buf.release(); - } - - isClosed = true; - } - } - - @Override - public ByteBuf readChunk(ChannelHandlerContext ctx) throws Exception { - if (isClosed) { - return null; - } else if (buf.readableBytes() <= chunkSize) { - isEndOfInput = true; - - // Don't retain as the consumer is responsible to release it - return buf.slice(); - } else { - // Return a chunk sized slice of the buffer. The ref count is - // shared with the original buffer. That's why we need to retain - // a reference here. - return buf.readSlice(chunkSize).retain(); - } - } - - @Override - public String toString() { - return "ChunkedByteBuf{" + - "buf=" + buf + - ", chunkSize=" + chunkSize + - ", isClosed=" + isClosed + - ", isEndOfInput=" + isEndOfInput + - '}'; - } -} http://git-wip-us.apache.org/repos/asf/flink/blob/f48f5340/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/server/KvStateServerHandler.java ---------------------------------------------------------------------- diff --git a/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/server/KvStateServerHandler.java b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/server/KvStateServerHandler.java index 9a31fca..055a5d0 100644 --- a/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/server/KvStateServerHandler.java +++ b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/server/KvStateServerHandler.java @@ -18,31 +18,25 @@ package org.apache.flink.queryablestate.server; -import org.apache.flink.queryablestate.UnknownKeyOrNamespace; -import org.apache.flink.queryablestate.UnknownKvStateID; -import org.apache.flink.queryablestate.messages.KvStateRequest; +import org.apache.flink.annotation.Internal; +import org.apache.flink.queryablestate.UnknownKeyOrNamespaceException; +import org.apache.flink.queryablestate.UnknownKvStateIdException; +import org.apache.flink.queryablestate.messages.KvStateInternalRequest; +import org.apache.flink.queryablestate.messages.KvStateResponse; +import org.apache.flink.queryablestate.network.AbstractServerHandler; import org.apache.flink.queryablestate.network.messages.MessageSerializer; -import org.apache.flink.queryablestate.network.messages.MessageType; import org.apache.flink.runtime.query.KvStateRegistry; import org.apache.flink.runtime.query.netty.KvStateRequestStats; import org.apache.flink.runtime.state.internal.InternalKvState; import org.apache.flink.util.ExceptionUtils; +import org.apache.flink.util.Preconditions; -import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf; -import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFuture; -import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFutureListener; import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandler; -import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext; -import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandlerAdapter; -import org.apache.flink.shaded.netty4.io.netty.util.ReferenceCountUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; -import java.util.Objects; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.CompletableFuture; /** * This handler dispatches asynchronous tasks, which query {@link InternalKvState} @@ -52,257 +46,62 @@ import java.util.concurrent.TimeUnit; * query task. The actual query is handled in a separate thread as it might * otherwise block the network threads (file I/O etc.). */ +@Internal @ChannelHandler.Sharable -public class KvStateServerHandler extends ChannelInboundHandlerAdapter { +public class KvStateServerHandler extends AbstractServerHandler<KvStateInternalRequest, KvStateResponse> { private static final Logger LOG = LoggerFactory.getLogger(KvStateServerHandler.class); /** KvState registry holding references to the KvState instances. */ private final KvStateRegistry registry; - /** Thread pool for query execution. */ - private final ExecutorService queryExecutor; - - /** Exposed server statistics. */ - private final KvStateRequestStats stats; - /** - * Create the handler. + * Create the handler used by the {@link KvStateServerImpl}. * - * @param kvStateRegistry Registry to query. - * @param queryExecutor Thread pool for query execution. - * @param stats Exposed server statistics. + * @param server the {@link KvStateServerImpl} using the handler. + * @param kvStateRegistry registry to query. + * @param serializer the {@link MessageSerializer} used to (de-) serialize the different messages. + * @param stats server statistics collector. */ public KvStateServerHandler( - KvStateRegistry kvStateRegistry, - ExecutorService queryExecutor, - KvStateRequestStats stats) { + final KvStateServerImpl server, + final KvStateRegistry kvStateRegistry, + final MessageSerializer<KvStateInternalRequest, KvStateResponse> serializer, + final KvStateRequestStats stats) { - this.registry = Objects.requireNonNull(kvStateRegistry, "KvStateRegistry"); - this.queryExecutor = Objects.requireNonNull(queryExecutor, "Query thread pool"); - this.stats = Objects.requireNonNull(stats, "KvStateRequestStats"); + super(server, serializer, stats); + this.registry = Preconditions.checkNotNull(kvStateRegistry); } @Override - public void channelActive(ChannelHandlerContext ctx) throws Exception { - stats.reportActiveConnection(); - } - - @Override - public void channelInactive(ChannelHandlerContext ctx) throws Exception { - stats.reportInactiveConnection(); - } - - @Override - public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { - KvStateRequest request = null; + public CompletableFuture<KvStateResponse> handleRequest(final long requestId, final KvStateInternalRequest request) { + final CompletableFuture<KvStateResponse> responseFuture = new CompletableFuture<>(); try { - ByteBuf buf = (ByteBuf) msg; - MessageType msgType = MessageSerializer.deserializeHeader(buf); - - if (msgType == MessageType.REQUEST) { - // ------------------------------------------------------------ - // Request - // ------------------------------------------------------------ - request = MessageSerializer.deserializeKvStateRequest(buf); - - stats.reportRequest(); - - InternalKvState<?> kvState = registry.getKvState(request.getKvStateId()); - - if (kvState != null) { - // Execute actual query async, because it is possibly - // blocking (e.g. file I/O). - // - // A submission failure is not treated as fatal. - queryExecutor.submit(new AsyncKvStateQueryTask(ctx, request, kvState, stats)); - } else { - ByteBuf unknown = MessageSerializer.serializeKvStateRequestFailure( - ctx.alloc(), - request.getRequestId(), - new UnknownKvStateID(request.getKvStateId())); - - ctx.writeAndFlush(unknown); - - stats.reportFailedRequest(); - } + final InternalKvState<?> kvState = registry.getKvState(request.getKvStateId()); + if (kvState == null) { + responseFuture.completeExceptionally(new UnknownKvStateIdException(getServerName(), request.getKvStateId())); } else { - // ------------------------------------------------------------ - // Unexpected - // ------------------------------------------------------------ - ByteBuf failure = MessageSerializer.serializeServerFailure( - ctx.alloc(), - new IllegalArgumentException("Unexpected message type " + msgType - + ". KvStateServerHandler expects " - + MessageType.REQUEST + " messages.")); - - ctx.writeAndFlush(failure); - } - } catch (Throwable t) { - String stringifiedCause = ExceptionUtils.stringifyException(t); - - ByteBuf err; - if (request != null) { - String errMsg = "Failed to handle incoming request with ID " + - request.getRequestId() + ". Caused by: " + stringifiedCause; - err = MessageSerializer.serializeKvStateRequestFailure( - ctx.alloc(), - request.getRequestId(), - new RuntimeException(errMsg)); - - stats.reportFailedRequest(); - } else { - String errMsg = "Failed to handle incoming message. Caused by: " + stringifiedCause; - err = MessageSerializer.serializeServerFailure( - ctx.alloc(), - new RuntimeException(errMsg)); - } - - ctx.writeAndFlush(err); - } finally { - // IMPORTANT: We have to always recycle the incoming buffer. - // Otherwise we will leak memory out of Netty's buffer pool. - // - // If any operation ever holds on to the buffer, it is the - // responsibility of that operation to retain the buffer and - // release it later. - ReferenceCountUtil.release(msg); - } - } - - @Override - public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { - String stringifiedCause = ExceptionUtils.stringifyException(cause); - String msg = "Exception in server pipeline. Caused by: " + stringifiedCause; - - ByteBuf err = MessageSerializer.serializeServerFailure( - ctx.alloc(), - new RuntimeException(msg)); - - ctx.writeAndFlush(err).addListener(ChannelFutureListener.CLOSE); - } - - /** - * Task to execute the actual query against the {@link InternalKvState} instance. - */ - private static class AsyncKvStateQueryTask implements Runnable { - - private final ChannelHandlerContext ctx; - - private final KvStateRequest request; - - private final InternalKvState<?> kvState; - - private final KvStateRequestStats stats; - - private final long creationNanos; - - public AsyncKvStateQueryTask( - ChannelHandlerContext ctx, - KvStateRequest request, - InternalKvState<?> kvState, - KvStateRequestStats stats) { - - this.ctx = Objects.requireNonNull(ctx, "Channel handler context"); - this.request = Objects.requireNonNull(request, "State query"); - this.kvState = Objects.requireNonNull(kvState, "KvState"); - this.stats = Objects.requireNonNull(stats, "State query stats"); - this.creationNanos = System.nanoTime(); - } - - @Override - public void run() { - boolean success = false; - - try { - if (!ctx.channel().isActive()) { - return; - } - - // Query the KvState instance byte[] serializedKeyAndNamespace = request.getSerializedKeyAndNamespace(); - byte[] serializedResult = kvState.getSerializedValue(serializedKeyAndNamespace); + byte[] serializedResult = kvState.getSerializedValue(serializedKeyAndNamespace); if (serializedResult != null) { - // We found some data, success! - ByteBuf buf = MessageSerializer.serializeKvStateRequestResult( - ctx.alloc(), - request.getRequestId(), - serializedResult); - - int highWatermark = ctx.channel().config().getWriteBufferHighWaterMark(); - - ChannelFuture write; - if (buf.readableBytes() <= highWatermark) { - write = ctx.writeAndFlush(buf); - } else { - write = ctx.writeAndFlush(new ChunkedByteBuf(buf, highWatermark)); - } - - write.addListener(new QueryResultWriteListener()); - - success = true; + responseFuture.complete(new KvStateResponse(serializedResult)); } else { - // No data for the key/namespace. This is considered to be - // a failure. - ByteBuf unknownKey = MessageSerializer.serializeKvStateRequestFailure( - ctx.alloc(), - request.getRequestId(), - new UnknownKeyOrNamespace()); - - ctx.writeAndFlush(unknownKey); - } - } catch (Throwable t) { - try { - String stringifiedCause = ExceptionUtils.stringifyException(t); - String errMsg = "Failed to query state backend for query " + - request.getRequestId() + ". Caused by: " + stringifiedCause; - - ByteBuf err = MessageSerializer.serializeKvStateRequestFailure( - ctx.alloc(), request.getRequestId(), new RuntimeException(errMsg)); - - ctx.writeAndFlush(err); - } catch (IOException e) { - LOG.error("Failed to respond with the error after failed to query state backend", e); - } - } finally { - if (!success) { - stats.reportFailedRequest(); + responseFuture.completeExceptionally(new UnknownKeyOrNamespaceException(getServerName())); } } + return responseFuture; + } catch (Throwable t) { + String errMsg = "Error while processing request with ID " + requestId + + ". Caused by: " + ExceptionUtils.stringifyException(t); + responseFuture.completeExceptionally(new RuntimeException(errMsg)); + return responseFuture; } + } - @Override - public String toString() { - return "AsyncKvStateQueryTask{" + - ", request=" + request + - ", creationNanos=" + creationNanos + - '}'; - } - - /** - * Callback after query result has been written. - * - * <p>Gathers stats and logs errors. - */ - private class QueryResultWriteListener implements ChannelFutureListener { - - @Override - public void operationComplete(ChannelFuture future) throws Exception { - long durationNanos = System.nanoTime() - creationNanos; - long durationMillis = TimeUnit.MILLISECONDS.convert(durationNanos, TimeUnit.NANOSECONDS); - - if (future.isSuccess()) { - stats.reportSuccessfulRequest(durationMillis); - } else { - if (LOG.isDebugEnabled()) { - LOG.debug("Query " + request + " failed after " + durationMillis + " ms", future.cause()); - } - - stats.reportFailedRequest(); - } - } - } + @Override + public void shutdown() { + // do nothing } } http://git-wip-us.apache.org/repos/asf/flink/blob/f48f5340/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/server/KvStateServerImpl.java ---------------------------------------------------------------------- diff --git a/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/server/KvStateServerImpl.java b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/server/KvStateServerImpl.java index 4bf7e24..b4c548a 100644 --- a/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/server/KvStateServerImpl.java +++ b/flink-queryable-state/flink-queryable-state-java/src/main/java/org/apache/flink/queryablestate/server/KvStateServerImpl.java @@ -18,213 +18,93 @@ package org.apache.flink.queryablestate.server; -import org.apache.flink.queryablestate.messages.KvStateRequest; -import org.apache.flink.runtime.io.network.netty.NettyBufferPool; +import org.apache.flink.annotation.Internal; +import org.apache.flink.queryablestate.messages.KvStateInternalRequest; +import org.apache.flink.queryablestate.messages.KvStateResponse; +import org.apache.flink.queryablestate.network.AbstractServerBase; +import org.apache.flink.queryablestate.network.AbstractServerHandler; +import org.apache.flink.queryablestate.network.messages.MessageSerializer; import org.apache.flink.runtime.query.KvStateRegistry; import org.apache.flink.runtime.query.KvStateServer; import org.apache.flink.runtime.query.KvStateServerAddress; import org.apache.flink.runtime.query.netty.KvStateRequestStats; import org.apache.flink.util.Preconditions; -import org.apache.flink.shaded.guava18.com.google.common.util.concurrent.ThreadFactoryBuilder; -import org.apache.flink.shaded.netty4.io.netty.bootstrap.ServerBootstrap; -import org.apache.flink.shaded.netty4.io.netty.channel.Channel; -import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInitializer; -import org.apache.flink.shaded.netty4.io.netty.channel.ChannelOption; -import org.apache.flink.shaded.netty4.io.netty.channel.EventLoopGroup; -import org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoopGroup; -import org.apache.flink.shaded.netty4.io.netty.channel.socket.SocketChannel; -import org.apache.flink.shaded.netty4.io.netty.channel.socket.nio.NioServerSocketChannel; -import org.apache.flink.shaded.netty4.io.netty.handler.codec.LengthFieldBasedFrameDecoder; -import org.apache.flink.shaded.netty4.io.netty.handler.stream.ChunkedWriteHandler; - import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.net.InetAddress; -import java.net.InetSocketAddress; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.ThreadFactory; -import java.util.concurrent.TimeUnit; /** - * Netty-based server answering {@link KvStateRequest} messages. - * - * <p>Requests are handled by asynchronous query tasks (see {@link KvStateServerHandler.AsyncKvStateQueryTask}) - * that are executed by a separate query Thread pool. This pool is shared among - * all TCP connections. - * - * <p>The incoming pipeline looks as follows: - * <pre> - * Socket.read() -> LengthFieldBasedFrameDecoder -> KvStateServerHandler - * </pre> - * - * <p>Received binary messages are expected to contain a frame length field. Netty's - * {@link LengthFieldBasedFrameDecoder} is used to fully receive the frame before - * giving it to our {@link KvStateServerHandler}. - * - * <p>Connections are established and closed by the client. The server only - * closes the connection on a fatal failure that cannot be recovered. A - * server-side connection close is considered a failure by the client. + * The default implementation of the {@link KvStateServer}. */ -public class KvStateServerImpl implements KvStateServer { - - private static final Logger LOG = LoggerFactory.getLogger(KvStateServer.class); +@Internal +public class KvStateServerImpl extends AbstractServerBase<KvStateInternalRequest, KvStateResponse> implements KvStateServer { - /** Server config: low water mark. */ - private static final int LOW_WATER_MARK = 8 * 1024; + private static final Logger LOG = LoggerFactory.getLogger(KvStateServerImpl.class); - /** Server config: high water mark. */ - private static final int HIGH_WATER_MARK = 32 * 1024; + /** The {@link KvStateRegistry} to query for state instances. */ + private final KvStateRegistry kvStateRegistry; - /** Netty's ServerBootstrap. */ - private final ServerBootstrap bootstrap; + private final KvStateRequestStats stats; - /** Query executor thread pool. */ - private final ExecutorService queryExecutor; - - /** Address of this server. */ - private KvStateServerAddress serverAddress; + private MessageSerializer<KvStateInternalRequest, KvStateResponse> serializer; /** - * Creates the {@link KvStateServer}. + * Creates the state server. + * + * <p>The server is instantiated using reflection by the + * {@link org.apache.flink.runtime.query.QueryableStateUtils#createKvStateServer(InetAddress, int, int, int, KvStateRegistry, KvStateRequestStats) + * QueryableStateUtils.startKvStateServer(InetAddress, int, int, int, KvStateRegistry, KvStateRequestStats)}. * * <p>The server needs to be started via {@link #start()} in order to bind * to the configured bind address. * - * @param bindAddress Address to bind to - * @param bindPort Port to bind to. Pick random port if 0. - * @param numEventLoopThreads Number of event loop threads - * @param numQueryThreads Number of query threads - * @param kvStateRegistry KvStateRegistry to query for KvState instances - * @param stats Statistics tracker + * @param bindAddress the address to listen to. + * @param bindPort the port to listen to. + * @param numEventLoopThreads number of event loop threads. + * @param numQueryThreads number of query threads. + * @param kvStateRegistry {@link KvStateRegistry} to query for state instances. + * @param stats the statistics collector. */ public KvStateServerImpl( - InetAddress bindAddress, - Integer bindPort, - Integer numEventLoopThreads, - Integer numQueryThreads, - KvStateRegistry kvStateRegistry, - KvStateRequestStats stats) { - - Preconditions.checkArgument(bindPort >= 0 && bindPort <= 65536, "Port " + bindPort + - " is out of valid port range (0-65536)."); - - Preconditions.checkArgument(numEventLoopThreads >= 1, "Non-positive number of event loop threads."); - Preconditions.checkArgument(numQueryThreads >= 1, "Non-positive number of query threads."); - - Preconditions.checkNotNull(kvStateRegistry, "KvStateRegistry"); - Preconditions.checkNotNull(stats, "KvStateRequestStats"); - - NettyBufferPool bufferPool = new NettyBufferPool(numEventLoopThreads); - - ThreadFactory threadFactory = new ThreadFactoryBuilder() - .setDaemon(true) - .setNameFormat("Flink KvStateServer EventLoop Thread %d") - .build(); - - NioEventLoopGroup nioGroup = new NioEventLoopGroup(numEventLoopThreads, threadFactory); - - queryExecutor = createQueryExecutor(numQueryThreads); - - // Shared between all channels - KvStateServerHandler serverHandler = new KvStateServerHandler( - kvStateRegistry, - queryExecutor, - stats); - - bootstrap = new ServerBootstrap() - // Bind address and port - .localAddress(bindAddress, bindPort) - // NIO server channels - .group(nioGroup) - .channel(NioServerSocketChannel.class) - // Server channel Options - .option(ChannelOption.ALLOCATOR, bufferPool) - // Child channel options - .childOption(ChannelOption.ALLOCATOR, bufferPool) - .childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, HIGH_WATER_MARK) - .childOption(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, LOW_WATER_MARK) - // See initializer for pipeline details - .childHandler(new KvStateServerChannelInitializer(serverHandler)); + final InetAddress bindAddress, + final Integer bindPort, + final Integer numEventLoopThreads, + final Integer numQueryThreads, + final KvStateRegistry kvStateRegistry, + final KvStateRequestStats stats) { + + super("Queryable State Server", bindAddress, bindPort, numEventLoopThreads, numQueryThreads); + this.stats = Preconditions.checkNotNull(stats); + this.kvStateRegistry = Preconditions.checkNotNull(kvStateRegistry); } @Override - public void start() throws InterruptedException { - Channel channel = bootstrap.bind().sync().channel(); - - InetSocketAddress localAddress = (InetSocketAddress) channel.localAddress(); - serverAddress = new KvStateServerAddress(localAddress.getAddress(), localAddress.getPort()); + public AbstractServerHandler<KvStateInternalRequest, KvStateResponse> initializeHandler() { + this.serializer = new MessageSerializer<>( + new KvStateInternalRequest.KvStateInternalRequestDeserializer(), + new KvStateResponse.KvStateResponseDeserializer()); + return new KvStateServerHandler(this, kvStateRegistry, serializer, stats); } - @Override - public KvStateServerAddress getAddress() { - if (serverAddress == null) { - throw new IllegalStateException("KvStateServer not started yet."); - } - - return serverAddress; + public MessageSerializer<KvStateInternalRequest, KvStateResponse> getSerializer() { + Preconditions.checkState(serializer != null, "Server " + getServerName() + " has not been started."); + return serializer; } @Override - public void shutDown() { - if (bootstrap != null) { - EventLoopGroup group = bootstrap.group(); - if (group != null) { - group.shutdownGracefully(0, 10, TimeUnit.SECONDS); - } - } - - if (queryExecutor != null) { - queryExecutor.shutdown(); - } - - serverAddress = null; + public void start() throws InterruptedException { + super.start(); } - /** - * Creates a thread pool for the query execution. - * - * @param numQueryThreads Number of query threads. - * @return Thread pool for query execution - */ - private static ExecutorService createQueryExecutor(int numQueryThreads) { - ThreadFactory threadFactory = new ThreadFactoryBuilder() - .setDaemon(true) - .setNameFormat("Flink KvStateServer Query Thread %d") - .build(); - - return Executors.newFixedThreadPool(numQueryThreads, threadFactory); + @Override + public KvStateServerAddress getServerAddress() { + return super.getServerAddress(); } - /** - * Channel pipeline initializer. - * - * <p>The request handler is shared, whereas the other handlers are created - * per channel. - */ - private static final class KvStateServerChannelInitializer extends ChannelInitializer<SocketChannel> { - - /** The shared request handler. */ - private final KvStateServerHandler sharedRequestHandler; - - /** - * Creates the channel pipeline initializer with the shared request handler. - * - * @param sharedRequestHandler Shared request handler. - */ - public KvStateServerChannelInitializer(KvStateServerHandler sharedRequestHandler) { - this.sharedRequestHandler = Preconditions.checkNotNull(sharedRequestHandler, "Request handler"); - } - - @Override - protected void initChannel(SocketChannel ch) throws Exception { - ch.pipeline() - .addLast(new ChunkedWriteHandler()) - .addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4)) - .addLast(sharedRequestHandler); - } + @Override + public void shutdown() { + super.shutdown(); } - }
