http://git-wip-us.apache.org/repos/asf/flink/blob/2f12ba3e/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/SlotRequest.java
----------------------------------------------------------------------
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/SlotRequest.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/SlotRequest.java
new file mode 100644
index 0000000..896421b
--- /dev/null
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/SlotRequest.java
@@ -0,0 +1,74 @@
+/*
+ * 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.runtime.resourcemanager;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+
+import java.io.Serializable;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * This describes the requirement of the slot, mainly used by JobManager 
requesting slot from ResourceManager.
+ */
+public class SlotRequest implements Serializable {
+
+       private static final long serialVersionUID = -6586877187990445986L;
+
+       /** The JobID of the slot requested for */
+       private final JobID jobId;
+
+       /** The unique identification of this request */
+       private final AllocationID allocationId;
+
+       /** The resource profile of the required slot */
+       private final ResourceProfile resourceProfile;
+
+       public SlotRequest(JobID jobId, AllocationID allocationId, 
ResourceProfile resourceProfile) {
+               this.jobId = checkNotNull(jobId);
+               this.allocationId = checkNotNull(allocationId);
+               this.resourceProfile = checkNotNull(resourceProfile);
+       }
+
+       /**
+        * Get the JobID of the slot requested for.
+        * @return The job id
+        */
+       public JobID getJobId() {
+               return jobId;
+       }
+
+       /**
+        * Get the unique identification of this request
+        * @return the allocation id
+        */
+       public AllocationID getAllocationId() {
+               return allocationId;
+       }
+
+       /**
+        * Get the resource profile of the desired slot
+        * @return The resource profile
+        */
+       public ResourceProfile getResourceProfile() {
+               return resourceProfile;
+       }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/2f12ba3e/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/jobmaster/JobMaster.java
----------------------------------------------------------------------
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/jobmaster/JobMaster.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/jobmaster/JobMaster.java
deleted file mode 100644
index a046cb8..0000000
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/jobmaster/JobMaster.java
+++ /dev/null
@@ -1,244 +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.runtime.rpc.jobmaster;
-
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
-import org.apache.flink.runtime.jobgraph.JobGraph;
-import org.apache.flink.runtime.leaderelection.LeaderContender;
-import org.apache.flink.runtime.leaderelection.LeaderElectionService;
-import org.apache.flink.runtime.messages.Acknowledge;
-import org.apache.flink.runtime.rpc.RpcMethod;
-import org.apache.flink.runtime.rpc.resourcemanager.ResourceManagerGateway;
-import org.apache.flink.runtime.rpc.RpcEndpoint;
-import org.apache.flink.runtime.rpc.RpcService;
-import org.apache.flink.runtime.taskmanager.TaskExecutionState;
-import org.apache.flink.util.Preconditions;
-
-import java.util.UUID;
-
-/**
- * JobMaster implementation. The job master is responsible for the execution 
of a single
- * {@link org.apache.flink.runtime.jobgraph.JobGraph}.
- * <p>
- * It offers the following methods as part of its rpc interface to interact 
with the JobMaster
- * remotely:
- * <ul>
- *     <li>{@link #updateTaskExecutionState(TaskExecutionState)} updates the 
task execution state for
- * given task</li>
- * </ul>
- */
-public class JobMaster extends RpcEndpoint<JobMasterGateway> {
-
-       /** Gateway to connected resource manager, null iff not connected */
-       private ResourceManagerGateway resourceManager = null;
-
-       /** Logical representation of the job */
-       private final JobGraph jobGraph;
-       private final JobID jobID;
-
-       /** Configuration of the job */
-       private final Configuration configuration;
-
-       /** Service to contend for and retrieve the leadership of JM and RM */
-       private final HighAvailabilityServices highAvailabilityServices;
-
-       /** Leader Management */
-       private LeaderElectionService leaderElectionService = null;
-       private UUID leaderSessionID;
-
-       /**
-        * The JM's Constructor
-        *
-        * @param jobGraph The representation of the job's execution plan
-        * @param configuration The job's configuration
-        * @param rpcService The RPC service at which the JM serves
-        * @param highAvailabilityService The cluster's HA service from the JM 
can elect and retrieve leaders.
-        */
-       public JobMaster(
-               JobGraph jobGraph,
-               Configuration configuration,
-               RpcService rpcService,
-               HighAvailabilityServices highAvailabilityService) {
-
-               super(rpcService);
-
-               this.jobGraph = Preconditions.checkNotNull(jobGraph);
-               this.jobID = Preconditions.checkNotNull(jobGraph.getJobID());
-
-               this.configuration = Preconditions.checkNotNull(configuration);
-
-               this.highAvailabilityServices = 
Preconditions.checkNotNull(highAvailabilityService);
-       }
-
-       public ResourceManagerGateway getResourceManager() {
-               return resourceManager;
-       }
-
-       
//----------------------------------------------------------------------------------------------
-       // Initialization methods
-       
//----------------------------------------------------------------------------------------------
-       public void start() {
-               super.start();
-
-               // register at the election once the JM starts
-               registerAtElectionService();
-       }
-
-
-       
//----------------------------------------------------------------------------------------------
-       // JobMaster Leadership methods
-       
//----------------------------------------------------------------------------------------------
-
-       /**
-        * Retrieves the election service and contend for the leadership.
-        */
-       private void registerAtElectionService() {
-               try {
-                       leaderElectionService = 
highAvailabilityServices.getJobMasterLeaderElectionService(jobID);
-                       leaderElectionService.start(new 
JobMasterLeaderContender());
-               } catch (Exception e) {
-                       throw new RuntimeException("Fail to register at the 
election of JobMaster", e);
-               }
-       }
-
-       /**
-        * Start the execution when the leadership is granted.
-        *
-        * @param newLeaderSessionID The identifier of the new leadership 
session
-        */
-       public void grantJobMasterLeadership(final UUID newLeaderSessionID) {
-               runAsync(new Runnable() {
-                       @Override
-                       public void run() {
-                               log.info("JobManager {} grants leadership with 
session id {}.", getAddress(), newLeaderSessionID);
-
-                               // The operation may be blocking, but since JM 
is idle before it grants the leadership, it's okay that
-                               // JM waits here for the operation's 
completeness.
-                               leaderSessionID = newLeaderSessionID;
-                               
leaderElectionService.confirmLeaderSessionID(newLeaderSessionID);
-
-                               // TODO:: execute the job when the leadership 
is granted.
-                       }
-               });
-       }
-
-       /**
-        * Stop the execution when the leadership is revoked.
-        */
-       public void revokeJobMasterLeadership() {
-               runAsync(new Runnable() {
-                       @Override
-                       public void run() {
-                               log.info("JobManager {} was revoked 
leadership.", getAddress());
-
-                               // TODO:: cancel the job's execution and notify 
all listeners
-                               cancelAndClearEverything(new 
Exception("JobManager is no longer the leader."));
-
-                               leaderSessionID = null;
-                       }
-               });
-       }
-
-       /**
-        * Handles error occurring in the leader election service
-        *
-        * @param exception Exception thrown in the leader election service
-        */
-       public void onJobMasterElectionError(final Exception exception) {
-               runAsync(new Runnable() {
-                       @Override
-                       public void run() {
-                               log.error("Received an error from the 
LeaderElectionService.", exception);
-
-                               // TODO:: cancel the job's execution and 
shutdown the JM
-                               cancelAndClearEverything(exception);
-
-                               leaderSessionID = null;
-                       }
-               });
-
-       }
-
-       
//----------------------------------------------------------------------------------------------
-       // RPC methods
-       
//----------------------------------------------------------------------------------------------
-
-       /**
-        * Updates the task execution state for a given task.
-        *
-        * @param taskExecutionState New task execution state for a given task
-        * @return Acknowledge the task execution state update
-        */
-       @RpcMethod
-       public Acknowledge updateTaskExecutionState(TaskExecutionState 
taskExecutionState) {
-               System.out.println("TaskExecutionState: " + taskExecutionState);
-               return Acknowledge.get();
-       }
-
-       /**
-        * Triggers the registration of the job master at the resource manager.
-        *
-        * @param address Address of the resource manager
-        */
-       @RpcMethod
-       public void registerAtResourceManager(final String address) {
-               //TODO:: register at the RM
-       }
-
-       
//----------------------------------------------------------------------------------------------
-       // Helper methods
-       
//----------------------------------------------------------------------------------------------
-
-       /**
-        * Cancel the current job and notify all listeners the job's 
cancellation.
-        *
-        * @param cause Cause for the cancelling.
-        */
-       private void cancelAndClearEverything(Throwable cause) {
-               // currently, nothing to do here
-       }
-
-       // 
------------------------------------------------------------------------
-       //  Utility classes
-       // 
------------------------------------------------------------------------
-       private class JobMasterLeaderContender implements LeaderContender {
-
-               @Override
-               public void grantLeadership(UUID leaderSessionID) {
-                       
JobMaster.this.grantJobMasterLeadership(leaderSessionID);
-               }
-
-               @Override
-               public void revokeLeadership() {
-                       JobMaster.this.revokeJobMasterLeadership();
-               }
-
-               @Override
-               public String getAddress() {
-                       return JobMaster.this.getAddress();
-               }
-
-               @Override
-               public void handleError(Exception exception) {
-                       onJobMasterElectionError(exception);
-               }
-       }
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/2f12ba3e/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/jobmaster/JobMasterGateway.java
----------------------------------------------------------------------
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/jobmaster/JobMasterGateway.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/jobmaster/JobMasterGateway.java
deleted file mode 100644
index 17a4c3a..0000000
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/jobmaster/JobMasterGateway.java
+++ /dev/null
@@ -1,45 +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.runtime.rpc.jobmaster;
-
-import org.apache.flink.runtime.messages.Acknowledge;
-import org.apache.flink.runtime.rpc.RpcGateway;
-import org.apache.flink.runtime.taskmanager.TaskExecutionState;
-import scala.concurrent.Future;
-
-/**
- * {@link JobMaster} rpc gateway interface
- */
-public interface JobMasterGateway extends RpcGateway {
-
-       /**
-        * Updates the task execution state for a given task.
-        *
-        * @param taskExecutionState New task execution state for a given task
-        * @return Future acknowledge of the task execution state update
-        */
-       Future<Acknowledge> updateTaskExecutionState(TaskExecutionState 
taskExecutionState);
-
-       /**
-        * Triggers the registration of the job master at the resource manager.
-        *
-        * @param address Address of the resource manager
-        */
-       void registerAtResourceManager(final String address);
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/2f12ba3e/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/registration/RegistrationResponse.java
----------------------------------------------------------------------
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/registration/RegistrationResponse.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/registration/RegistrationResponse.java
deleted file mode 100644
index 2de560a..0000000
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/registration/RegistrationResponse.java
+++ /dev/null
@@ -1,84 +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.runtime.rpc.registration;
-
-import java.io.Serializable;
-
-/**
- * Base class for responses given to registration attempts from {@link 
RetryingRegistration}.
- */
-public abstract class RegistrationResponse implements Serializable {
-
-       private static final long serialVersionUID = 1L;
-
-       // 
----------------------------------------------------------------------------
-       
-       /**
-        * Base class for a successful registration. Concrete registration 
implementations
-        * will typically extend this class to attach more information.
-        */
-       public static class Success extends RegistrationResponse {
-               private static final long serialVersionUID = 1L;
-               
-               @Override
-               public String toString() {
-                       return "Registration Successful";
-               }
-       }
-
-       // 
----------------------------------------------------------------------------
-
-       /**
-        * A rejected (declined) registration.
-        */
-       public static final class Decline extends RegistrationResponse {
-               private static final long serialVersionUID = 1L;
-
-               /** the rejection reason */
-               private final String reason;
-
-               /**
-                * Creates a new rejection message.
-                * 
-                * @param reason The reason for the rejection.
-                */
-               public Decline(String reason) {
-                       this.reason = reason != null ? reason : "(unknown)";
-               }
-
-               /**
-                * Gets the reason for the rejection.
-                */
-               public String getReason() {
-                       return reason;
-               }
-
-               @Override
-               public String toString() {
-                       return "Registration Declined (" + reason + ')';
-               }
-       }
-}
-
-
-
-
-
-
-

http://git-wip-us.apache.org/repos/asf/flink/blob/2f12ba3e/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/registration/RetryingRegistration.java
----------------------------------------------------------------------
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/registration/RetryingRegistration.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/registration/RetryingRegistration.java
deleted file mode 100644
index dcb5011..0000000
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/registration/RetryingRegistration.java
+++ /dev/null
@@ -1,296 +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.runtime.rpc.registration;
-
-import akka.dispatch.OnFailure;
-import akka.dispatch.OnSuccess;
-
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.runtime.rpc.RpcGateway;
-import org.apache.flink.runtime.rpc.RpcService;
-
-import org.slf4j.Logger;
-
-import scala.concurrent.Future;
-import scala.concurrent.Promise;
-import scala.concurrent.impl.Promise.DefaultPromise;
-
-import java.util.UUID;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
-import static org.apache.flink.util.Preconditions.checkArgument;
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
-
-/**
- * This utility class implements the basis of registering one component at 
another component,
- * for example registering the TaskExecutor at the ResourceManager.
- * This {@code RetryingRegistration} implements both the initial address 
resolution
- * and the retries-with-backoff strategy.
- * 
- * <p>The registration gives access to a future that is completed upon 
successful registration.
- * The registration can be canceled, for example when the target where it 
tries to register
- * at looses leader status.
- * 
- * @param <Gateway> The type of the gateway to connect to.
- * @param <Success> The type of the successful registration responses.
- */
-public abstract class RetryingRegistration<Gateway extends RpcGateway, Success 
extends RegistrationResponse.Success> {
-
-       // 
------------------------------------------------------------------------
-       //  default configuration values
-       // 
------------------------------------------------------------------------
-
-       /** default value for the initial registration timeout (milliseconds) */
-       private static final long INITIAL_REGISTRATION_TIMEOUT_MILLIS = 100;
-
-       /** default value for the maximum registration timeout, after 
exponential back-off (milliseconds) */
-       private static final long MAX_REGISTRATION_TIMEOUT_MILLIS = 30000;
-
-       /** The pause (milliseconds) made after an registration attempt caused 
an exception (other than timeout) */
-       private static final long ERROR_REGISTRATION_DELAY_MILLIS = 10000;
-
-       /** The pause (milliseconds) made after the registration attempt was 
refused */
-       private static final long REFUSED_REGISTRATION_DELAY_MILLIS = 30000;
-
-       // 
------------------------------------------------------------------------
-       // Fields
-       // 
------------------------------------------------------------------------
-
-       private final Logger log;
-
-       private final RpcService rpcService;
-
-       private final String targetName;
-
-       private final Class<Gateway> targetType;
-
-       private final String targetAddress;
-
-       private final UUID leaderId;
-
-       private final Promise<Tuple2<Gateway, Success>> completionPromise;
-
-       private final long initialRegistrationTimeout;
-
-       private final long maxRegistrationTimeout;
-
-       private final long delayOnError;
-
-       private final long delayOnRefusedRegistration;
-
-       private volatile boolean canceled;
-
-       // 
------------------------------------------------------------------------
-
-       public RetryingRegistration(
-                       Logger log,
-                       RpcService rpcService,
-                       String targetName,
-                       Class<Gateway> targetType,
-                       String targetAddress,
-                       UUID leaderId) {
-               this(log, rpcService, targetName, targetType, targetAddress, 
leaderId,
-                               INITIAL_REGISTRATION_TIMEOUT_MILLIS, 
MAX_REGISTRATION_TIMEOUT_MILLIS,
-                               ERROR_REGISTRATION_DELAY_MILLIS, 
REFUSED_REGISTRATION_DELAY_MILLIS);
-       }
-
-       public RetryingRegistration(
-                       Logger log,
-                       RpcService rpcService,
-                       String targetName, 
-                       Class<Gateway> targetType,
-                       String targetAddress,
-                       UUID leaderId,
-                       long initialRegistrationTimeout,
-                       long maxRegistrationTimeout,
-                       long delayOnError,
-                       long delayOnRefusedRegistration) {
-
-               checkArgument(initialRegistrationTimeout > 0, "initial 
registration timeout must be greater than zero");
-               checkArgument(maxRegistrationTimeout > 0, "maximum registration 
timeout must be greater than zero");
-               checkArgument(delayOnError >= 0, "delay on error must be 
non-negative");
-               checkArgument(delayOnRefusedRegistration >= 0, "delay on 
refused registration must be non-negative");
-
-               this.log = checkNotNull(log);
-               this.rpcService = checkNotNull(rpcService);
-               this.targetName = checkNotNull(targetName);
-               this.targetType = checkNotNull(targetType);
-               this.targetAddress = checkNotNull(targetAddress);
-               this.leaderId = checkNotNull(leaderId);
-               this.initialRegistrationTimeout = initialRegistrationTimeout;
-               this.maxRegistrationTimeout = maxRegistrationTimeout;
-               this.delayOnError = delayOnError;
-               this.delayOnRefusedRegistration = delayOnRefusedRegistration;
-
-               this.completionPromise = new DefaultPromise<>();
-       }
-
-       // 
------------------------------------------------------------------------
-       //  completion and cancellation
-       // 
------------------------------------------------------------------------
-
-       public Future<Tuple2<Gateway, Success>> getFuture() {
-               return completionPromise.future();
-       }
-
-       /**
-        * Cancels the registration procedure.
-        */
-       public void cancel() {
-               canceled = true;
-       }
-
-       /**
-        * Checks if the registration was canceled.
-        * @return True if the registration was canceled, false otherwise.
-        */
-       public boolean isCanceled() {
-               return canceled;
-       }
-
-       // 
------------------------------------------------------------------------
-       //  registration
-       // 
------------------------------------------------------------------------
-
-       protected abstract Future<RegistrationResponse> invokeRegistration(
-                       Gateway gateway, UUID leaderId, long timeoutMillis) 
throws Exception;
-
-       /**
-        * This method resolves the target address to a callable gateway and 
starts the
-        * registration after that.
-        */
-       @SuppressWarnings("unchecked")
-       public void startRegistration() {
-               try {
-                       // trigger resolution of the resource manager address 
to a callable gateway
-                       Future<Gateway> resourceManagerFuture = 
rpcService.connect(targetAddress, targetType);
-       
-                       // upon success, start the registration attempts
-                       resourceManagerFuture.onSuccess(new 
OnSuccess<Gateway>() {
-                               @Override
-                               public void onSuccess(Gateway result) {
-                                       log.info("Resolved {} address, 
beginning registration", targetName);
-                                       register(result, 1, 
initialRegistrationTimeout);
-                               }
-                       }, rpcService.getExecutionContext());
-       
-                       // upon failure, retry, unless this is cancelled
-                       resourceManagerFuture.onFailure(new OnFailure() {
-                               @Override
-                               public void onFailure(Throwable failure) {
-                                       if (!isCanceled()) {
-                                               log.warn("Could not resolve {} 
address {}, retrying...", targetName, targetAddress);
-                                               startRegistration();
-                                       }
-                               }
-                       }, rpcService.getExecutionContext());
-               }
-               catch (Throwable t) {
-                       cancel();
-                       completionPromise.tryFailure(t);
-               }
-       }
-
-       /**
-        * This method performs a registration attempt and triggers either a 
success notification or a retry,
-        * depending on the result.
-        */
-       @SuppressWarnings("unchecked")
-       private void register(final Gateway gateway, final int attempt, final 
long timeoutMillis) {
-               // eager check for canceling to avoid some unnecessary work
-               if (canceled) {
-                       return;
-               }
-
-               try {
-                       log.info("Registration at {} attempt {} 
(timeout={}ms)", targetName, attempt, timeoutMillis);
-                       Future<RegistrationResponse> registrationFuture = 
invokeRegistration(gateway, leaderId, timeoutMillis);
-       
-                       // if the registration was successful, let the 
TaskExecutor know
-                       registrationFuture.onSuccess(new 
OnSuccess<RegistrationResponse>() {
-                               
-                               @Override
-                               public void onSuccess(RegistrationResponse 
result) throws Throwable {
-                                       if (!isCanceled()) {
-                                               if (result instanceof 
RegistrationResponse.Success) {
-                                                       // registration 
successful!
-                                                       Success success = 
(Success) result;
-                                                       
completionPromise.success(new Tuple2<>(gateway, success));
-                                               }
-                                               else {
-                                                       // registration refused 
or unknown
-                                                       if (result instanceof 
RegistrationResponse.Decline) {
-                                                               
RegistrationResponse.Decline decline = (RegistrationResponse.Decline) result;
-                                                               
log.info("Registration at {} was declined: {}", targetName, 
decline.getReason());
-                                                       } else {
-                                                               
log.error("Received unknown response to registration attempt: " + result);
-                                                       }
-
-                                                       log.info("Pausing and 
re-attempting registration in {} ms", delayOnRefusedRegistration);
-                                                       registerLater(gateway, 
1, initialRegistrationTimeout, delayOnRefusedRegistration);
-                                               }
-                                       }
-                               }
-                       }, rpcService.getExecutionContext());
-       
-                       // upon failure, retry
-                       registrationFuture.onFailure(new OnFailure() {
-                               @Override
-                               public void onFailure(Throwable failure) {
-                                       if (!isCanceled()) {
-                                               if (failure instanceof 
TimeoutException) {
-                                                       // we simply have not 
received a response in time. maybe the timeout was
-                                                       // very low (initial 
fast registration attempts), maybe the target endpoint is
-                                                       // currently down.
-                                                       if 
(log.isDebugEnabled()) {
-                                                               
log.debug("Registration at {} ({}) attempt {} timed out after {} ms",
-                                                                               
targetName, targetAddress, attempt, timeoutMillis);
-                                                       }
-       
-                                                       long newTimeoutMillis = 
Math.min(2 * timeoutMillis, maxRegistrationTimeout);
-                                                       register(gateway, 
attempt + 1, newTimeoutMillis);
-                                               }
-                                               else {
-                                                       // a serious failure 
occurred. we still should not give up, but keep trying
-                                                       log.error("Registration 
at " + targetName + " failed due to an error", failure);
-                                                       log.info("Pausing and 
re-attempting registration in {} ms", delayOnError);
-       
-                                                       registerLater(gateway, 
1, initialRegistrationTimeout, delayOnError);
-                                               }
-                                       }
-                               }
-                       }, rpcService.getExecutionContext());
-               }
-               catch (Throwable t) {
-                       cancel();
-                       completionPromise.tryFailure(t);
-               }
-       }
-
-       private void registerLater(final Gateway gateway, final int attempt, 
final long timeoutMillis, long delay) {
-               rpcService.scheduleRunnable(new Runnable() {
-                       @Override
-                       public void run() {
-                               register(gateway, attempt, timeoutMillis);
-                       }
-               }, delay, TimeUnit.MILLISECONDS);
-       }
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/2f12ba3e/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/JobMasterRegistration.java
----------------------------------------------------------------------
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/JobMasterRegistration.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/JobMasterRegistration.java
deleted file mode 100644
index 7a2deae..0000000
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/JobMasterRegistration.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.rpc.resourcemanager;
-
-import java.io.Serializable;
-
-public class JobMasterRegistration implements Serializable {
-       private static final long serialVersionUID = 8411214999193765202L;
-
-       private final String address;
-
-       public JobMasterRegistration(String address) {
-               this.address = address;
-       }
-
-       public String getAddress() {
-               return address;
-       }
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/2f12ba3e/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/RegistrationResponse.java
----------------------------------------------------------------------
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/RegistrationResponse.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/RegistrationResponse.java
deleted file mode 100644
index 8ac9e49..0000000
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/RegistrationResponse.java
+++ /dev/null
@@ -1,43 +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.runtime.rpc.resourcemanager;
-
-import org.apache.flink.runtime.instance.InstanceID;
-
-import java.io.Serializable;
-
-public class RegistrationResponse implements Serializable {
-       private static final long serialVersionUID = -2379003255993119993L;
-
-       private final boolean isSuccess;
-       private final InstanceID instanceID;
-
-       public RegistrationResponse(boolean isSuccess, InstanceID instanceID) {
-               this.isSuccess = isSuccess;
-               this.instanceID = instanceID;
-       }
-
-       public boolean isSuccess() {
-               return isSuccess;
-       }
-
-       public InstanceID getInstanceID() {
-               return instanceID;
-       }
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/2f12ba3e/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/ResourceManager.java
----------------------------------------------------------------------
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/ResourceManager.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/ResourceManager.java
deleted file mode 100644
index f7147c9..0000000
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/ResourceManager.java
+++ /dev/null
@@ -1,214 +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.runtime.rpc.resourcemanager;
-
-import akka.dispatch.Mapper;
-
-import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.runtime.clusterframework.types.ResourceID;
-import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
-import org.apache.flink.runtime.instance.InstanceID;
-import org.apache.flink.runtime.leaderelection.LeaderContender;
-import org.apache.flink.runtime.leaderelection.LeaderElectionService;
-import org.apache.flink.runtime.rpc.RpcMethod;
-import org.apache.flink.runtime.rpc.RpcEndpoint;
-import org.apache.flink.runtime.rpc.RpcService;
-import org.apache.flink.runtime.rpc.jobmaster.JobMaster;
-import org.apache.flink.runtime.rpc.jobmaster.JobMasterGateway;
-import 
org.apache.flink.runtime.rpc.taskexecutor.TaskExecutorRegistrationSuccess;
-
-import scala.concurrent.Future;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.UUID;
-
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
-/**
- * ResourceManager implementation. The resource manager is responsible for 
resource de-/allocation
- * and bookkeeping.
- *
- * It offers the following methods as part of its rpc interface to interact 
with the him remotely:
- * <ul>
- *     <li>{@link #registerJobMaster(JobMasterRegistration)} registers a 
{@link JobMaster} at the resource manager</li>
- *     <li>{@link #requestSlot(SlotRequest)} requests a slot from the resource 
manager</li>
- * </ul>
- */
-public class ResourceManager extends RpcEndpoint<ResourceManagerGateway> {
-       private final Map<JobMasterGateway, InstanceID> jobMasterGateways;
-       private final HighAvailabilityServices highAvailabilityServices;
-       private LeaderElectionService leaderElectionService = null;
-       private UUID leaderSessionID = null;
-
-       public ResourceManager(RpcService rpcService, HighAvailabilityServices 
highAvailabilityServices) {
-               super(rpcService);
-               this.highAvailabilityServices = 
checkNotNull(highAvailabilityServices);
-               this.jobMasterGateways = new HashMap<>();
-       }
-
-       @Override
-       public void start() {
-               // start a leader
-               try {
-                       super.start();
-                       leaderElectionService = 
highAvailabilityServices.getResourceManagerLeaderElectionService();
-                       leaderElectionService.start(new 
ResourceManagerLeaderContender());
-               } catch (Throwable e) {
-                       log.error("A fatal error happened when starting the 
ResourceManager", e);
-                       throw new RuntimeException("A fatal error happened when 
starting the ResourceManager", e);
-               }
-       }
-
-       @Override
-       public void shutDown() {
-               try {
-                       leaderElectionService.stop();
-                       super.shutDown();
-               } catch(Throwable e) {
-                       log.error("A fatal error happened when shutdown the 
ResourceManager", e);
-                       throw new RuntimeException("A fatal error happened when 
shutdown the ResourceManager", e);
-               }
-       }
-
-       /**
-        * Gets the leader session id of current resourceManager.
-        *
-        * @return return the leaderSessionId of current resourceManager, this 
returns null until the current resourceManager is granted leadership.
-        */
-       @VisibleForTesting
-       UUID getLeaderSessionID() {
-               return leaderSessionID;
-       }
-
-       /**
-        * Register a {@link JobMaster} at the resource manager.
-        *
-        * @param jobMasterRegistration Job master registration information
-        * @return Future registration response
-        */
-       @RpcMethod
-       public Future<RegistrationResponse> 
registerJobMaster(JobMasterRegistration jobMasterRegistration) {
-               Future<JobMasterGateway> jobMasterFuture = 
getRpcService().connect(jobMasterRegistration.getAddress(), 
JobMasterGateway.class);
-
-               return jobMasterFuture.map(new Mapper<JobMasterGateway, 
RegistrationResponse>() {
-                       @Override
-                       public RegistrationResponse apply(final 
JobMasterGateway jobMasterGateway) {
-                               InstanceID instanceID;
-
-                               if 
(jobMasterGateways.containsKey(jobMasterGateway)) {
-                                       instanceID = 
jobMasterGateways.get(jobMasterGateway);
-                               } else {
-                                       instanceID = new InstanceID();
-                                       jobMasterGateways.put(jobMasterGateway, 
instanceID);
-                               }
-
-                               return new RegistrationResponse(true, 
instanceID);
-                       }
-               }, getMainThreadExecutionContext());
-       }
-
-       /**
-        * Requests a slot from the resource manager.
-        *
-        * @param slotRequest Slot request
-        * @return Slot assignment
-        */
-       @RpcMethod
-       public SlotAssignment requestSlot(SlotRequest slotRequest) {
-               System.out.println("SlotRequest: " + slotRequest);
-               return new SlotAssignment();
-       }
-
-
-       /**
-        *
-        * @param resourceManagerLeaderId  The fencing token for the 
ResourceManager leader 
-        * @param taskExecutorAddress      The address of the TaskExecutor that 
registers
-        * @param resourceID               The resource ID of the TaskExecutor 
that registers
-        *
-        * @return The response by the ResourceManager.
-        */
-       @RpcMethod
-       public org.apache.flink.runtime.rpc.registration.RegistrationResponse 
registerTaskExecutor(
-                       UUID resourceManagerLeaderId,
-                       String taskExecutorAddress,
-                       ResourceID resourceID) {
-
-               return new TaskExecutorRegistrationSuccess(new InstanceID(), 
5000);
-       }
-
-       private class ResourceManagerLeaderContender implements LeaderContender 
{
-
-               /**
-                * Callback method when current resourceManager is granted 
leadership
-                *
-                * @param leaderSessionID unique leadershipID
-                */
-               @Override
-               public void grantLeadership(final UUID leaderSessionID) {
-                       runAsync(new Runnable() {
-                               @Override
-                               public void run() {
-                                       log.info("ResourceManager {} was 
granted leadership with leader session ID {}", getAddress(), leaderSessionID);
-                                       ResourceManager.this.leaderSessionID = 
leaderSessionID;
-                                       // confirming the leader session ID 
might be blocking,
-                                       
leaderElectionService.confirmLeaderSessionID(leaderSessionID);
-                               }
-                       });
-               }
-
-               /**
-                * Callback method when current resourceManager lose leadership.
-                */
-               @Override
-               public void revokeLeadership() {
-                       runAsync(new Runnable() {
-                               @Override
-                               public void run() {
-                                       log.info("ResourceManager {} was 
revoked leadership.", getAddress());
-                                       jobMasterGateways.clear();
-                                       leaderSessionID = null;
-                               }
-                       });
-               }
-
-               @Override
-               public String getAddress() {
-                       return ResourceManager.this.getAddress();
-               }
-
-               /**
-                * Handles error occurring in the leader election service
-                *
-                * @param exception Exception being thrown in the leader 
election service
-                */
-               @Override
-               public void handleError(final Exception exception) {
-                       runAsync(new Runnable() {
-                               @Override
-                               public void run() {
-                                       log.error("ResourceManager received an 
error from the LeaderElectionService.", exception);
-                                       // terminate ResourceManager in case of 
an error
-                                       shutDown();
-                               }
-                       });
-               }
-       }
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/2f12ba3e/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/ResourceManagerGateway.java
----------------------------------------------------------------------
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/ResourceManagerGateway.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/ResourceManagerGateway.java
deleted file mode 100644
index afddb01..0000000
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/ResourceManagerGateway.java
+++ /dev/null
@@ -1,77 +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.runtime.rpc.resourcemanager;
-
-import org.apache.flink.runtime.clusterframework.types.ResourceID;
-import org.apache.flink.runtime.rpc.RpcGateway;
-import org.apache.flink.runtime.rpc.RpcTimeout;
-import org.apache.flink.runtime.rpc.jobmaster.JobMaster;
-
-import scala.concurrent.Future;
-import scala.concurrent.duration.FiniteDuration;
-
-import java.util.UUID;
-
-/**
- * The {@link ResourceManager}'s RPC gateway interface.
- */
-public interface ResourceManagerGateway extends RpcGateway {
-
-       /**
-        * Register a {@link JobMaster} at the resource manager.
-        *
-        * @param jobMasterRegistration Job master registration information
-        * @param timeout Timeout for the future to complete
-        * @return Future registration response
-        */
-       Future<RegistrationResponse> registerJobMaster(
-               JobMasterRegistration jobMasterRegistration,
-               @RpcTimeout FiniteDuration timeout);
-
-       /**
-        * Register a {@link JobMaster} at the resource manager.
-        *
-        * @param jobMasterRegistration Job master registration information
-        * @return Future registration response
-        */
-       Future<RegistrationResponse> registerJobMaster(JobMasterRegistration 
jobMasterRegistration);
-
-       /**
-        * Requests a slot from the resource manager.
-        *
-        * @param slotRequest Slot request
-        * @return Future slot assignment
-        */
-       Future<SlotAssignment> requestSlot(SlotRequest slotRequest);
-
-       /**
-        * 
-        * @param resourceManagerLeaderId  The fencing token for the 
ResourceManager leader 
-        * @param taskExecutorAddress      The address of the TaskExecutor that 
registers
-        * @param resourceID               The resource ID of the TaskExecutor 
that registers
-        * @param timeout                  The timeout for the response.
-        * 
-        * @return The future to the response by the ResourceManager.
-        */
-       Future<org.apache.flink.runtime.rpc.registration.RegistrationResponse> 
registerTaskExecutor(
-                       UUID resourceManagerLeaderId,
-                       String taskExecutorAddress,
-                       ResourceID resourceID,
-                       @RpcTimeout FiniteDuration timeout);
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/2f12ba3e/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/SlotAssignment.java
----------------------------------------------------------------------
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/SlotAssignment.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/SlotAssignment.java
deleted file mode 100644
index 86cd8b7..0000000
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/SlotAssignment.java
+++ /dev/null
@@ -1,25 +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.runtime.rpc.resourcemanager;
-
-import java.io.Serializable;
-
-public class SlotAssignment implements Serializable{
-       private static final long serialVersionUID = -6990813455942742322L;
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/2f12ba3e/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/SlotRequest.java
----------------------------------------------------------------------
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/SlotRequest.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/SlotRequest.java
deleted file mode 100644
index 74c7c39..0000000
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/SlotRequest.java
+++ /dev/null
@@ -1,74 +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.runtime.rpc.resourcemanager;
-
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.runtime.clusterframework.types.AllocationID;
-import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
-
-import java.io.Serializable;
-
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
-/**
- * This describes the requirement of the slot, mainly used by JobManager 
requesting slot from ResourceManager.
- */
-public class SlotRequest implements Serializable {
-
-       private static final long serialVersionUID = -6586877187990445986L;
-
-       /** The JobID of the slot requested for */
-       private final JobID jobId;
-
-       /** The unique identification of this request */
-       private final AllocationID allocationId;
-
-       /** The resource profile of the required slot */
-       private final ResourceProfile resourceProfile;
-
-       public SlotRequest(JobID jobId, AllocationID allocationId, 
ResourceProfile resourceProfile) {
-               this.jobId = checkNotNull(jobId);
-               this.allocationId = checkNotNull(allocationId);
-               this.resourceProfile = checkNotNull(resourceProfile);
-       }
-
-       /**
-        * Get the JobID of the slot requested for.
-        * @return The job id
-        */
-       public JobID getJobId() {
-               return jobId;
-       }
-
-       /**
-        * Get the unique identification of this request
-        * @return the allocation id
-        */
-       public AllocationID getAllocationId() {
-               return allocationId;
-       }
-
-       /**
-        * Get the resource profile of the desired slot
-        * @return The resource profile
-        */
-       public ResourceProfile getResourceProfile() {
-               return resourceProfile;
-       }
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/2f12ba3e/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/SlotReport.java
----------------------------------------------------------------------
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/SlotReport.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/SlotReport.java
deleted file mode 100644
index c372ecb..0000000
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/SlotReport.java
+++ /dev/null
@@ -1,56 +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.runtime.rpc.taskexecutor;
-
-import org.apache.flink.runtime.clusterframework.types.ResourceID;
-
-import java.io.Serializable;
-import java.util.List;
-
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
-/**
- * A report about the current status of all slots of the TaskExecutor, 
describing
- * which slots are available and allocated, and what jobs (JobManagers) the 
allocated slots
- * have been allocated to.
- */
-public class SlotReport implements Serializable {
-
-       private static final long serialVersionUID = -3150175198722481689L;
-
-       /** The slots status of the TaskManager */
-       private final List<SlotStatus> slotsStatus;
-
-       /** The resource id which identifies the TaskManager */
-       private final ResourceID resourceID;
-
-       public SlotReport(final List<SlotStatus> slotsStatus, final ResourceID 
resourceID) {
-               this.slotsStatus = checkNotNull(slotsStatus);
-               this.resourceID = checkNotNull(resourceID);
-       }
-
-       public List<SlotStatus> getSlotsStatus() {
-               return slotsStatus;
-       }
-
-       public ResourceID getResourceID() {
-               return resourceID;
-       }
-
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/2f12ba3e/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/SlotStatus.java
----------------------------------------------------------------------
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/SlotStatus.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/SlotStatus.java
deleted file mode 100644
index e8e2084..0000000
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/SlotStatus.java
+++ /dev/null
@@ -1,129 +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.runtime.rpc.taskexecutor;
-
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.runtime.clusterframework.types.AllocationID;
-import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
-import org.apache.flink.runtime.clusterframework.types.SlotID;
-
-import java.io.Serializable;
-
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
-/**
- * This describes the slot current status which located in TaskManager.
- */
-public class SlotStatus implements Serializable {
-
-       private static final long serialVersionUID = 5099191707339664493L;
-
-       /** slotID to identify a slot */
-       private final SlotID slotID;
-
-       /** the resource profile of the slot */
-       private final ResourceProfile profiler;
-
-       /** if the slot is allocated, allocationId identify its allocation; 
else, allocationId is null */
-       private final AllocationID allocationID;
-
-       /** if the slot is allocated, jobId identify which job this slot is 
allocated to; else, jobId is null */
-       private final JobID jobID;
-
-       public SlotStatus(SlotID slotID, ResourceProfile profiler) {
-               this(slotID, profiler, null, null);
-       }
-
-       public SlotStatus(SlotID slotID, ResourceProfile profiler, AllocationID 
allocationID, JobID jobID) {
-               this.slotID = checkNotNull(slotID, "slotID cannot be null");
-               this.profiler = checkNotNull(profiler, "profile cannot be 
null");
-               this.allocationID = allocationID;
-               this.jobID = jobID;
-       }
-
-       /**
-        * Get the unique identification of this slot
-        *
-        * @return The slot id
-        */
-       public SlotID getSlotID() {
-               return slotID;
-       }
-
-       /**
-        * Get the resource profile of this slot
-        *
-        * @return The resource profile
-        */
-       public ResourceProfile getProfiler() {
-               return profiler;
-       }
-
-       /**
-        * Get the allocation id of this slot
-        *
-        * @return The allocation id if this slot is allocated, otherwise null
-        */
-       public AllocationID getAllocationID() {
-               return allocationID;
-       }
-
-       /**
-        * Get the job id of the slot allocated for
-        *
-        * @return The job id if this slot is allocated, otherwise null
-        */
-       public JobID getJobID() {
-               return jobID;
-       }
-
-       @Override
-       public boolean equals(Object o) {
-               if (this == o) {
-                       return true;
-               }
-               if (o == null || getClass() != o.getClass()) {
-                       return false;
-               }
-
-               SlotStatus that = (SlotStatus) o;
-
-               if (!slotID.equals(that.slotID)) {
-                       return false;
-               }
-               if (!profiler.equals(that.profiler)) {
-                       return false;
-               }
-               if (allocationID != null ? 
!allocationID.equals(that.allocationID) : that.allocationID != null) {
-                       return false;
-               }
-               return jobID != null ? jobID.equals(that.jobID) : that.jobID == 
null;
-
-       }
-
-       @Override
-       public int hashCode() {
-               int result = slotID.hashCode();
-               result = 31 * result + profiler.hashCode();
-               result = 31 * result + (allocationID != null ? 
allocationID.hashCode() : 0);
-               result = 31 * result + (jobID != null ? jobID.hashCode() : 0);
-               return result;
-       }
-
-}

Reply via email to