flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-1984) Integrate Flink with Apache Mesos
Date Thu, 18 Aug 2016 12:07:20 GMT

    [ https://issues.apache.org/jira/browse/FLINK-1984?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15426327#comment-15426327
] 

ASF GitHub Bot commented on FLINK-1984:
---------------------------------------

Github user mxm commented on a diff in the pull request:

    https://github.com/apache/flink/pull/2315#discussion_r75295468
  
    --- Diff: flink-mesos/src/main/java/org/apache/flink/mesos/util/MesosArtifactServer.java
---
    @@ -0,0 +1,304 @@
    +/*
    + * 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.mesos.util;
    +
    +import io.netty.bootstrap.ServerBootstrap;
    +import io.netty.buffer.Unpooled;
    +import io.netty.channel.Channel;
    +import io.netty.channel.ChannelFuture;
    +import io.netty.channel.ChannelFutureListener;
    +import io.netty.channel.ChannelHandler;
    +import io.netty.channel.ChannelHandlerContext;
    +import io.netty.channel.ChannelInitializer;
    +import io.netty.channel.DefaultFileRegion;
    +import io.netty.channel.SimpleChannelInboundHandler;
    +import io.netty.channel.nio.NioEventLoopGroup;
    +import io.netty.channel.socket.SocketChannel;
    +import io.netty.channel.socket.nio.NioServerSocketChannel;
    +import io.netty.handler.codec.http.DefaultFullHttpResponse;
    +import io.netty.handler.codec.http.DefaultHttpResponse;
    +import io.netty.handler.codec.http.FullHttpResponse;
    +import io.netty.handler.codec.http.HttpHeaders;
    +import io.netty.handler.codec.http.HttpRequest;
    +import io.netty.handler.codec.http.HttpResponse;
    +import io.netty.handler.codec.http.HttpResponseStatus;
    +import io.netty.handler.codec.http.HttpServerCodec;
    +import io.netty.handler.codec.http.LastHttpContent;
    +import io.netty.handler.codec.http.router.Handler;
    +import io.netty.handler.codec.http.router.Routed;
    +import io.netty.handler.codec.http.router.Router;
    +import io.netty.util.CharsetUtil;
    +import org.jets3t.service.utils.Mimetypes;
    +import org.slf4j.Logger;
    +import org.slf4j.LoggerFactory;
    +
    +import java.io.File;
    +import java.io.FileNotFoundException;
    +import java.io.RandomAccessFile;
    +import java.net.InetSocketAddress;
    +import java.net.MalformedURLException;
    +import java.net.URL;
    +
    +import static io.netty.handler.codec.http.HttpHeaders.Names.CACHE_CONTROL;
    +import static io.netty.handler.codec.http.HttpHeaders.Names.CONNECTION;
    +import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_TYPE;
    +import static io.netty.handler.codec.http.HttpMethod.GET;
    +import static io.netty.handler.codec.http.HttpMethod.HEAD;
    +import static io.netty.handler.codec.http.HttpResponseStatus.GONE;
    +import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;
    +import static io.netty.handler.codec.http.HttpResponseStatus.METHOD_NOT_ALLOWED;
    +import static io.netty.handler.codec.http.HttpResponseStatus.NOT_FOUND;
    +import static io.netty.handler.codec.http.HttpResponseStatus.OK;
    +import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
    +
    +
    +/**
    + * A generic Mesos artifact server, designed specifically for use by the Mesos Fetcher.
    + *
    + * More information:
    + * http://mesos.apache.org/documentation/latest/fetcher/
    + * http://mesos.apache.org/documentation/latest/fetcher-cache-internals/
    + */
    +public class MesosArtifactServer {
    +
    +	private static final Logger LOG = LoggerFactory.getLogger(MesosArtifactServer.class);
    +
    +	private final Router router;
    +
    +	private ServerBootstrap bootstrap;
    +
    +	private Channel serverChannel;
    +
    +	private URL baseURL;
    +
    +	public MesosArtifactServer(String sessionID, String serverHostname, int configuredPort)
throws Exception {
    +		if (configuredPort < 0 || configuredPort > 0xFFFF) {
    +			throw new IllegalArgumentException("File server port is invalid: " + configuredPort);
    +		}
    +
    +		router = new Router();
    +
    +		ChannelInitializer<SocketChannel> initializer = new ChannelInitializer<SocketChannel>()
{
    +
    +			@Override
    +			protected void initChannel(SocketChannel ch) {
    +				Handler handler = new Handler(router);
    +
    +				ch.pipeline()
    +					.addLast(new HttpServerCodec())
    +					.addLast(handler.name(), handler)
    +					.addLast(new UnknownFileHandler());
    +			}
    +		};
    +
    +		NioEventLoopGroup bossGroup   = new NioEventLoopGroup(1);
    +		NioEventLoopGroup workerGroup = new NioEventLoopGroup();
    +
    +		this.bootstrap = new ServerBootstrap();
    +		this.bootstrap
    +			.group(bossGroup, workerGroup)
    +			.channel(NioServerSocketChannel.class)
    +			.childHandler(initializer);
    +
    +		Channel ch = this.bootstrap.bind(serverHostname, configuredPort).sync().channel();
    +		this.serverChannel = ch;
    +
    +		InetSocketAddress bindAddress = (InetSocketAddress) ch.localAddress();
    +		String address = bindAddress.getAddress().getHostAddress();
    +		int port = bindAddress.getPort();
    +
    +		baseURL = new URL("http", serverHostname, port, "/" + sessionID + "/");
    +
    +		LOG.info("Mesos artifact server listening at " + address + ':' + port);
    +	}
    +
    +	/**
    +	 * Get the server port on which the artifact server is listening.
    +     */
    +	public synchronized int getServerPort() {
    --- End diff --
    
    unused method but we'll probably need it at a later point in time.


> Integrate Flink with Apache Mesos
> ---------------------------------
>
>                 Key: FLINK-1984
>                 URL: https://issues.apache.org/jira/browse/FLINK-1984
>             Project: Flink
>          Issue Type: New Feature
>          Components: Cluster Management
>            Reporter: Robert Metzger
>            Assignee: Eron Wright 
>            Priority: Minor
>         Attachments: 251.patch
>
>
> There are some users asking for an integration of Flink into Mesos.
> -There also is a pending pull request for adding Mesos support for Flink-: https://github.com/apache/flink/pull/251
> Update (May '16):  a new effort is now underway, building on the recent ResourceManager
work.
> Design document:  ([google doc|https://docs.google.com/document/d/1WItafBmGbjlaBbP8Of5PAFOH9GUJQxf5S4hjEuPchuU/edit?usp=sharing])



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Mime
View raw message