{"action":"created","comment":{"url":"https://api.github.com/repos/apache/flink/pulls/comments/56368853","id":56368853,"diff_hunk":"@@ -0,0 +1,796 @@\n+/*\n+ * Licensed to the Apache Software Foundation (ASF) under one\n+ * or more contributor license agreements.  See the NOTICE file\n+ * distributed with this work for additional information\n+ * regarding copyright ownership.  The ASF licenses this file\n+ * to you under the Apache License, Version 2.0 (the\n+ * \"License\"); you may not use this file except in compliance\n+ * with the License.  You may obtain a copy of the License at\n+ *\n+ *     http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an \"AS IS\" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+\n+package org.apache.flink.runtime.clusterframework;\n+\n+import akka.actor.ActorRef;\n+import akka.actor.ActorSelection;\n+import akka.actor.ActorSystem;\n+import akka.actor.Props;\n+import akka.dispatch.OnComplete;\n+import akka.pattern.Patterns;\n+import akka.util.Timeout;\n+\n+import com.google.common.base.Preconditions;\n+import org.apache.flink.configuration.ConfigConstants;\n+import org.apache.flink.configuration.Configuration;\n+import org.apache.flink.runtime.akka.AkkaUtils;\n+import org.apache.flink.runtime.akka.FlinkUntypedActor;\n+import org.apache.flink.runtime.clusterframework.messages.CheckAndAllocateContainers;\n+import org.apache.flink.runtime.clusterframework.messages.FatalErrorOccurred;\n+import org.apache.flink.runtime.clusterframework.messages.InfoMessage;\n+import org.apache.flink.runtime.clusterframework.messages.RegisterInfoMessageListenerSuccessful;\n+import org.apache.flink.runtime.clusterframework.messages.RegisterResource;\n+import org.apache.flink.runtime.clusterframework.messages.RegisterResourceFailed;\n+import org.apache.flink.runtime.clusterframework.messages.RegisterResourceManagerSuccessful;\n+import org.apache.flink.runtime.clusterframework.messages.RegisterResourceSuccessful;\n+import org.apache.flink.runtime.clusterframework.messages.NewLeaderAvailable;\n+import org.apache.flink.runtime.clusterframework.messages.RegisterInfoMessageListener;\n+import org.apache.flink.runtime.clusterframework.messages.RegisterResourceManager;\n+import org.apache.flink.runtime.clusterframework.messages.RemoveResource;\n+import org.apache.flink.runtime.clusterframework.messages.ResourceRemoved;\n+import org.apache.flink.runtime.clusterframework.messages.SetWorkerPoolSize;\n+import org.apache.flink.runtime.clusterframework.messages.StopCluster;\n+import org.apache.flink.runtime.clusterframework.messages.TriggerRegistrationAtJobManager;\n+import org.apache.flink.runtime.clusterframework.messages.UnRegisterInfoMessageListener;\n+import org.apache.flink.runtime.clusterframework.types.ResourceID;\n+import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalListener;\n+import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;\n+import org.apache.flink.runtime.messages.JobManagerMessages.LeaderSessionMessage;\n+\n+import org.apache.flink.runtime.messages.RegistrationMessages;\n+import org.apache.flink.util.ExceptionUtils;\n+import org.slf4j.Logger;\n+import org.slf4j.LoggerFactory;\n+\n+import scala.concurrent.Future;\n+import scala.concurrent.duration.Duration;\n+import scala.concurrent.duration.FiniteDuration;\n+\n+import java.util.ArrayList;\n+import java.util.Collection;\n+import java.util.HashMap;\n+import java.util.HashSet;\n+import java.util.List;\n+import java.util.Map;\n+import java.util.Set;\n+import java.util.UUID;\n+import java.util.concurrent.TimeUnit;\n+\n+import static java.util.Objects.requireNonNull;\n+\n+/**\n+ *\n+ * <h1>Worker allocation steps</h1>\n+ *\n+ * <ol>\n+ *     <li>The resource manager decides to request more workers. This can happen in order\n+ *         to fill the initial pool, or as a result of the JobManager requesting more workers.</li>\n+ *\n+ *     <li>The resource master calls {@link #requestNewWorkers(int)}, which triggers requests\n+ *         for more containers. After that, the {@link #getNumWorkerRequestsPending()}\n+ *         should reflect the pending requests.</li>\n+ *\n+ *     <li>The concrete framework may acquire containers and then trigger to start TaskManagers\n+ *         in those containers. That should be reflected in {@link #getNumWorkersPendingRegistration()}.</li>\n+ *\n+ *     <li>At some point, the TaskManager processes will have started and send a registration\n+ *         message to the JobManager. The JobManager will perform\n+ *         a lookup with the ResourceManager to check if it really started this TaskManager.\n+ *         The method {@link #workerRegistered(ResourceID)} will be called\n+ *         to inform about a registered worker.</li>\n+ * </ol>\n+ *\n+ */\n+public abstract class FlinkResourceManager<WorkerType extends ResourceID> extends FlinkUntypedActor {\n+\n+\t/** The exit code with which the process is stopped in case of a fatal error */\n+\tprotected static final int EXIT_CODE_FATAL_ERROR = -13;\n+\n+\t/** The default name of the resource manager actor */\n+\tpublic static final String RESOURCE_MANAGER_NAME = \"resourcemanager\";\n+\n+\t// ------------------------------------------------------------------------\n+\n+\t/** The logger, named for the actual implementing class */\n+\tprotected final Logger log = LoggerFactory.getLogger(getClass());\n+\n+\t/** The Flink configuration object */\n+\tprotected final Configuration config;\n+\n+\t/** The timeout for actor messages sent to the JobManager / TaskManagers */\n+\tprivate final FiniteDuration messageTimeout;\n+\n+\t/** The service to find the right leader JobManager (to support high availability) */\n+\tprivate final LeaderRetrievalService leaderRetriever;\n+\n+\t/** The currently registered resources */\n+\tprivate final Map<ResourceID, WorkerType> registeredWorkers;\n+\n+\t/** List of listeners for info messages */\n+\tprivate final Set<ActorRef> infoMessageListeners;\n+\n+\t/** The JobManager that the framework master manages resources for */\n+\tprivate ActorRef jobManager;\n+\n+\t/** Our JobManager's leader session */\n+\tprivate UUID leaderSessionID;\n+\n+\t/** The size of the worker pool that the resource master strives to maintain */\n+\tprivate int designatedPoolSize;\n+\n+\t// ------------------------------------------------------------------------\n+\n+\t/**\n+\t * Creates a AbstractFrameworkMaster actor.\n+\t *\n+\t * @param flinkConfig The Flink configuration object.\n+\t */\n+\tprotected FlinkResourceManager(\n+\t\t\tint numInitialTaskManagers,\n+\t\t\tConfiguration flinkConfig,\n+\t\t\tLeaderRetrievalService leaderRetriever) {\n+\t\tthis.config = requireNonNull(flinkConfig);\n+\t\tthis.leaderRetriever = requireNonNull(leaderRetriever);\n+\t\tthis.registeredWorkers = new HashMap<>();\n+\n+\t\tFiniteDuration lt;\n+\t\ttry {\n+\t\t\tlt = AkkaUtils.getLookupTimeout(config);\n+\t\t}\n+\t\tcatch (Exception e) {\n+\t\t\tlt = new FiniteDuration(\n+\t\t\t\tDuration.apply(ConfigConstants.DEFAULT_AKKA_LOOKUP_TIMEOUT).toMillis(),\n+\t\t\t\tTimeUnit.MILLISECONDS);\n+\t\t}\n+\t\tthis.messageTimeout = lt;\n+\t\tthis.designatedPoolSize = numInitialTaskManagers;\n+\t\tthis.infoMessageListeners = new HashSet<>();\n+\t}\n+\n+\t// ------------------------------------------------------------------------\n+\t//  Actor Behavior\n+\t// ------------------------------------------------------------------------\n+\n+\t@Override\n+\tpublic void preStart() {\n+\t\ttry {\n+\t\t\t// we start our leader retrieval service to make sure we get informed\n+\t\t\t// about JobManager leader changes\n+\t\t\tleaderRetriever.start(new LeaderRetrievalListener() {\n+\n+\t\t\t\t@Override\n+\t\t\t\tpublic void notifyLeaderAddress(String leaderAddress, UUID leaderSessionID) {\n+\t\t\t\t\tself().tell(\n+\t\t\t\t\t\tnew NewLeaderAvailable(leaderAddress, leaderSessionID),\n+\t\t\t\t\t\tActorRef.noSender());\n+\t\t\t\t}\n+\n+\t\t\t\t@Override\n+\t\t\t\tpublic void handleError(Exception e) {\n+\t\t\t\t\tself().tell(\n+\t\t\t\t\t\tnew FatalErrorOccurred(\"Leader retrieval service failed\", e),\n+\t\t\t\t\t\tActorRef.noSender());\n+\t\t\t\t}\n+\t\t\t});\n+\n+\t\t\t// framework specific initialization\n+\t\t\tinitialize();\n+\n+\t\t}\n+\t\tcatch (Throwable t) {\n+\t\t\tself().tell(\n+\t\t\t\tnew FatalErrorOccurred(\"Error during startup of ResourceManager actor\", t),\n+\t\t\t\tActorRef.noSender());\n+\t\t}\n+\t}\n+\n+\t@Override\n+\tpublic void postStop() {\n+\t\ttry {\n+\t\t\tleaderRetriever.stop();\n+\t\t}\n+\t\tcatch (Throwable t) {\n+\t\t\tLOG.error(\"Could not cleanly shut down leader retrieval service\", t);\n+\t\t}\n+\t}\n+\n+\t/**\n+\t *\n+\t * This method receives the actor messages after they have been filtered for\n+\t * a match with the leader session.\n+\t *\n+\t * @param message The incoming actor message.\n+\t */\n+\t@Override\n+\tprotected void handleMessage(Object message) {\n+\t\ttry {\n+\t\t\t// --- messages about worker allocation and pool sizes\n+\n+\t\t\tif (message instanceof CheckAndAllocateContainers) {\n+\t\t\t\tcheckWorkersPool();\n+\t\t\t}\n+\t\t\telse if (message instanceof SetWorkerPoolSize) {\n+\t\t\t\tSetWorkerPoolSize msg = (SetWorkerPoolSize) message;\n+\t\t\t\tadjustDesignatedNumberOfWorkers(msg.numberOfWorkers());\n+\t\t\t}\n+\t\t\telse if (message instanceof RemoveResource) {\n+\t\t\t\tRemoveResource msg = (RemoveResource) message;\n+\t\t\t\tremoveRegisteredResource(msg.resourceId());\n+\t\t\t}\n+\n+\t\t\t// --- lookup of registered resources\n+\n+\t\t\telse if (message instanceof RegisterResource) {\n+\t\t\t\tRegisterResource msg = (RegisterResource) message;\n+\t\t\t\thandleRegisterResource(sender(), msg.getTaskManager(), msg.getRegisterMessage());\n+\t\t\t}\n+\n+\t\t\t// --- messages about JobManager leader status and registration\n+\n+\t\t\telse if (message instanceof NewLeaderAvailable) {\n+\t\t\t\tNewLeaderAvailable msg = (NewLeaderAvailable) message;\n+\t\t\t\tnewJobManagerLeaderAvailable(msg.leaderAddress(), msg.leaderSessionId());\n+\t\t\t}\n+\t\t\telse if (message instanceof TriggerRegistrationAtJobManager) {\n+\t\t\t\tTriggerRegistrationAtJobManager msg = (TriggerRegistrationAtJobManager) message;\n+\t\t\t\ttriggerConnectingToJobManager(msg.jobManagerAddress());\n+\t\t\t}\n+\t\t\telse if (message instanceof RegisterResourceManagerSuccessful) {\n+\t\t\t\tRegisterResourceManagerSuccessful msg = (RegisterResourceManagerSuccessful) message;\n+\t\t\t\tjobManagerLeaderConnected(msg.jobManager(), msg.currentlyRegisteredTaskManagers());\n+\t\t\t}\n+\n+\t\t\t// --- end of application\n+\n+\t\t\telse if (message instanceof StopCluster) {\n+\t\t\t\tStopCluster msg = (StopCluster) message;\n+\t\t\t\tshutdownCluster(msg.finalStatus(), msg.message());\n+\t\t\t}\n+\n+\t\t\t// --- miscellaneous messages\n+\n+\t\t\telse if (message instanceof RegisterInfoMessageListener) {\n+\t\t\t\tif (jobManager != null) {\n+\t\t\t\t\tinfoMessageListeners.add(sender());\n+\t\t\t\t\tsender().tell(decorateMessage(\n+\t\t\t\t\t\tRegisterInfoMessageListenerSuccessful.get()),\n+\t\t\t\t\t\t// answer as the JobManager\n+\t\t\t\t\t\tjobManager);\n+\t\t\t\t}\n+\t\t\t}\n+\n+\t\t\telse if (message instanceof UnRegisterInfoMessageListener) {\n+\t\t\t\tinfoMessageListeners.remove(sender());\n+\t\t\t}\n+\n+\t\t\t// --- unknown messages\n+\n+\t\t\telse {\n+\t\t\t\tLOG.error(\"Discarding unknown message: {}\", message);\n+\t\t\t}\n+\t\t}\n+\t\tcatch (Throwable t) {\n+\t\t\t// fatal error, needs master recovery\n+\t\t\tfatalError(\"Error processing actor message\", t);\n+\t\t}\n+\t}\n+\n+\t@Override\n+\tprotected final UUID getLeaderSessionID() {\n+\t\treturn leaderSessionID;\n+\t}\n+\n+\t// ------------------------------------------------------------------------\n+\t//  Status\n+\t// ------------------------------------------------------------------------\n+\n+\t/**\n+\t * Gets the current designated worker pool size, meaning the number of workers\n+\t * that the resource master strives to maintain. The actual number of workers\n+\t * may be lower (if worker requests are still pending) or higher (if workers have\n+\t * not yet been released).\n+\t *\n+\t * @return The designated worker pool size.\n+\t */\n+\tpublic int getDesignatedWorkerPoolSize() {\n+\t\treturn designatedPoolSize;\n+\t}\n+\n+\t/**\n+\t * Gets the number of currently registered TaskManagers.\n+\t *\n+\t * @return The number of currently registered TaskManagers.\n+\t */\n+\tpublic int getNumberOfRegisteredTaskManagers() {\n+\t\treturn registeredWorkers.size();\n+\t}\n+\n+\t/**\n+\t * Gets the currently registered resources.\n+\t * @return\n+\t */\n+\tpublic Collection<WorkerType> getRegisteredTaskManagers() {\n+\t\treturn registeredWorkers.values();\n+\t}\n+\n+\t/**\n+\t * Gets the registered worker for a given resource ID, if one is available.\n+\t *\n+\t * @param resourceId The resource ID for the worker.\n+\t * @return True if already registered, otherwise false\n+\t */\n+\tpublic boolean isRegistered(ResourceID resourceId) {\n+\t\treturn registeredWorkers.containsKey(resourceId);\n+\t}\n+\n+\t/**\n+\t * Gets an iterable for all currently registered TaskManagers.\n+\t *\n+\t * @return All currently registered TaskManagers.\n+\t */\n+\tpublic Collection<WorkerType> allRegisteredWorkers() {\n+\t\treturn registeredWorkers.values();\n+\t}\n+\n+\t/**\n+\t * Register a resource on which a TaskManager has been started\n+\t * @param jobManager The sender (JobManager) of the message\n+\t * @param taskManager The task manager who wants to register\n+\t * @param msg The task manager's registration message\n+\t */\n+\tprivate void handleRegisterResource(ActorRef jobManager, ActorRef taskManager,\n+\t\t\t\tRegistrationMessages.RegisterTaskManager msg) {\n+\n+\t\tResourceID resourceID = msg.resourceId();\n+\t\ttry {\n+\t\t\tPreconditions.checkNotNull(resourceID);\n+\t\t\tWorkerType newWorker = workerRegistered(msg.resourceId());\n+\t\t\tWorkerType oldWorker = registeredWorkers.put(resourceID, newWorker);\n+\t\t\tif (oldWorker != null) {\n+\t\t\t\tLOG.warn(\"Worker {} had been registered before.\", resourceID);\n+\t\t\t}\n+\t\t\tjobManager.tell(decorateMessage(\n+\t\t\t\tnew RegisterResourceSuccessful(taskManager, msg)),\n+\t\t\t\tself());\n+\t\t} catch (Exception e) {\n+\t\t\t// This may happen on duplicate task manager registration message to the job manager\n+\t\t\tLOG.warn(\"TaskManager resource registration failed for {}\", resourceID);\n+\n+\t\t\t// tell the JobManager about the failure\n+\t\t\tString eStr = ExceptionUtils.stringifyException(e);\n+\t\t\tsender().tell(decorateMessage(\n+\t\t\t\tnew RegisterResourceFailed(taskManager, resourceID, eStr)), self());\n+\t\t}\n+\t}\n+\n+\t/**\n+\t * Releases the given resource. Note that this does not automatically shrink\n+\t * the designated worker pool size.\n+\t *\n+\t * @param resourceId The TaskManager's resource id.\n+\t */\n+\tprivate void removeRegisteredResource(ResourceID resourceId) {\n+\n+\t\tWorkerType worker = registeredWorkers.remove(resourceId);\n+\t\tif (worker != null) {\n+\t\t\treleaseRegisteredWorker(worker);\n+\t\t} else {\n+\t\t\tLOG.warn(\"Resource {} could not be released\", resourceId);\n+\t\t}\n+\t}\n+\n+\n+\t// ------------------------------------------------------------------------\n+\t//  Registration and consolidation with JobManager Leader\n+\t// ------------------------------------------------------------------------\n+\n+\t/**\n+\t * Called as soon as we discover (via leader election) that a JobManager lost leadership\n+\t * or a different one gained leadership.\n+\t *\n+\t * @param leaderAddress The address (Akka URL) of the new leader. Null if there is currently no leader.\n+\t * @param leaderSessionID The unique session ID marking the leadership session.\n+\t */\n+\tprotected void newJobManagerLeaderAvailable(String leaderAddress, UUID leaderSessionID) {\n+\t\tlog.debug(\"Received new leading JobManager {}. Connecting.\", leaderAddress);\n+\n+\t\t// disconnect from the current leader (no-op if no leader yet)\n+\t\tjobManagerLostLeadership();\n+\n+\t\t// a null leader address means that only a leader disconnect\n+\t\t// happened, without a new leader yet\n+\t\tif (leaderAddress != null) {\n+\t\t\t// the leaderSessionID implicitly filters out success and failure messages\n+\t\t\t// that come after leadership changed again\n+\t\t\tthis.leaderSessionID = leaderSessionID;\n+\t\t\ttriggerConnectingToJobManager(leaderAddress);\n+\t\t}\n+\t}\n+\n+\t/**\n+\t * Causes the resource manager to announce itself at the new leader JobManager and\n+\t * obtains its connection information and currently known TaskManagers.\n+\t *\n+\t * @param leaderAddress The akka actor URL of the new leader JobManager.\n+\t */\n+\tprivate void triggerConnectingToJobManager(String leaderAddress) {\n+\t\tLOG.info(\"Trying to associate with JobManager leader \" + leaderAddress);\n+\n+\t\tfinal Object registerMessage = decorateMessage(new RegisterResourceManager(self()));\n+\t\tfinal Object retryMessage = decorateMessage(new TriggerRegistrationAtJobManager(leaderAddress));\n+\n+\t\t// send the registration message to the JobManager\n+\t\tActorSelection jobManagerSel = context().actorSelection(leaderAddress);\n+\t\tFuture<Object> future = Patterns.ask(jobManagerSel, registerMessage, new Timeout(messageTimeout));\n+\n+\t\tfuture.onComplete(new OnComplete<Object>() {\n+\n+\t\t\t@Override\n+\t\t\tpublic void onComplete(Throwable failure, Object msg) {\n+\t\t\t\tif (msg != null) {\n+\t\t\t\t\tif (msg instanceof LeaderSessionMessage &&\n+\t\t\t\t\t\t((LeaderSessionMessage) msg).message() instanceof RegisterResourceManagerSuccessful)\n+\t\t\t\t\t{\n+\t\t\t\t\t\tself().tell(msg, ActorRef.noSender());\n+\t\t\t\t\t}\n+\t\t\t\t\telse {\n+\t\t\t\t\t\tLOG.error(\"Invalid response type to registration at JobManager: {}\", msg);\n+\t\t\t\t\t\tself().tell(retryMessage, ActorRef.noSender());\n+\t\t\t\t\t}\n+\t\t\t\t}\n+\t\t\t\telse {\n+\t\t\t\t\t// no success\n+\t\t\t\t\tLOG.error(\"Resource manager could not register at JobManager\", failure);\n+\t\t\t\t\tself().tell(retryMessage, ActorRef.noSender());\n+\t\t\t\t}\n+\t\t\t}\n+\n+\t\t}, context().dispatcher());\n+\t}\n+\n+\t/**\n+\t * This method disassociates from the current leader JobManager. All currently registered\n+\t * TaskManagers are put under \"awaiting registration\".\n+\t */\n+\tprivate void jobManagerLostLeadership() {\n+\t\tif (jobManager != null) {\n+\t\t\tLOG.info(\"Associated JobManager {} lost leader status\", jobManager);\n+\n+\t\t\tjobManager = null;\n+\t\t\tleaderSessionID = null;\n+\n+\t\t\tinfoMessageListeners.clear();\n+\n+\t\t\tregisteredWorkers.clear();\n+\t\t}\n+\t}\n+\n+\t/**\n+\t * Callback when we're informed about a new leading JobManager.\n+\t * @param newJobManagerLeader The ActorRef of the new jobManager\n+\t * @param workers The existing workers the JobManager has registered.\n+\t */\n+\tprivate void jobManagerLeaderConnected(\n+\t\t\t\t\t\tActorRef newJobManagerLeader,\n+\t\t\t\t\t\tCollection<ResourceID> workers) {\n+\n+\t\tif (jobManager == null) {\n+\t\t\tLOG.info(\"Resource Manager associating with leading JobManager {} - leader session {}\",\n+\t\t\t\t\t\tnewJobManagerLeader, leaderSessionID);\n+\n+\t\t\tjobManager = newJobManagerLeader;\n+\n+\t\t\t// inform the framework that we have updated the leader\n+\t\t\tleaderUpdated();\n+\n+\t\t\tif (workers.size() > 0) {\n+\t\t\t\tLOG.info(\"Received TaskManagers that were registered at the leader JobManager. \" +\n+\t\t\t\t\t\t\"Trying to consolidate.\");\n+\n+\t\t\t\t// keep track of which TaskManagers are not handled\n+\t\t\t\tList<ResourceID> toHandle = new ArrayList<>(workers.size());\n+\t\t\t\tfor (ResourceID resourceID : workers) {\n+\t\t\t\t\ttoHandle.add(resourceID);\n+\t\t\t\t}\n+\n+\t\t\t\ttry {\n+\t\t\t\t\t// ask the framework to tell us which ones we should keep for now\n+\t\t\t\t\tCollection<WorkerType> consolidated = reacceptRegisteredWorkers(workers);\n+\t\t\t\t\tlog.info(\"Consolidated {} TaskManagers\", consolidated.size());\n+\n+\t\t\t\t\t// put the consolidated TaskManagers into our bookkeeping\n+\t\t\t\t\tfor (WorkerType worker : consolidated) {\n+\t\t\t\t\t\tregisteredWorkers.put(worker, worker);\n+\t\t\t\t\t\ttoHandle.remove(worker);\n+\t\t\t\t\t}\n+\t\t\t\t}\n+\t\t\t\tcatch (Throwable t) {\n+\t\t\t\t\tLOG.error(\"Error during consolidation of known TaskManagers\", t);\n+\t\t\t\t\t// the framework should release the remaining unclear resources\n+\t\t\t\t\tfor (ResourceID id : toHandle) {\n+\t\t\t\t\t\treleasePendingWorker(id);\n+\t\t\t\t\t}\n+\t\t\t\t}\n+\n+\t\t\t}\n+\n+\t\t\t// trigger initial check for requesting new workers\n+\t\t\tcheckWorkersPool();\n+\n+\t\t} else {\n+\t\t\tString msg = \"Attempting to associate with new JobManager leader \" + newJobManagerLeader\n+\t\t\t\t+ \" without previously disassociating from current leader \" + jobManager;\n+\t\t\tfatalError(msg, new Exception(msg));\n+\t\t}\n+\t}\n+\n+\t// ------------------------------------------------------------------------\n+\t//  Cluster Shutdown\n+\t// ------------------------------------------------------------------------\n+\n+\tprivate void shutdownCluster(ApplicationStatus status, String diagnostics) {\n+\t\tLOG.info(\"Shutting down cluster with status {} : {}\", status, diagnostics);\n+\n+\t\tshutdownApplication(status, diagnostics);\n+\t}\n+\n+\t// ------------------------------------------------------------------------\n+\t//  Worker pool size management\n+\t// ------------------------------------------------------------------------\n+\n+\t/**\n+\t * This method causes the resource framework master to <b>synchronously</b>re-examine\n+\t * the set of available and pending workers containers, and allocate containers\n+\t * if needed.\n+\t *\n+\t * This method does not automatically release workers, because it is not visible to\n+\t * this resource master which workers can be released. Instead, the JobManager must\n+\t * explicitly release individual workers.\n+\t */\n+\tprivate void checkWorkersPool() {\n+\t\tint numWorkersPending = getNumWorkerRequestsPending();\n+\t\tint numWorkersPendingRegistration = getNumWorkersPendingRegistration();\n+\n+\t\t// sanity checks\n+\t\tPreconditions.checkState(numWorkersPending >= 0,\n+\t\t\t\"Number of pending workers should never be below 0.\");\n+\t\tPreconditions.checkState(numWorkersPendingRegistration >= 0,\n+\t\t\t\"Number of pending workers pending registration should never be below 0.\");\n+\n+\t\t// see how many workers we want, and whether we have enough\n+\t\tint allAvailableAndPending = registeredWorkers.size() +\n+\t\t\tnumWorkersPending + numWorkersPendingRegistration;\n+\n+\t\tint missing = designatedPoolSize - allAvailableAndPending;\n+\n+\t\tif (missing > 0) {\n+\t\t\trequestNewWorkers(missing);\n+\t\t}\n+\t}\n+\n+\t/**\n+\t * Sets the designated worker pool size. If this size is larger than the current pool\n+\t * size, then the resource manager will try to acquire more TaskManagers.\n+\t *\n+\t * @param num The number of workers in the pool.\n+\t */\n+\tprivate void adjustDesignatedNumberOfWorkers(int num) {\n+\t\tif (num >= 0) {\n+\t\t\tlog.info(\"Adjusting designated worker pool size to {}\", num);\n+\t\t\tdesignatedPoolSize = num;\n+\t\t\tcheckWorkersPool();\n+\t\t} else {\n+\t\t\tlog.warn(\"Ignoring invalid designated worker pool size: \" + num);\n+\t\t}\n+\t}\n+\n+\t// ------------------------------------------------------------------------\n+\t//  Callbacks\n+\t// ------------------------------------------------------------------------\n+\n+\t/**\n+\t * This method causes the resource framework master to <b>asynchronously</b>re-examine\n+\t * the set of available and pending workers containers, and release or allocate\n+\t * containers if needed. The method sends an actor message which will trigger the\n+\t * re-examination.\n+\t */\n+\tpublic void triggerCheckWorkers() {\n+\t\tself().tell(\n+\t\t\tdecorateMessage(\n+\t\t\t\tCheckAndAllocateContainers.get()),\n+\t\t\tActorRef.noSender());\n+\t}\n+\n+\t/**\n+\t * This method should be called by the framework once it detects that a currently registered\n+\t * worker has failed.\n+\t *\n+\t * @param resourceID Id of the worker that has failed.\n+\t * @param message An informational message that explains why the worker failed.\n+\t */\n+\tpublic void notifyWorkerFailed(ResourceID resourceID, String message) {\n+\t\tWorkerType worker = registeredWorkers.remove(resourceID);\n+\t\tif (worker != null) {\n+\t\t\tjobManager.tell(\n+\t\t\t\tdecorateMessage(\n+\t\t\t\t\tnew ResourceRemoved(resourceID, message)),\n+\t\t\t\tself());\n+\t\t}\n+\t}\n+\n+\t// ------------------------------------------------------------------------\n+\t//  Framework specific behavior\n+\t// ------------------------------------------------------------------------\n+\n+\t/**\n+\t * Initializes the framework specific components.\n+\t *\n+\t * @throws Exception Exceptions during initialization cause the resource manager to fail.\n+\t *                   If the framework is able to recover this resource manager, it will be\n+\t *                   restarted.\n+\t */\n+\tprotected abstract void initialize() throws Exception;\n+\n+\t/**\n+\t * Provides codes to handle an update of the leader (relevant for HA). The framework has to deal\n+\t * with the consequences of a leader update.\n+\t */\n+\tprotected abstract void leaderUpdated();\n+\n+\t/**\n+\t * The framework specific code for shutting down the application. This should report the\n+\t * application's final status and shut down the resource manager cleanly.\n+\t *\n+\t * This method also needs to make sure all pending containers that are not registered\n+\t * yet are returned.\n+\t *\n+\t * @param finalStatus The application status to report.\n+\t * @param optionalDiagnostics An optional diagnostics message.\n+\t */\n+\tprotected abstract void shutdownApplication(ApplicationStatus finalStatus, String optionalDiagnostics);\n+\n+\t/**\n+\t * Notifies the resource master of a fatal error.\n+\t *\n+\t * <p><b>IMPORTANT:</b> This should not cleanly shut down this master, but exit it in\n+\t * such a way that a high-availability setting would restart this or fail over\n+\t * to another master.\n+\t */\n+\tprotected abstract void fatalError(String message, Throwable error);\n+\n+\t/**\n+\t * Requests to allocate a certain number of new workers.\n+\t *\n+\t * @param numWorkers The number of workers to allocate.\n+\t */\n+\tprotected abstract void requestNewWorkers(int numWorkers);\n+\n+\t/**\n+\t * Trigger a release of a pending worker.\n+\t * @param resourceID The worker resource id\n+\t */\n+\tprotected abstract void releasePendingWorker(ResourceID resourceID);\n+\n+\t/**\n+\t * Trigger a release of a registered worker.\n+\t * @param resourceID The worker resource id\n+\t */\n+\tprotected abstract void releaseRegisteredWorker(WorkerType resourceID);\n+\n+\t/**\n+\t * Callback when a worker was registered.\n+\t * @param resourceID The worker resource id\n+\t */\n+\tprotected abstract WorkerType workerRegistered(ResourceID resourceID) throws Exception;\n+\n+\t/**\n+\t * This method is called when the resource manager starts after a failure and reconnects to\n+\t * the leader JobManager, who still has some workers registered. The method is used to consolidate\n+\t * the view between resource manager and JobManager. The resource manager gets the list of TaskManagers\n+\t * that the JobManager considers available and should return a list or nodes that the\n+\t * resource manager considers available.\n+\t *\n+\t * After that, the JobManager is informed of loss of all TaskManagers that are not part of the\n+\t * returned list.\n+\t *\n+\t * It is possible that the resource manager initially confirms some TaskManagers to be alive, even\n+\t * through they are in an uncertain status, if it later sends necessary failure notifications\n+\t * via calling {@link #notifyWorkerFailed(ResourceID, String)}.\n+\t *\n+\t * @param registered The list of TaskManagers that the JobManager knows.\n+\t * @return The subset of TaskManagers that the resource manager can confirm to be alive.\n+\t */\n+\tprotected abstract Collection<WorkerType> reacceptRegisteredWorkers(Collection<ResourceID> registered);\n+\n+\t/**\n+\t * Gets the number of requested workers that have not yet been granted.\n+\t *\n+\t * @return The number pending worker requests. Must never be smaller than 0.\n+\t */\n+\tprotected abstract int getNumWorkerRequestsPending();\n+\n+\t/**\n+\t * Gets the number of containers that have been started, but where the TaskManager\n+\t * has not yet registered at the job manager.\n+\t *\n+\t * @return The number of started containers pending TaskManager registration.\n+\t * Must never be smaller than 0.\n+\t */\n+\tprotected abstract int getNumWorkersPendingRegistration();\n+\n+\t// ------------------------------------------------------------------------\n+\t//  Info messaging\n+\t// ------------------------------------------------------------------------\n+\n+\tprotected void sendInfoMessage(String message) {\n+\t\tfor (ActorRef listener : infoMessageListeners) {\n+\t\t\tlistener.tell(decorateMessage(new InfoMessage(message)), self());\n+\t\t}\n+\t}\n+\n+\n+\t// ------------------------------------------------------------------------\n+\t//  Startup\n+\t// ------------------------------------------------------------------------\n+\n+\tpublic static void startResourceManager() {\n+\n+\t}\n+\n+\t/**\n+\t * Starts the resource manager actors.\n+\t * @param configuration The configuration for the resource manager\n+\t * @param actorSystem The actor system to start the resource manager in\n+\t * @param leaderRetriever The leader retriever service to intialize the resource manager\n+\t * @param resourceManagerClass The class of the ResourceManager to be started\n+\t * @return ActorRef of the resource manager\n+\t */\n+\tpublic static ActorRef startResourceManagerActors(\n+\t\tConfiguration configuration,\n+\t\tActorSystem actorSystem,\n+\t\tLeaderRetrievalService leaderRetriever,\n+\t\tClass<? extends FlinkResourceManager<?>> resourceManagerClass\n+\t) {","path":"flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/FlinkResourceManager.java","position":769,"original_position":769,"commit_id":"424c9f47d7c06cd5dfe33647b323bf10010293ed","original_commit_id":"424c9f47d7c06cd5dfe33647b323bf10010293ed","user":{"login":"tillrohrmann","id":5756858,"avatar_url":"https://avatars.githubusercontent.com/u/5756858?v=3","gravatar_id":"","url":"https://api.github.com/users/tillrohrmann","html_url":"https://github.com/tillrohrmann","followers_url":"https://api.github.com/users/tillrohrmann/followers","following_url":"https://api.github.com/users/tillrohrmann/following{/other_user}","gists_url":"https://api.github.com/users/tillrohrmann/gists{/gist_id}","starred_url":"https://api.github.com/users/tillrohrmann/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tillrohrmann/subscriptions","organizations_url":"https://api.github.com/users/tillrohrmann/orgs","repos_url":"https://api.github.com/users/tillrohrmann/repos","events_url":"https://api.github.com/users/tillrohrmann/events{/privacy}","received_events_url":"https://api.github.com/users/tillrohrmann/received_events","type":"User","site_admin":false},"body":"one line break too much","created_at":"2016-03-16T16:43:40Z","updated_at":"2016-03-16T16:43:40Z","html_url":"https://github.com/apache/flink/pull/1741#discussion_r56368853","pull_request_url":"https://api.github.com/repos/apache/flink/pulls/1741","_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/comments/56368853"},"html":{"href":"https://github.com/apache/flink/pull/1741#discussion_r56368853"},"pull_request":{"href":"https://api.github.com/repos/apache/flink/pulls/1741"}}},"pull_request":{"url":"https://api.github.com/repos/apache/flink/pulls/1741","id":61116269,"html_url":"https://github.com/apache/flink/pull/1741","diff_url":"https://github.com/apache/flink/pull/1741.diff","patch_url":"https://github.com/apache/flink/pull/1741.patch","issue_url":"https://api.github.com/repos/apache/flink/issues/1741","number":1741,"state":"open","locked":false,"title":"[FLINK-3544] Introduce ResourceManager component","user":{"login":"mxm","id":837221,"avatar_url":"https://avatars.githubusercontent.com/u/837221?v=3","gravatar_id":"","url":"https://api.github.com/users/mxm","html_url":"https://github.com/mxm","followers_url":"https://api.github.com/users/mxm/followers","following_url":"https://api.github.com/users/mxm/following{/other_user}","gists_url":"https://api.github.com/users/mxm/gists{/gist_id}","starred_url":"https://api.github.com/users/mxm/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mxm/subscriptions","organizations_url":"https://api.github.com/users/mxm/orgs","repos_url":"https://api.github.com/users/mxm/repos","events_url":"https://api.github.com/users/mxm/events{/privacy}","received_events_url":"https://api.github.com/users/mxm/received_events","type":"User","site_admin":false},"body":"So far the JobManager has been the central instance which is responsible for resource management and allocation.\r\n\r\nWhile thinking about how to integrate Mesos support in Flink, people from the Flink community realized that it would be nice to delegate resource allocation to a dedicated process. This process may run independently of the JobManager which is a requirement for proper integration of cluster allocation frameworks like Mesos.\r\n\r\nThis has led to the idea of creating a new component called the ResourceManager. Its task is to allocate and maintain resources requested by the JobManager. The ResourceManager has a very abstract notion of resources.\r\n\r\nInitially, we thought we could make the ResourceManager deal with resource allocation and the registration/supervision of the TaskManagers. However, this approach proved to add unnecessary complexity to the runtime. Registration state of TaskManagers had to be kept in sync at both the JobManager and the ResourceManager.\r\n\r\nThat's why @StephanEwen and me changed the ResourceManager's role to simply deal with the resource acquisition. The TaskManagers still register with the JobManager which informs the ResourceManager about the successful registration of a TaskManager. The ResourceManager may inform the JobManager of failed TaskManagers. Due to the insight which the ResourceManager has in the resource health, it may detect failed TaskManagers much earlier than the heartbeat-based monitoring of the JobManager.\r\n\r\nAt this stage, the ResourceManager is an optional component. That means the JobManager doesn't depend on the ResourceManager as long as it has enough resources to perform the computation. All bookkeeping is performed by the JobManager. When the ResourceManager connects to the JobManager, it receives the current resources, i.e. task manager instances, and allocates more containers if necessary. The JobManager adjusts the number of containers through the SetWorkerPoolSize method.\r\n\r\nIn standalone mode, the ResourceManager may be deactivated or simply use the StandaloneResourceManager which does practically nothing because we don't need to allocate resources in standalone mode.\r\n\r\nIn YARN mode, the ResourceManager takes care of communicating with the Yarn resource manager. When containers fail, it informs the JobManager and tries to allocate new containers. The ResourceManager runs as an actor within the same actor system as the JobManager. It could, however, also run independently. The independent mode would be the default behavior for Mesos where the framework master is expected to just deal with resource allocation.\r\n\r\nThe attached figures depict the message flow between ResourceManager, JobManager, and TaskManager.\r\n\r\n[ResourceManagerSketch.pdf](https://github.com/apache/flink/files/151425/ResourceManagerSketch.pdf)\r\n","created_at":"2016-02-29T17:47:17Z","updated_at":"2016-03-16T16:43:40Z","closed_at":null,"merged_at":null,"merge_commit_sha":"27d4fde070ba7c31675bbf6069741486d7251e03","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/flink/pulls/1741/commits","review_comments_url":"https://api.github.com/repos/apache/flink/pulls/1741/comments","review_comment_url":"https://api.github.com/repos/apache/flink/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/flink/issues/1741/comments","statuses_url":"https://api.github.com/repos/apache/flink/statuses/424c9f47d7c06cd5dfe33647b323bf10010293ed","head":{"label":"mxm:resourceManager-v2-squashed","ref":"resourceManager-v2-squashed","sha":"424c9f47d7c06cd5dfe33647b323bf10010293ed","user":{"login":"mxm","id":837221,"avatar_url":"https://avatars.githubusercontent.com/u/837221?v=3","gravatar_id":"","url":"https://api.github.com/users/mxm","html_url":"https://github.com/mxm","followers_url":"https://api.github.com/users/mxm/followers","following_url":"https://api.github.com/users/mxm/following{/other_user}","gists_url":"https://api.github.com/users/mxm/gists{/gist_id}","starred_url":"https://api.github.com/users/mxm/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mxm/subscriptions","organizations_url":"https://api.github.com/users/mxm/orgs","repos_url":"https://api.github.com/users/mxm/repos","events_url":"https://api.github.com/users/mxm/events{/privacy}","received_events_url":"https://api.github.com/users/mxm/received_events","type":"User","site_admin":false},"repo":{"id":25965625,"name":"flink","full_name":"mxm/flink","owner":{"login":"mxm","id":837221,"avatar_url":"https://avatars.githubusercontent.com/u/837221?v=3","gravatar_id":"","url":"https://api.github.com/users/mxm","html_url":"https://github.com/mxm","followers_url":"https://api.github.com/users/mxm/followers","following_url":"https://api.github.com/users/mxm/following{/other_user}","gists_url":"https://api.github.com/users/mxm/gists{/gist_id}","starred_url":"https://api.github.com/users/mxm/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mxm/subscriptions","organizations_url":"https://api.github.com/users/mxm/orgs","repos_url":"https://api.github.com/users/mxm/repos","events_url":"https://api.github.com/users/mxm/events{/privacy}","received_events_url":"https://api.github.com/users/mxm/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/mxm/flink","description":"Mirror of Apache Flink","fork":true,"url":"https://api.github.com/repos/mxm/flink","forks_url":"https://api.github.com/repos/mxm/flink/forks","keys_url":"https://api.github.com/repos/mxm/flink/keys{/key_id}","collaborators_url":"https://api.github.com/repos/mxm/flink/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/mxm/flink/teams","hooks_url":"https://api.github.com/repos/mxm/flink/hooks","issue_events_url":"https://api.github.com/repos/mxm/flink/issues/events{/number}","events_url":"https://api.github.com/repos/mxm/flink/events","assignees_url":"https://api.github.com/repos/mxm/flink/assignees{/user}","branches_url":"https://api.github.com/repos/mxm/flink/branches{/branch}","tags_url":"https://api.github.com/repos/mxm/flink/tags","blobs_url":"https://api.github.com/repos/mxm/flink/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/mxm/flink/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/mxm/flink/git/refs{/sha}","trees_url":"https://api.github.com/repos/mxm/flink/git/trees{/sha}","statuses_url":"https://api.github.com/repos/mxm/flink/statuses/{sha}","languages_url":"https://api.github.com/repos/mxm/flink/languages","stargazers_url":"https://api.github.com/repos/mxm/flink/stargazers","contributors_url":"https://api.github.com/repos/mxm/flink/contributors","subscribers_url":"https://api.github.com/repos/mxm/flink/subscribers","subscription_url":"https://api.github.com/repos/mxm/flink/subscription","commits_url":"https://api.github.com/repos/mxm/flink/commits{/sha}","git_commits_url":"https://api.github.com/repos/mxm/flink/git/commits{/sha}","comments_url":"https://api.github.com/repos/mxm/flink/comments{/number}","issue_comment_url":"https://api.github.com/repos/mxm/flink/issues/comments{/number}","contents_url":"https://api.github.com/repos/mxm/flink/contents/{+path}","compare_url":"https://api.github.com/repos/mxm/flink/compare/{base}...{head}","merges_url":"https://api.github.com/repos/mxm/flink/merges","archive_url":"https://api.github.com/repos/mxm/flink/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/mxm/flink/downloads","issues_url":"https://api.github.com/repos/mxm/flink/issues{/number}","pulls_url":"https://api.github.com/repos/mxm/flink/pulls{/number}","milestones_url":"https://api.github.com/repos/mxm/flink/milestones{/number}","notifications_url":"https://api.github.com/repos/mxm/flink/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/mxm/flink/labels{/name}","releases_url":"https://api.github.com/repos/mxm/flink/releases{/id}","deployments_url":"https://api.github.com/repos/mxm/flink/deployments","created_at":"2014-10-30T10:24:48Z","updated_at":"2015-04-29T12:00:18Z","pushed_at":"2016-03-16T13:06:11Z","git_url":"git://github.com/mxm/flink.git","ssh_url":"git@github.com:mxm/flink.git","clone_url":"https://github.com/mxm/flink.git","svn_url":"https://github.com/mxm/flink","homepage":null,"size":85817,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"543c04949479191ddf4e7ec6035fce6a01e8c5a0","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":20587599,"name":"flink","full_name":"apache/flink","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/flink","description":"Mirror of Apache Flink","fork":false,"url":"https://api.github.com/repos/apache/flink","forks_url":"https://api.github.com/repos/apache/flink/forks","keys_url":"https://api.github.com/repos/apache/flink/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/flink/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/flink/teams","hooks_url":"https://api.github.com/repos/apache/flink/hooks","issue_events_url":"https://api.github.com/repos/apache/flink/issues/events{/number}","events_url":"https://api.github.com/repos/apache/flink/events","assignees_url":"https://api.github.com/repos/apache/flink/assignees{/user}","branches_url":"https://api.github.com/repos/apache/flink/branches{/branch}","tags_url":"https://api.github.com/repos/apache/flink/tags","blobs_url":"https://api.github.com/repos/apache/flink/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/flink/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/flink/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/flink/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/flink/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/flink/languages","stargazers_url":"https://api.github.com/repos/apache/flink/stargazers","contributors_url":"https://api.github.com/repos/apache/flink/contributors","subscribers_url":"https://api.github.com/repos/apache/flink/subscribers","subscription_url":"https://api.github.com/repos/apache/flink/subscription","commits_url":"https://api.github.com/repos/apache/flink/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/flink/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/flink/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/flink/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/flink/contents/{+path}","compare_url":"https://api.github.com/repos/apache/flink/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/flink/merges","archive_url":"https://api.github.com/repos/apache/flink/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/flink/downloads","issues_url":"https://api.github.com/repos/apache/flink/issues{/number}","pulls_url":"https://api.github.com/repos/apache/flink/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/flink/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/flink/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/flink/labels{/name}","releases_url":"https://api.github.com/repos/apache/flink/releases{/id}","deployments_url":"https://api.github.com/repos/apache/flink/deployments","created_at":"2014-06-07T07:00:10Z","updated_at":"2016-03-16T14:17:39Z","pushed_at":"2016-03-16T16:32:57Z","git_url":"git://github.com/apache/flink.git","ssh_url":"git@github.com:apache/flink.git","clone_url":"https://github.com/apache/flink.git","svn_url":"https://github.com/apache/flink","homepage":null,"size":112590,"stargazers_count":1120,"watchers_count":1120,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":691,"mirror_url":"git://git.apache.org/flink.git","open_issues_count":82,"forks":691,"open_issues":82,"watchers":1120,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/1741"},"html":{"href":"https://github.com/apache/flink/pull/1741"},"issue":{"href":"https://api.github.com/repos/apache/flink/issues/1741"},"comments":{"href":"https://api.github.com/repos/apache/flink/issues/1741/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/flink/pulls/1741/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/flink/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/flink/pulls/1741/commits"},"statuses":{"href":"https://api.github.com/repos/apache/flink/statuses/424c9f47d7c06cd5dfe33647b323bf10010293ed"}}}}