You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2015/01/06 11:45:28 UTC
[1/4] incubator-flink git commit: [FLINK-1349] [runtime] Various
cleanups to make scala runtime code interact smoother with java
Repository: incubator-flink
Updated Branches:
refs/heads/master 24c47362e -> 2b86e9536
[FLINK-1349] [runtime] Various cleanups to make scala runtime code interact smoother with java
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/cec30ff4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/cec30ff4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/cec30ff4
Branch: refs/heads/master
Commit: cec30ff4f1b75a9a4642819ac28c1ee1939d4343
Parents: 972a7b0
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Jan 6 00:40:38 2015 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Jan 6 11:07:02 2015 +0100
----------------------------------------------------------------------
.../flink/runtime/executiongraph/Execution.java | 11 ++++----
.../runtime/messages/TaskManagerMessages.scala | 27 ++++++++++++++++++-
.../flink/runtime/taskmanager/TaskManager.scala | 17 +++++++++---
.../runtime/taskmanager/TaskManagerTest.java | 28 +++++++++++---------
.../testingUtils/TestingTaskManager.scala | 8 +++++-
.../TestingTaskManagerMessages.scala | 28 +++++++++++++++++---
6 files changed, 91 insertions(+), 28 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/cec30ff4/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
index 9100000..e7f4333 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
@@ -86,8 +86,8 @@ public class Execution implements Serializable {
private static final int NUM_CANCEL_CALL_TRIES = 3;
- public static FiniteDuration timeout = new FiniteDuration(ConfigConstants
- .DEFAULT_AKKA_ASK_TIMEOUT, TimeUnit.SECONDS);
+ public static FiniteDuration timeout = new FiniteDuration(
+ ConfigConstants.DEFAULT_AKKA_ASK_TIMEOUT, TimeUnit.SECONDS);
// --------------------------------------------------------------------------------------------
@@ -289,9 +289,9 @@ public class Execution implements Serializable {
@Override
public void onComplete(Throwable failure, Object success) throws Throwable {
- if(failure != null){
+ if (failure != null) {
markFailed(failure);
- }else{
+ } else {
TaskOperationResult result = (TaskOperationResult) success;
if (success == null) {
markFailed(new Exception("Failed to deploy the task to slot " + slot + ": TaskOperationResult was null"));
@@ -305,8 +305,7 @@ public class Execution implements Serializable {
else {
// deployment failed :(
markFailed(new Exception("Failed to deploy the task " +
- getVertexWithAttempt() + " to slot " + slot + ": " + result
- .description()));
+ getVertexWithAttempt() + " to slot " + slot + ": " + result.description()));
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/cec30ff4/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskManagerMessages.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskManagerMessages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskManagerMessages.scala
index aad9efe..f69b629 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskManagerMessages.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskManagerMessages.scala
@@ -21,10 +21,11 @@ package org.apache.flink.runtime.messages
import org.apache.flink.core.io.InputSplit
import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID
-import org.apache.flink.runtime.instance.InstanceID
+import org.apache.flink.runtime.instance.InstanceID
object TaskManagerMessages {
+
/**
* Cancels the task associated with [[attemptID]]. The result is sent back to the sender as a
* [[TaskOperationResult]] message.
@@ -113,4 +114,28 @@ object TaskManagerMessages {
* @param cause reason for the external failure
*/
case class FailTask(executionID: ExecutionAttemptID, cause: Throwable)
+
+ // --------------------------------------------------------------------------
+ // Utility methods to allow simpler case object access from Java
+ // --------------------------------------------------------------------------
+
+ def getNotifyWhenRegisteredAtJobManagerMessage() : AnyRef = {
+ NotifyWhenRegisteredAtJobManager
+ }
+
+ def getRegisteredAtJobManagerMessage() : AnyRef = {
+ RegisteredAtJobManager
+ }
+
+ def getRegisterAtJobManagerMessage() : AnyRef = {
+ RegisterAtJobManager
+ }
+
+ def getSendHeartbeatMessage() : AnyRef = {
+ SendHeartbeat
+ }
+
+ def getLogMemoryUsageMessage() : AnyRef = {
+ RegisteredAtJobManager
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/cec30ff4/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
index 5b6ee86..4676ae9 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
@@ -171,14 +171,16 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
if (registered) {
registrationScheduler.foreach(_.cancel())
- } else if (registrationAttempts <= TaskManager.MAX_REGISTRATION_ATTEMPTS) {
+ }
+ else if (registrationAttempts <= TaskManager.MAX_REGISTRATION_ATTEMPTS) {
log.info(s"Try to register at master ${jobManagerAkkaURL}. ${registrationAttempts}. " +
s"Attempt")
val jobManager = context.actorSelection(jobManagerAkkaURL)
jobManager ! RegisterTaskManager(connectionInfo, hardwareDescription, numberOfSlots)
- } else {
+ }
+ else {
log.error("TaskManager could not register at JobManager.");
self ! PoisonPill
}
@@ -212,6 +214,10 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
waitForRegistration.clear()
}
}
+
+ case SubmitTask(tdd) => {
+ submitTask(tdd)
+ }
case CancelTask(executionID) => {
runningTasks.get(executionID) match {
@@ -502,7 +508,8 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
}
/**
- * TaskManager companion object. Contains TaskManager executable entry point, command line parsing, and constants.
+ * TaskManager companion object. Contains TaskManager executable entry point, command
+ * line parsing, and constants.
*/
object TaskManager {
@@ -749,7 +756,9 @@ object TaskManager {
s"NON HEAP: $nonHeapUsed/$nonHeapCommitted/$nonHeapMax MB (used/committed/max)]"
}
- private def getGarbageCollectorStatsAsString(gcMXBeans: Iterable[GarbageCollectorMXBean]): String = {
+ private def getGarbageCollectorStatsAsString(gcMXBeans: Iterable[GarbageCollectorMXBean])
+ : String =
+ {
val beans = gcMXBeans map {
bean =>
s"[${bean.getName}, GC TIME (ms): ${bean.getCollectionTime}, " +
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/cec30ff4/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
index 500532d..4355298 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
@@ -26,6 +26,7 @@ import akka.japi.Creator;
import akka.pattern.Patterns;
import akka.testkit.JavaTestKit;
import akka.util.Timeout;
+
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.GlobalConfiguration;
@@ -46,15 +47,16 @@ import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.messages.JobManagerMessages;
import org.apache.flink.runtime.messages.RegistrationMessages;
import org.apache.flink.runtime.messages.TaskManagerMessages.CancelTask;
-import org.apache.flink.runtime.messages.TaskManagerMessages.NotifyWhenRegisteredAtJobManager$;
import org.apache.flink.runtime.messages.TaskManagerMessages.SubmitTask;
import org.apache.flink.runtime.messages.TaskManagerMessages.TaskOperationResult;
+import org.apache.flink.runtime.messages.TaskManagerMessages;
import org.apache.flink.runtime.testingUtils.TestingTaskManagerMessages;
import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.apache.flink.runtime.types.IntegerRecord;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
+
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;
@@ -162,8 +164,8 @@ public class TaskManagerTest {
expectMsgEquals(new TaskOperationResult(eid1, true));
expectMsgEquals(new TaskOperationResult(eid2, true));
-
- tm.tell(TestingTaskManagerMessages.RequestRunningTasks$.MODULE$, getRef());
+
+ tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), getRef());
Map<ExecutionAttemptID, Task> runningTasks = expectMsgClass(TestingTaskManagerMessages
.ResponseRunningTasks.class).asJava();
@@ -187,7 +189,7 @@ public class TaskManagerTest {
assertEquals(ExecutionState.CANCELED, t1.getExecutionState());
- tm.tell(TestingTaskManagerMessages.RequestRunningTasks$.MODULE$, getRef());
+ tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), getRef());
runningTasks = expectMsgClass(TestingTaskManagerMessages
.ResponseRunningTasks.class).asJava();
@@ -206,7 +208,7 @@ public class TaskManagerTest {
assertEquals(ExecutionState.CANCELED, t2.getExecutionState());
- tm.tell(TestingTaskManagerMessages.RequestRunningTasks$.MODULE$, getRef());
+ tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), getRef());
runningTasks = expectMsgClass(TestingTaskManagerMessages
.ResponseRunningTasks.class).asJava();
@@ -276,7 +278,7 @@ public class TaskManagerTest {
expectMsgEquals(true);
expectMsgEquals(true);
- tm.tell(TestingTaskManagerMessages.RequestRunningTasks$.MODULE$, getRef());
+ tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), getRef());
Map<ExecutionAttemptID, Task> tasks = expectMsgClass(TestingTaskManagerMessages
.ResponseRunningTasks.class).asJava();
@@ -337,7 +339,7 @@ public class TaskManagerTest {
tm.tell(new SubmitTask(tdd1), getRef());
expectMsgEquals(new TaskOperationResult(eid1, true));
- tm.tell(TestingTaskManagerMessages.RequestRunningTasks$.MODULE$, getRef());
+ tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), getRef());
Map<ExecutionAttemptID, Task> tasks = expectMsgClass(TestingTaskManagerMessages.ResponseRunningTasks
.class).asJava();
@@ -359,7 +361,7 @@ public class TaskManagerTest {
assertEquals(ExecutionState.FINISHED, t2.getExecutionState());
}
- tm.tell(TestingTaskManagerMessages.RequestRunningTasks$.MODULE$, getRef());
+ tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), getRef());
tasks = expectMsgClass(TestingTaskManagerMessages.ResponseRunningTasks
.class).asJava();
@@ -424,7 +426,7 @@ public class TaskManagerTest {
expectMsgEquals(new TaskOperationResult(eid2, true));
expectMsgEquals(new TaskOperationResult(eid1, true));
- tm.tell(TestingTaskManagerMessages.RequestRunningTasks$.MODULE$, getRef());
+ tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), getRef());
Map<ExecutionAttemptID, Task> tasks = expectMsgClass(TestingTaskManagerMessages
.ResponseRunningTasks.class).asJava();
@@ -450,7 +452,7 @@ public class TaskManagerTest {
Await.ready(response, d);
}
- tm.tell(TestingTaskManagerMessages.RequestRunningTasks$.MODULE$, getRef());
+ tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), getRef());
tasks = expectMsgClass(TestingTaskManagerMessages
.ResponseRunningTasks.class).asJava();
@@ -514,6 +516,7 @@ public class TaskManagerTest {
}
}
+ @SuppressWarnings("serial")
public static class SimpleLookupJobManagerCreator implements Creator<SimpleLookupJobManager>{
private final ChannelID receiverID;
@@ -527,6 +530,7 @@ public class TaskManagerTest {
}
}
+ @SuppressWarnings("serial")
public static class SimpleLookupFailingUpdateJobManagerCreator implements
Creator<SimpleLookupFailingUpdateJobManager>{
private final ChannelID receiverID;
@@ -550,8 +554,8 @@ public class TaskManagerTest {
ActorRef taskManager = TestingUtils.startTestingTaskManagerWithConfiguration("localhost", cfg, system);
- Future<Object> response = Patterns.ask(taskManager, NotifyWhenRegisteredAtJobManager$.MODULE$,
- timeout);
+ Future<Object> response = Patterns.ask(taskManager,
+ TaskManagerMessages.getNotifyWhenRegisteredAtJobManagerMessage(), timeout);
try {
FiniteDuration d = new FiniteDuration(20, TimeUnit.SECONDS);
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/cec30ff4/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManager.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManager.scala
index 080af11..b1aa437 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManager.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManager.scala
@@ -39,8 +39,10 @@ trait TestingTaskManager extends ActorLogMessages {
}
def receiveTestMessages: Receive = {
+
case RequestRunningTasks =>
sender ! ResponseRunningTasks(runningTasks.toMap)
+
case NotifyWhenTaskRemoved(executionID) =>
runningTasks.get(executionID) match {
case Some(_) =>
@@ -48,16 +50,19 @@ trait TestingTaskManager extends ActorLogMessages {
waitForRemoval += (executionID -> (set + sender))
case None => sender ! true
}
+
case UnregisterTask(executionID) =>
super.receiveWithLogMessages(UnregisterTask(executionID))
waitForRemoval.get(executionID) match {
case Some(actors) => for(actor <- actors) actor ! true
case None =>
}
+
case RequestBroadcastVariablesWithReferences => {
sender ! ResponseBroadcastVariablesWithReferences(
bcVarManager.getNumberOfVariablesWithReferences)
}
+
case NotifyWhenJobRemoved(jobID) => {
if(runningTasks.values.exists(_.getJobID == jobID)){
val set = waitForJobRemoval.getOrElse(jobID, Set())
@@ -71,13 +76,14 @@ trait TestingTaskManager extends ActorLogMessages {
}
}
}
+
case CheckIfJobRemoved(jobID) => {
if(runningTasks.values.forall(_.getJobID != jobID)){
waitForJobRemoval.get(jobID) match {
case Some(listeners) => listeners foreach (_ ! true)
case None =>
}
- }else{
+ } else {
import context.dispatcher
context.system.scheduler.scheduleOnce(200 milliseconds, this.self, CheckIfJobRemoved(jobID))
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/cec30ff4/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManagerMessages.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManagerMessages.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManagerMessages.scala
index cb5282e..38cc829 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManagerMessages.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingTaskManagerMessages.scala
@@ -22,16 +22,36 @@ import org.apache.flink.runtime.executiongraph.ExecutionAttemptID
import org.apache.flink.runtime.jobgraph.JobID
import org.apache.flink.runtime.taskmanager.Task
-object TestingTaskManagerMessages{
+/**
+ * Additional messages that the [[TestingTaskManager]] understands.
+ */
+object TestingTaskManagerMessages {
+
case class NotifyWhenTaskRemoved(executionID: ExecutionAttemptID)
-
- case object RequestRunningTasks
+
case class ResponseRunningTasks(tasks: Map[ExecutionAttemptID, Task]){
import collection.JavaConverters._
def asJava: java.util.Map[ExecutionAttemptID, Task] = tasks.asJava
}
- case object RequestBroadcastVariablesWithReferences
+
case class ResponseBroadcastVariablesWithReferences(number: Int)
case class CheckIfJobRemoved(jobID: JobID)
+
+ case object RequestRunningTasks
+
+ case object RequestBroadcastVariablesWithReferences
+
+ // --------------------------------------------------------------------------
+ // Utility methods to allow simpler case object access from Java
+ // --------------------------------------------------------------------------
+
+ def getRequestRunningTasksMessage() : AnyRef = {
+ RequestRunningTasks
+ }
+
+ def getRequestBroadcastVariablesWithReferencesMessage() : AnyRef = {
+ RequestBroadcastVariablesWithReferences
+ }
}
+
[2/4] incubator-flink git commit: [FLINK-1349] [runtime] Various
fixes concerning Akka - Remove obsolete code from old IPC net utils - Smaller
Writable/IOReadableWritable serialzation buffer start size (most messages are
rather small) - For message lo
Posted by se...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
index d8950bd..5b6ee86 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
@@ -23,7 +23,7 @@ import java.lang.management.{GarbageCollectorMXBean, MemoryMXBean, ManagementFac
import java.net.{InetAddress, InetSocketAddress}
import java.util
import java.util.concurrent.{FutureTask, TimeUnit}
-
+import scala.collection.JavaConverters._
import akka.actor._
import akka.pattern.ask
import org.apache.flink.api.common.cache.DistributedCache
@@ -34,35 +34,46 @@ import org.apache.flink.runtime.akka.AkkaUtils
import org.apache.flink.runtime.blob.BlobCache
import org.apache.flink.runtime.broadcast.BroadcastVariableManager
import org.apache.flink.runtime.execution.{CancelTaskException, ExecutionState, RuntimeEnvironment}
-import org.apache.flink.runtime.execution.librarycache.{BlobLibraryCacheManager,
-FallbackLibraryCacheManager, LibraryCacheManager}
+import org.apache.flink.runtime.execution.librarycache.{BlobLibraryCacheManager, FallbackLibraryCacheManager, LibraryCacheManager}
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID
import org.apache.flink.runtime.filecache.FileCache
import org.apache.flink.runtime.instance.{InstanceConnectionInfo, HardwareDescription, InstanceID}
import org.apache.flink.runtime.io.disk.iomanager.{IOManagerAsync}
import org.apache.flink.runtime.io.network.netty.NettyConnectionManager
-import org.apache.flink.runtime.io.network.{NetworkConnectionManager, LocalConnectionManager,
-ChannelManager}
+import org.apache.flink.runtime.io.network.{NetworkConnectionManager, LocalConnectionManager, ChannelManager}
import org.apache.flink.runtime.jobgraph.JobID
import org.apache.flink.runtime.jobmanager.JobManager
import org.apache.flink.runtime.memorymanager.DefaultMemoryManager
import org.apache.flink.runtime.messages.JobManagerMessages.UpdateTaskExecutionState
-import org.apache.flink.runtime.messages.RegistrationMessages.{RegisterTaskManager,
-AcknowledgeRegistration}
+import org.apache.flink.runtime.messages.RegistrationMessages.{RegisterTaskManager, AcknowledgeRegistration}
import org.apache.flink.runtime.messages.TaskManagerMessages._
-import org.apache.flink.runtime.messages.TaskManagerProfilerMessages.{UnmonitorTask, MonitorTask,
-RegisterProfilingListener}
+import org.apache.flink.runtime.messages.TaskManagerProfilerMessages.{UnmonitorTask, MonitorTask, RegisterProfilingListener}
import org.apache.flink.runtime.net.NetUtils
import org.apache.flink.runtime.profiling.ProfilingUtils
import org.apache.flink.runtime.util.EnvironmentInformation
import org.apache.flink.util.ExceptionUtils
import org.slf4j.LoggerFactory
-
import scala.concurrent.Future
import scala.concurrent.duration._
import scala.util.Failure
import scala.util.Success
-
+import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor
+
+/**
+ *
+ *
+ * The TaskManager has the following phases:
+ *
+ * - Waiting to be registered with its JobManager. In that phase, it periodically sends
+ * [[RegisterAtJobManager]] messages to itself, which trigger the sending of
+ * a [[RegisterTaskManager]] message to the JobManager.
+ *
+ * - Upon successful registration, the JobManager replies with an [[AcknowledgeRegistration]]
+ * message. This stops the registration messages and initializes all fields
+ * that require the JobManager's actor reference
+ *
+ * - ...
+ */
class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkkaURL: String,
val taskManagerConfig: TaskManagerConfiguration,
val networkConnectionConfig: NetworkConnectionConfiguration)
@@ -70,17 +81,12 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
import context._
import taskManagerConfig.{timeout => tmTimeout, _}
- import scala.collection.JavaConverters._
implicit val timeout = tmTimeout
log.info(s"Starting task manager at ${self.path}.")
- val REGISTRATION_DELAY = 0 seconds
- val REGISTRATION_INTERVAL = 10 seconds
- val MAX_REGISTRATION_ATTEMPTS = 10
- val HEARTBEAT_INTERVAL = 5000 millisecond
-
TaskManager.checkTempDirs(tmpDirPaths)
+
val ioManager = new IOManagerAsync(tmpDirPaths)
val memoryManager = new DefaultMemoryManager(memorySize, numberOfSlots, pageSize)
val bcVarManager = new BroadcastVariableManager();
@@ -92,29 +98,24 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
val waitForRegistration = scala.collection.mutable.Set[ActorRef]();
val profiler = profilingInterval match {
- case Some(interval) => Some(TaskManager.startProfiler(self.path.toSerializationFormat,
- interval))
+ case Some(interval) =>
+ Some(TaskManager.startProfiler(self.path.toSerializationFormat, interval))
case None => None
}
- var libraryCacheManager: LibraryCacheManager = null
+ var libraryCacheManager: LibraryCacheManager = _
var channelManager: Option[ChannelManager] = None
var registrationScheduler: Option[Cancellable] = None
var registrationAttempts: Int = 0
var registered: Boolean = false
var currentJobManager = ActorRef.noSender
var instanceID: InstanceID = null;
- var memoryMXBean: Option[MemoryMXBean] = None
- var gcMXBeans: Option[Iterable[GarbageCollectorMXBean]] = None
var heartbeatScheduler: Option[Cancellable] = None
if (log.isDebugEnabled) {
memoryLogggingIntervalMs.foreach {
interval =>
val d = FiniteDuration(interval, TimeUnit.MILLISECONDS)
- memoryMXBean = Some(ManagementFactory.getMemoryMXBean)
- gcMXBeans = Some(ManagementFactory.getGarbageCollectorMXBeans.asScala)
-
context.system.scheduler.schedule(d, d, self, LogMemoryUsage)
}
}
@@ -137,8 +138,7 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
try {
channelManager.shutdown()
} catch {
- case t: Throwable =>
- log.error(t, "ChannelManager did not shutdown properly.")
+ case t: Throwable => log.error(t, "ChannelManager did not shutdown properly.")
}
}
@@ -146,16 +146,22 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
memoryManager.shutdown()
fileCache.shutdown()
- if(libraryCacheManager != null){
- libraryCacheManager.shutdown()
+ if (libraryCacheManager != null) {
+ try {
+ libraryCacheManager.shutdown()
+ }
+ catch {
+ case t: Throwable => log.error(t, "LibraryCacheManager did not shutdown properly.")
+ }
}
}
- def tryJobManagerRegistration(): Unit = {
+ private def tryJobManagerRegistration(): Unit = {
registrationAttempts = 0
import context.dispatcher
- registrationScheduler = Some(context.system.scheduler.schedule(REGISTRATION_DELAY,
- REGISTRATION_INTERVAL, self, RegisterAtJobManager))
+ registrationScheduler = Some(context.system.scheduler.schedule(
+ TaskManager.REGISTRATION_DELAY, TaskManager.REGISTRATION_INTERVAL,
+ self, RegisterAtJobManager))
}
@@ -165,7 +171,7 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
if (registered) {
registrationScheduler.foreach(_.cancel())
- } else if (registrationAttempts <= MAX_REGISTRATION_ATTEMPTS) {
+ } else if (registrationAttempts <= TaskManager.MAX_REGISTRATION_ATTEMPTS) {
log.info(s"Try to register at master ${jobManagerAkkaURL}. ${registrationAttempts}. " +
s"Attempt")
@@ -187,16 +193,13 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
context.watch(currentJobManager)
log.info(s"TaskManager successfully registered at JobManager ${
- currentJobManager.path
- .toString
- }.")
+ currentJobManager.path.toString }.")
setupChannelManager()
setupLibraryCacheManager(blobPort)
- heartbeatScheduler = Some(context.system.scheduler.schedule(HEARTBEAT_INTERVAL,
- HEARTBEAT_INTERVAL, self,
- SendHeartbeat))
+ heartbeatScheduler = Some(context.system.scheduler.schedule(
+ TaskManager.HEARTBEAT_INTERVAL, TaskManager.HEARTBEAT_INTERVAL, self, SendHeartbeat))
profiler foreach {
_.tell(RegisterProfilingListener, JobManager.getProfiler(currentJobManager))
@@ -218,56 +221,121 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
}
sender ! new TaskOperationResult(executionID, true)
case None =>
- sender ! new TaskOperationResult(executionID, false, "No task with that execution ID " +
- "was " +
- "found.")
+ sender ! new TaskOperationResult(executionID, false,
+ "No task with that execution ID was found.")
}
}
- case SubmitTask(tdd) => {
- val jobID = tdd.getJobID
- val vertexID = tdd.getVertexID
- val executionID = tdd.getExecutionId
- val taskIndex = tdd.getIndexInSubtaskGroup
- val numSubtasks = tdd.getCurrentNumberOfSubtasks
- var startRegisteringTask = 0L
- var task: Task = null
+ case UnregisterTask(executionID) => {
+ unregisterTask(executionID)
+ }
- try {
- if(log.isDebugEnabled){
- startRegisteringTask = System.currentTimeMillis()
+ case SendHeartbeat => {
+ currentJobManager ! Heartbeat(instanceID)
+ }
+
+ case LogMemoryUsage => {
+ logMemoryStats()
+ }
+
+ case NotifyWhenRegisteredAtJobManager => {
+ if (registered) {
+ sender ! RegisteredAtJobManager
+ } else {
+ waitForRegistration += sender
+ }
+ }
+
+ case FailTask(executionID, cause) => {
+ runningTasks.get(executionID) match {
+ case Some(task) =>
+ Future{
+ task.failExternally(cause)
+ }
+ case None =>
+ }
+ }
+
+ case Terminated(jobManager) => {
+ log.info(s"Job manager ${jobManager.path} is no longer reachable. "
+ + "Cancelling all tasks and trying to reregister.")
+
+ cancelAndClearEverything(new Throwable("Lost connection to JobManager"))
+ tryJobManagerRegistration()
+ }
+ }
+
+ def notifyExecutionStateChange(jobID: JobID, executionID: ExecutionAttemptID,
+ executionState: ExecutionState,
+ optionalError: Throwable): Unit = {
+ log.info(s"Update execution state to ${executionState}.")
+ val futureResponse = (currentJobManager ? UpdateTaskExecutionState(new TaskExecutionState
+ (jobID, executionID, executionState, optionalError)))(timeout)
+
+ futureResponse.mapTo[Boolean].onComplete {
+ case Success(result) =>
+ if(!result){
+ self ! FailTask(executionID, new IllegalStateException("Task has been disposed on " +
+ "JobManager."))
}
- libraryCacheManager.registerTask(jobID, executionID, tdd.getRequiredJarFiles());
- if(log.isDebugEnabled){
- log.debug(s"Register task ${executionID} took ${(System.currentTimeMillis() -
- startRegisteringTask)/1000.0}s")
+ if (!result || executionState == ExecutionState.FINISHED || executionState ==
+ ExecutionState.CANCELED || executionState == ExecutionState.FAILED) {
+ self ! UnregisterTask(executionID)
}
+ case Failure(t) => {
+ log.warning(s"Execution state change notification failed for task ${executionID} " +
+ s"of job ${jobID}. Cause ${t.getMessage}.")
+ self ! UnregisterTask(executionID)
+ }
+ }
+ }
- val userCodeClassLoader = libraryCacheManager.getClassLoader(jobID)
+ private def submitTask(tdd: TaskDeploymentDescriptor) : Unit = {
+ val jobID = tdd.getJobID
+ val vertexID = tdd.getVertexID
+ val executionID = tdd.getExecutionId
+ val taskIndex = tdd.getIndexInSubtaskGroup
+ val numSubtasks = tdd.getCurrentNumberOfSubtasks
+ var startRegisteringTask = 0L
+ var task: Task = null
- if (userCodeClassLoader == null) {
- throw new RuntimeException("No user code Classloader available.")
- }
+ try {
+ if (log.isDebugEnabled) {
+ startRegisteringTask = System.currentTimeMillis()
+ }
+ libraryCacheManager.registerTask(jobID, executionID, tdd.getRequiredJarFiles());
- task = new Task(jobID, vertexID, taskIndex, numSubtasks, executionID,
- tdd.getTaskName, this)
+ if (log.isDebugEnabled) {
+ log.debug(s"Register task ${executionID} took ${
+ (System.currentTimeMillis() - startRegisteringTask)/1000.0}s")
+ }
- runningTasks.put(executionID, task) match {
- case Some(_) => throw new RuntimeException(s"TaskManager contains already a task with " +
- s"executionID ${executionID}.")
- case None =>
- }
+ val userCodeClassLoader = libraryCacheManager.getClassLoader(jobID)
- val splitProvider = new TaskInputSplitProvider(currentJobManager, jobID, vertexID,
+ if (userCodeClassLoader == null) {
+ throw new RuntimeException("No user code Classloader available.")
+ }
+
+ task = new Task(jobID, vertexID, taskIndex, numSubtasks, executionID,
+ tdd.getTaskName, this)
+
+ runningTasks.put(executionID, task) match {
+ case Some(_) => throw new RuntimeException(
+ s"TaskManager contains already a task with executionID ${executionID}.")
+ case None =>
+ }
+
+ val splitProvider = new TaskInputSplitProvider(currentJobManager, jobID, vertexID,
executionID, timeout)
- val env = new RuntimeEnvironment(task, tdd, userCodeClassLoader, memoryManager,
+
+ val env = new RuntimeEnvironment(task, tdd, userCodeClassLoader, memoryManager,
ioManager, splitProvider, currentJobManager, bcVarManager)
- task.setEnvironment(env)
+ task.setEnvironment(env)
- // register the task with the network stack and profilers
- channelManager match {
+ // register the task with the network stack and profilers
+ channelManager match {
case Some(cm) => cm.register(task)
case None => throw new RuntimeException("ChannelManager has not been properly " +
"instantiated.")
@@ -311,88 +379,15 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
}
libraryCacheManager.unregisterTask(jobID, executionID)
- }catch{
+ } catch {
case t: Throwable => log.error("Error during cleanup of task deployment.", t)
}
sender ! new TaskOperationResult(executionID, false, message)
}
- }
-
- case UnregisterTask(executionID) => {
- unregisterTask(executionID)
- }
-
- case SendHeartbeat => {
- currentJobManager ! Heartbeat(instanceID)
- }
-
- case LogMemoryUsage => {
- memoryMXBean foreach {
- mxbean =>
- if(log.isDebugEnabled) {
- log.debug(TaskManager.getMemoryUsageStatsAsString(mxbean))
- }
- }
-
- gcMXBeans foreach {
- mxbeans =>
- if(log.isDebugEnabled) {
- log.debug(TaskManager.getGarbageCollectorStatsAsString(mxbeans))
- }
- }
- }
-
- case NotifyWhenRegisteredAtJobManager => {
- registered match {
- case true => sender ! RegisteredAtJobManager
- case false => waitForRegistration += sender
- }
- }
-
- case FailTask(executionID, cause) => {
- runningTasks.get(executionID) match {
- case Some(task) =>
- Future{
- task.failExternally(cause)
- }
- case None =>
- }
- }
-
- case Terminated(jobManager) => {
- log.info(s"Job manager ${jobManager.path} is no longer reachable. Try to reregister.")
- tryJobManagerRegistration()
- }
}
-
- def notifyExecutionStateChange(jobID: JobID, executionID: ExecutionAttemptID,
- executionState: ExecutionState,
- optionalError: Throwable): Unit = {
- log.info(s"Update execution state to ${executionState}.")
- val futureResponse = (currentJobManager ? UpdateTaskExecutionState(new TaskExecutionState
- (jobID, executionID, executionState, optionalError)))(timeout)
-
- futureResponse.mapTo[Boolean].onComplete {
- case Success(result) =>
- if(!result){
- self ! FailTask(executionID, new IllegalStateException("Task has been disposed on " +
- "JobManager."))
- }
-
- if (!result || executionState == ExecutionState.FINISHED || executionState ==
- ExecutionState.CANCELED || executionState == ExecutionState.FAILED) {
- self ! UnregisterTask(executionID)
- }
- case Failure(t) => {
- log.warning(s"Execution state change notification failed for task ${executionID} " +
- s"of job ${jobID}. Cause ${t.getMessage}.")
- self ! UnregisterTask(executionID)
- }
- }
- }
-
- def setupChannelManager(): Unit = {
+
+ private def setupChannelManager(): Unit = {
//shutdown existing channel manager
channelManager foreach {
cm =>
@@ -424,12 +419,23 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
}
}
- def setupLibraryCacheManager(blobPort: Int): Unit = {
- if(blobPort > 0){
+ private def setupLibraryCacheManager(blobPort: Int): Unit = {
+ // shutdown existing library cache manager
+ if (libraryCacheManager != null) {
+ try {
+ libraryCacheManager.shutdown()
+ }
+ catch {
+ case t: Throwable => log.error(t, "Could not properly shut down LibraryCacheManager.")
+ }
+ libraryCacheManager = null
+ }
+
+ if (blobPort > 0) {
val address = new InetSocketAddress(currentJobManager.path.address.host.getOrElse
("localhost"), blobPort)
libraryCacheManager = new BlobLibraryCacheManager(new BlobCache(address), cleanupInterval)
- }else{
+ } else {
libraryCacheManager = new FallbackLibraryCacheManager
}
}
@@ -437,7 +443,7 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
/**
* Removes all tasks from this TaskManager.
*/
- def cancelAndClearEverything(cause: Throwable) {
+ private def cancelAndClearEverything(cause: Throwable) {
if (runningTasks.size > 0) {
log.info("Cancelling all computations and discarding all cached data.")
for (t <- runningTasks.values) {
@@ -447,7 +453,7 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
}
}
- def unregisterTask(executionID: ExecutionAttemptID): Unit = {
+ private def unregisterTask(executionID: ExecutionAttemptID): Unit = {
log.info(s"Unregister task with execution ID ${executionID}.")
runningTasks.remove(executionID) match {
case Some(task) =>
@@ -460,7 +466,7 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
}
}
- def removeAllTaskResources(task: Task): Unit = {
+ private def removeAllTaskResources(task: Task): Unit = {
if(task.getEnvironment != null) {
try {
for (entry <- DistributedCache.readFileInfoFromConfig(task.getEnvironment
@@ -483,8 +489,21 @@ class TaskManager(val connectionInfo: InstanceConnectionInfo, val jobManagerAkka
task.unregisterMemoryManager(memoryManager)
}
+
+ private def logMemoryStats(): Unit = {
+ if (log.isDebugEnabled) {
+ val memoryMXBean = ManagementFactory.getMemoryMXBean()
+ val gcMXBeans = ManagementFactory.getGarbageCollectorMXBeans().asScala
+
+ log.debug(TaskManager.getMemoryUsageStatsAsString(memoryMXBean))
+ log.debug(TaskManager.getGarbageCollectorStatsAsString(gcMXBeans))
+ }
+ }
}
+/**
+ * TaskManager companion object. Contains TaskManager executable entry point, command line parsing, and constants.
+ */
object TaskManager {
val LOG = LoggerFactory.getLogger(classOf[TaskManager])
@@ -492,7 +511,13 @@ object TaskManager {
val TASK_MANAGER_NAME = "taskmanager"
val PROFILER_NAME = "profiler"
+
+ val REGISTRATION_DELAY = 0 seconds
+ val REGISTRATION_INTERVAL = 10 seconds
+ val MAX_REGISTRATION_ATTEMPTS = 10
+ val HEARTBEAT_INTERVAL = 5000 millisecond
+
def main(args: Array[String]): Unit = {
val (hostname, port, configuration) = parseArgs(args)
@@ -599,7 +624,7 @@ object TaskManager {
val networkConnectionConfiguration = if(localExecution){
LocalNetworkConfiguration(numBuffers, bufferSize)
- }else{
+ } else {
val numInThreads = configuration.getInteger(
ConfigConstants.TASK_MANAGER_NET_NUM_IN_THREADS_KEY,
ConfigConstants.DEFAULT_TASK_MANAGER_NET_NUM_IN_THREADS)
@@ -708,7 +733,7 @@ object TaskManager {
}
}
- def getMemoryUsageStatsAsString(memoryMXBean: MemoryMXBean): String = {
+ private def getMemoryUsageStatsAsString(memoryMXBean: MemoryMXBean): String = {
val heap = memoryMXBean.getHeapMemoryUsage()
val nonHeap = memoryMXBean.getNonHeapMemoryUsage
@@ -724,7 +749,7 @@ object TaskManager {
s"NON HEAP: $nonHeapUsed/$nonHeapCommitted/$nonHeapMax MB (used/committed/max)]"
}
- def getGarbageCollectorStatsAsString(gcMXBeans: Iterable[GarbageCollectorMXBean]): String = {
+ private def getGarbageCollectorStatsAsString(gcMXBeans: Iterable[GarbageCollectorMXBean]): String = {
val beans = gcMXBeans map {
bean =>
s"[${bean.getName}, GC TIME (ms): ${bean.getCollectionTime}, " +
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java
index 1735f68..c7914fc 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java
@@ -104,7 +104,7 @@ public class ExecutionGraphDeploymentTest {
ExecutionVertex vertex = ejv.getTaskVertices()[3];
// create synchronous task manager
- final TestActorRef simpleTaskManager = TestActorRef.create(system,
+ final TestActorRef<?> simpleTaskManager = TestActorRef.create(system,
Props.create(ExecutionGraphTestUtils
.SimpleAcknowledgingTaskManager.class));
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java
index 207be1a..a4adda8 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java
@@ -32,6 +32,7 @@ import akka.actor.Status;
import akka.actor.UntypedActor;
import akka.japi.Creator;
import akka.testkit.JavaTestKit;
+
import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.instance.AllocatedSlot;
@@ -46,7 +47,9 @@ import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
+@SuppressWarnings("serial")
public class ExecutionVertexCancelTest {
+
private static ActorSystem system;
@BeforeClass
@@ -209,7 +212,7 @@ public class ExecutionVertexCancelTest {
// task manager mock actor
// first return NOT SUCCESS (task not found, cancel call overtook deploy call), then success (cancel call after deploy call)
- TestActorRef taskManager = TestActorRef.create(system, Props.create(new
+ TestActorRef<?> taskManager = TestActorRef.create(system, Props.create(new
CancelSequenceTaskManagerCreator(new
TaskOperationResult(execId, false), new TaskOperationResult(execId, true))));
@@ -281,7 +284,7 @@ public class ExecutionVertexCancelTest {
final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0]);
final ExecutionAttemptID execId = vertex.getCurrentExecutionAttempt().getAttemptId();
- final TestActorRef taskManager = TestActorRef.create(system,
+ final TestActorRef<?> taskManager = TestActorRef.create(system,
Props.create(new CancelSequenceTaskManagerCreator(new
TaskOperationResult(execId, true))));
@@ -465,7 +468,6 @@ public class ExecutionVertexCancelTest {
final ExecutionJobVertex ejv = getExecutionVertex(jid);
final ExecutionVertex vertex = new ExecutionVertex(ejv, 0, new IntermediateResult[0]);
- final ExecutionAttemptID execId = vertex.getCurrentExecutionAttempt().getAttemptId();
final ActorRef taskManager = system.actorOf(Props.create(new CancelSequenceTaskManagerCreator()));
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/test/java/org/apache/flink/runtime/operators/DataSinkTaskTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/DataSinkTaskTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/DataSinkTaskTest.java
index 3219a21..cb657a0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/DataSinkTaskTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/DataSinkTaskTest.java
@@ -34,7 +34,6 @@ import org.slf4j.LoggerFactory;
import org.apache.flink.api.common.typeutils.record.RecordComparatorFactory;
import org.apache.flink.api.java.record.io.DelimitedOutputFormat;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.operators.testutils.InfiniteInputIterator;
import org.apache.flink.runtime.operators.testutils.TaskCancelThread;
import org.apache.flink.runtime.operators.testutils.TaskTestBase;
@@ -54,7 +53,7 @@ public class DataSinkTaskTest extends TaskTestBase
private static final int NETWORK_BUFFER_SIZE = 1024;
- private final String tempTestPath = Path.constructTestPath(DataSinkTaskTest.class, "dst_test");
+ private final String tempTestPath = constructTestPath(DataSinkTaskTest.class, "dst_test");
@After
public void cleanUp() {
@@ -491,5 +490,19 @@ public class DataSinkTaskTest extends TaskTestBase
}
}
+ public static String constructTestPath(Class<?> forClass, String folder) {
+ // we create test path that depends on class to prevent name clashes when two tests
+ // create temp files with the same name
+ String path = System.getProperty("java.io.tmpdir");
+ if (!(path.endsWith("/") || path.endsWith("\\")) ) {
+ path += System.getProperty("file.separator");
+ }
+ path += (forClass.getName() + "-" + folder);
+ return path;
+ }
+
+ public static String constructTestURI(Class<?> forClass, String folder) {
+ return new File(constructTestPath(forClass, folder)).toURI().toString();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/test/java/org/apache/flink/runtime/operators/DataSourceTaskTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/DataSourceTaskTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/DataSourceTaskTest.java
index b04ab30..aac7bc9 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/DataSourceTaskTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/DataSourceTaskTest.java
@@ -31,7 +31,6 @@ import java.util.List;
import org.junit.Assert;
import org.apache.flink.api.java.record.io.DelimitedInputFormat;
-import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.operators.testutils.NirvanaOutputList;
import org.apache.flink.runtime.operators.testutils.TaskCancelThread;
import org.apache.flink.runtime.operators.testutils.TaskTestBase;
@@ -50,7 +49,7 @@ public class DataSourceTaskTest extends TaskTestBase {
private List<Record> outList;
- private String tempTestPath = Path.constructTestPath(DataSourceTaskTest.class, "dst_test");
+ private String tempTestPath = DataSinkTaskTest.constructTestPath(DataSourceTaskTest.class, "dst_test");
@After
public void cleanUp() {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java b/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
index fe2d7fc..6c4b228 100644
--- a/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
+++ b/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java
@@ -22,6 +22,7 @@ import akka.actor.ActorRef;
import akka.dispatch.Futures;
import akka.pattern.Patterns;
import akka.util.Timeout;
+
import org.apache.commons.io.FileUtils;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
@@ -29,6 +30,7 @@ import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.testingUtils.TestingTaskManagerMessages;
import org.apache.hadoop.fs.FileSystem;
import org.junit.Assert;
+
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;
@@ -351,4 +353,19 @@ public class TestBaseUtils {
f.delete();
}
}
+
+ public static String constructTestPath(Class<?> forClass, String folder) {
+ // we create test path that depends on class to prevent name clashes when two tests
+ // create temp files with the same name
+ String path = System.getProperty("java.io.tmpdir");
+ if (!(path.endsWith("/") || path.endsWith("\\")) ) {
+ path += System.getProperty("file.separator");
+ }
+ path += (forClass.getName() + "-" + folder);
+ return path;
+ }
+
+ public static String constructTestURI(Class<?> forClass, String folder) {
+ return new File(constructTestPath(forClass, folder)).toURI().toString();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-tests/pom.xml
----------------------------------------------------------------------
diff --git a/flink-tests/pom.xml b/flink-tests/pom.xml
index 0750264..21ccbdb 100644
--- a/flink-tests/pom.xml
+++ b/flink-tests/pom.xml
@@ -126,7 +126,6 @@ under the License.
<dependency>
<groupId>org.scalatest</groupId>
<artifactId>scalatest_2.10</artifactId>
- <version>2.2.0</version>
<scope>test</scope>
</dependency>
</dependencies>
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/customdanglingpagerank/CustomCompensatableDanglingPageRank.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/customdanglingpagerank/CustomCompensatableDanglingPageRank.java b/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/customdanglingpagerank/CustomCompensatableDanglingPageRank.java
index 6327db9..d79b39b 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/customdanglingpagerank/CustomCompensatableDanglingPageRank.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/customdanglingpagerank/CustomCompensatableDanglingPageRank.java
@@ -24,7 +24,6 @@ import org.apache.flink.api.common.typeutils.TypePairComparatorFactory;
import org.apache.flink.api.common.typeutils.TypeSerializerFactory;
import org.apache.flink.api.java.record.io.FileOutputFormat;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.io.network.channels.ChannelType;
import org.apache.flink.runtime.iterative.task.IterationHeadPactTask;
import org.apache.flink.runtime.iterative.task.IterationIntermediatePactTask;
@@ -56,6 +55,7 @@ import org.apache.flink.test.iterative.nephele.customdanglingpagerank.types.Vert
import org.apache.flink.test.iterative.nephele.customdanglingpagerank.types.VertexWithRankSerializerFactory;
import org.apache.flink.test.iterative.nephele.danglingpagerank.DiffL1NormConvergenceCriterion;
import org.apache.flink.test.iterative.nephele.danglingpagerank.PageRankStatsAggregator;
+import org.apache.flink.test.util.TestBaseUtils;
public class CustomCompensatableDanglingPageRank {
@@ -101,7 +101,7 @@ public class CustomCompensatableDanglingPageRank {
String pageWithRankInputPath = ""; //"file://" + PlayConstants.PLAY_DIR + "test-inputs/danglingpagerank/pageWithRank";
String adjacencyListInputPath = ""; //"file://" + PlayConstants.PLAY_DIR +
// "test-inputs/danglingpagerank/adjacencylists";
- String outputPath = Path.constructTestURI(CustomCompensatableDanglingPageRank.class, "flink_iterations");
+ String outputPath = TestBaseUtils.constructTestURI(CustomCompensatableDanglingPageRank.class, "flink_iterations");
// String confPath = PlayConstants.PLAY_DIR + "local-conf";
int minorConsumer = 2;
int matchMemory = 5;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/customdanglingpagerank/CustomCompensatableDanglingPageRankWithCombiner.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/customdanglingpagerank/CustomCompensatableDanglingPageRankWithCombiner.java b/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/customdanglingpagerank/CustomCompensatableDanglingPageRankWithCombiner.java
index 8bede88..6add8c2 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/customdanglingpagerank/CustomCompensatableDanglingPageRankWithCombiner.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/customdanglingpagerank/CustomCompensatableDanglingPageRankWithCombiner.java
@@ -24,7 +24,6 @@ import org.apache.flink.api.common.typeutils.TypePairComparatorFactory;
import org.apache.flink.api.common.typeutils.TypeSerializerFactory;
import org.apache.flink.api.java.record.io.FileOutputFormat;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.io.network.channels.ChannelType;
import org.apache.flink.runtime.iterative.task.IterationHeadPactTask;
import org.apache.flink.runtime.iterative.task.IterationIntermediatePactTask;
@@ -57,6 +56,7 @@ import org.apache.flink.test.iterative.nephele.customdanglingpagerank.types.Vert
import org.apache.flink.test.iterative.nephele.customdanglingpagerank.types.VertexWithRankSerializerFactory;
import org.apache.flink.test.iterative.nephele.danglingpagerank.DiffL1NormConvergenceCriterion;
import org.apache.flink.test.iterative.nephele.danglingpagerank.PageRankStatsAggregator;
+import org.apache.flink.test.util.TestBaseUtils;
public class CustomCompensatableDanglingPageRankWithCombiner {
@@ -102,7 +102,7 @@ public class CustomCompensatableDanglingPageRankWithCombiner {
String pageWithRankInputPath = ""; //"file://" + PlayConstants.PLAY_DIR + "test-inputs/danglingpagerank/pageWithRank";
String adjacencyListInputPath = ""; //"file://" + PlayConstants.PLAY_DIR +
// "test-inputs/danglingpagerank/adjacencylists";
- String outputPath = Path.constructTestURI(CustomCompensatableDanglingPageRankWithCombiner.class, "flink_iterations");
+ String outputPath = TestBaseUtils.constructTestURI(CustomCompensatableDanglingPageRankWithCombiner.class, "flink_iterations");
int minorConsumer = 2;
int matchMemory = 5;
int coGroupSortMemory = 5;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/GroupReduceITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/GroupReduceITCase.java b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/GroupReduceITCase.java
index 6cd484d..8fb61af 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/GroupReduceITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/GroupReduceITCase.java
@@ -34,14 +34,11 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.test.javaApiOperators.util.CollectionDataSets;
import org.apache.flink.test.javaApiOperators.util.CollectionDataSets.CrazyNested;
import org.apache.flink.test.javaApiOperators.util.CollectionDataSets.CustomType;
-import org.apache.flink.test.javaApiOperators.util.CollectionDataSets.FromTuple;
import org.apache.flink.test.javaApiOperators.util.CollectionDataSets.FromTupleWithCTor;
import org.apache.flink.test.javaApiOperators.util.CollectionDataSets.POJO;
import org.apache.flink.test.javaApiOperators.util.CollectionDataSets.PojoContainingTupleAndWritable;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.apache.flink.util.Collector;
-import org.hamcrest.core.IsEqual;
-import org.hamcrest.core.IsNot;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
@@ -302,14 +299,13 @@ public class GroupReduceITCase extends MultipleProgramsTestBase {
"23,2,Hi again!\n";
}
- @Test public void testCorrectnessOfGroupReduceOnCustomTypeWithKeyExtractorAndCombine() throws
- Exception {
+ @Test
+ public void testCorrectnessOfGroupReduceOnCustomTypeWithKeyExtractorAndCombine()
+ throws Exception {
/*
* check correctness of groupReduce on custom type with key extractor and combine
*/
-
- org.junit.Assume.assumeThat(mode, new IsNot(new IsEqual<ExecutionMode>(ExecutionMode
- .COLLECTION)));
+ org.junit.Assume.assumeTrue(mode != ExecutionMode.COLLECTION);
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
@@ -339,11 +335,11 @@ public class GroupReduceITCase extends MultipleProgramsTestBase {
@Test
public void testCorrectnessOfGroupReduceOnTuplesWithCombine() throws Exception {
+
/*
* check correctness of groupReduce on tuples with combine
*/
- org.junit.Assume.assumeThat(mode, new IsNot(new IsEqual<ExecutionMode>(ExecutionMode
- .COLLECTION)));
+ org.junit.Assume.assumeTrue(mode != ExecutionMode.COLLECTION);
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.setDegreeOfParallelism(2); // important because it determines how often the combiner is called
@@ -365,11 +361,11 @@ public class GroupReduceITCase extends MultipleProgramsTestBase {
@Test
public void testCorrectnessOfAllGroupReduceForTuplesWithCombine() throws Exception {
+
/*
* check correctness of all-groupreduce for tuples with combine
*/
- org.junit.Assume.assumeThat(mode, new IsNot(new IsEqual<ExecutionMode>(ExecutionMode
- .COLLECTION)));
+ org.junit.Assume.assumeTrue(mode != ExecutionMode.COLLECTION);
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
@@ -525,16 +521,10 @@ public class GroupReduceITCase extends MultipleProgramsTestBase {
}
public static class GroupReducer2 implements GroupReduceFunction<FromTupleWithCTor, Integer> {
- private static final long serialVersionUID = 1L;
+
@Override
- public void reduce(Iterable<FromTupleWithCTor> values,
- Collector<Integer> out)
- throws Exception {
- int c = 0;
- for(FromTuple v : values) {
- c++;
- }
- out.collect(c);
+ public void reduce(Iterable<FromTupleWithCTor> values, Collector<Integer> out) {
+ out.collect(countElements(values));
}
}
@@ -556,17 +546,10 @@ public class GroupReduceITCase extends MultipleProgramsTestBase {
}
public static class GroupReducer3 implements GroupReduceFunction<PojoContainingTupleAndWritable, Integer> {
- private static final long serialVersionUID = 1L;
-
+
@Override
- public void reduce(Iterable<PojoContainingTupleAndWritable> values,
- Collector<Integer> out)
- throws Exception {
- int c = 0;
- for(PojoContainingTupleAndWritable v : values) {
- c++;
- }
- out.collect(c);
+ public void reduce(Iterable<PojoContainingTupleAndWritable> values, Collector<Integer> out) {
+ out.collect(countElements(values));
}
}
@@ -590,14 +573,8 @@ public class GroupReduceITCase extends MultipleProgramsTestBase {
public static class GroupReducer4 implements GroupReduceFunction<Tuple3<Integer,CrazyNested, POJO>, Integer> {
private static final long serialVersionUID = 1L;
@Override
- public void reduce(Iterable<Tuple3<Integer,CrazyNested, POJO>> values,
- Collector<Integer> out)
- throws Exception {
- int c = 0;
- for(Tuple3<Integer,CrazyNested, POJO> v : values) {
- c++;
- }
- out.collect(c);
+ public void reduce(Iterable<Tuple3<Integer,CrazyNested, POJO>> values, Collector<Integer> out) {
+ out.collect(countElements(values));
}
}
@@ -1188,8 +1165,15 @@ public class GroupReduceITCase extends MultipleProgramsTestBase {
}
public static final class IdentityMapper<T> extends RichMapFunction<T, T> {
-
@Override
public T map(T value) { return value; }
}
+
+ private static int countElements(Iterable<?> iterable) {
+ int c = 0;
+ for (@SuppressWarnings("unused") Object o : iterable) {
+ c++;
+ }
+ return c;
+ }
}
[3/4] incubator-flink git commit: [FLINK-1349] [runtime] Various
fixes concerning Akka - Remove obsolete code from old IPC net utils - Smaller
Writable/IOReadableWritable serialzation buffer start size (most messages are
rather small) - For message lo
Posted by se...@apache.org.
[FLINK-1349] [runtime] Various fixes concerning Akka
- Remove obsolete code from old IPC net utils
- Smaller Writable/IOReadableWritable serialzation buffer start size (most messages are rather small)
- For message logging, make system calls (timestamps) only in debug mode
- Clean up warnings / code simplifications
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/972a7b01
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/972a7b01
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/972a7b01
Branch: refs/heads/master
Commit: 972a7b01980b4b06d4893df6aa6aebd4e26e5990
Parents: 24c4736
Author: Stephan Ewen <se...@apache.org>
Authored: Sun Dec 28 10:55:21 2014 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Jan 6 11:07:02 2015 +0100
----------------------------------------------------------------------
.../main/java/org/apache/flink/yarn/Utils.java | 2 +-
.../flink/configuration/ConfigConstants.java | 11 +-
.../java/org/apache/flink/core/fs/Path.java | 23 +-
flink-dist/pom.xml | 2 +-
flink-runtime/pom.xml | 1 -
.../runtime/accumulators/AccumulatorEvent.java | 2 +
.../org/apache/flink/runtime/blob/BlobKey.java | 88 ++--
.../deployment/ChannelDeploymentDescriptor.java | 2 +
.../deployment/GateDeploymentDescriptor.java | 2 +
.../deployment/TaskDeploymentDescriptor.java | 11 +
.../runtime/fs/hdfs/DistributedFileStatus.java | 10 +-
.../network/ConnectionInfoLookupResponse.java | 4 +-
.../runtime/io/network/RemoteReceiver.java | 2 +
.../apache/flink/runtime/jobgraph/JobGraph.java | 18 +-
.../runtime/jobmanager/scheduler/SubSlot.java | 3 +
.../runtime/jobmanager/web/WebInfoServer.java | 11 +-
.../org/apache/flink/runtime/net/NetUtils.java | 317 ++-----------
.../flink/runtime/net/SocketIOWithTimeout.java | 457 -------------------
.../flink/runtime/net/SocketInputStream.java | 179 --------
.../flink/runtime/net/SocketOutputStream.java | 222 ---------
.../apache/flink/runtime/ActorLogMessages.scala | 16 +-
.../IOReadableWritableSerializer.scala | 2 +-
.../akka/serialization/WritableSerializer.scala | 2 +-
.../flink/runtime/taskmanager/TaskManager.scala | 341 +++++++-------
.../ExecutionGraphDeploymentTest.java | 2 +-
.../ExecutionVertexCancelTest.java | 8 +-
.../runtime/operators/DataSinkTaskTest.java | 17 +-
.../runtime/operators/DataSourceTaskTest.java | 3 +-
.../apache/flink/test/util/TestBaseUtils.java | 17 +
flink-tests/pom.xml | 1 -
.../CustomCompensatableDanglingPageRank.java | 4 +-
...mpensatableDanglingPageRankWithCombiner.java | 4 +-
.../javaApiOperators/GroupReduceITCase.java | 64 +--
33 files changed, 372 insertions(+), 1476 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-addons/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java
----------------------------------------------------------------------
diff --git a/flink-addons/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java b/flink-addons/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java
index 6b541bf..bd5659a 100644
--- a/flink-addons/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java
+++ b/flink-addons/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java
@@ -121,7 +121,7 @@ public class Utils {
// chain-in a new classloader
URL fileUrl = null;
try {
- fileUrl = path.toURL();
+ fileUrl = path.toURI().toURL();
} catch (MalformedURLException e) {
throw new RuntimeException("Erroneous config file path", e);
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
index 2d97504..df96339 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
@@ -295,7 +295,7 @@ public final class ConfigConstants {
public static final String AKKA_STARTUP_TIMEOUT = "akka.startup-timeout";
/**
- * Hearbeat interval of the transport failure detector
+ * Heartbeat interval of the transport failure detector
*/
public static final String AKKA_TRANSPORT_HEARTBEAT_INTERVAL = "akka.transport.heartbeat.interval";
@@ -466,11 +466,6 @@ public final class ConfigConstants {
public static final int DEFAULT_TASK_MANAGER_NET_NETTY_HIGH_WATER_MARK = -1;
/**
- * The default interval for TaskManager heart beats (5000 msecs).
- */
- public static final int DEFAULT_TASK_MANAGER_HEARTBEAT_INTERVAL = 5000;
-
- /**
* Flag indicating whether to start a thread, which repeatedly logs the memory usage of the JVM.
*/
public static final boolean DEFAULT_TASK_MANAGER_DEBUG_MEMORY_USAGE_START_LOG_THREAD = false;
@@ -559,8 +554,8 @@ public final class ConfigConstants {
/**
* The default directory to store temporary objects (e.g. during file uploads).
*/
- public static final String DEFAULT_WEB_TMP_DIR = System.getProperty("java.io.tmpdir") == null ? "/tmp" : System
- .getProperty("java.io.tmpdir");
+ public static final String DEFAULT_WEB_TMP_DIR =
+ System.getProperty("java.io.tmpdir") == null ? "/tmp" : System.getProperty("java.io.tmpdir");
/**
* The default directory for temporary plan dumps from the web frontend.
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-core/src/main/java/org/apache/flink/core/fs/Path.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/Path.java b/flink-core/src/main/java/org/apache/flink/core/fs/Path.java
index 950a38b..a104d86 100644
--- a/flink-core/src/main/java/org/apache/flink/core/fs/Path.java
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/Path.java
@@ -16,16 +16,12 @@
* limitations under the License.
*/
-
-/**
- * This file is based on source code from the Hadoop Project (http://hadoop.apache.org/), licensed by the Apache
+/* This file is based on source code from the Hadoop Project (http://hadoop.apache.org/), licensed by the Apache
* Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
- * additional information regarding copyright ownership.
- */
+ * additional information regarding copyright ownership. */
package org.apache.flink.core.fs;
-import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.net.URI;
@@ -501,19 +497,4 @@ public class Path implements IOReadableWritable, Serializable {
}
}
-
- public static String constructTestPath(Class<?> forClass, String folder) {
- // we create test path that depends on class to prevent name clashes when two tests
- // create temp files with the same name
- String path = System.getProperty("java.io.tmpdir");
- if (!(path.endsWith("/") || path.endsWith("\\")) ) {
- path += System.getProperty("file.separator");
- }
- path += (forClass.getName() + "-" + folder);
- return path;
- }
-
- public static String constructTestURI(Class<?> forClass, String folder) {
- return new File(constructTestPath(forClass, folder)).toURI().toString();
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-dist/pom.xml
----------------------------------------------------------------------
diff --git a/flink-dist/pom.xml b/flink-dist/pom.xml
index f73a000..e753a05 100644
--- a/flink-dist/pom.xml
+++ b/flink-dist/pom.xml
@@ -210,7 +210,7 @@ under the License.
</plugin>
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
- <version>2.4</version>
+ <version>2.4</version><!--$NO-MVN-MAN-VER$-->
<executions>
<!-- yarn bin directory -->
<execution>
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/pom.xml
----------------------------------------------------------------------
diff --git a/flink-runtime/pom.xml b/flink-runtime/pom.xml
index b87977e..5a9ca8a 100644
--- a/flink-runtime/pom.xml
+++ b/flink-runtime/pom.xml
@@ -130,7 +130,6 @@ under the License.
<dependency>
<groupId>com.github.scopt</groupId>
<artifactId>scopt_2.10</artifactId>
- <version>3.2.0</version>
<exclusions>
<exclusion>
<groupId>org.scala-lang</groupId>
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorEvent.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorEvent.java b/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorEvent.java
index 761248b..3fed259 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorEvent.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorEvent.java
@@ -39,6 +39,8 @@ import org.apache.flink.util.InstantiationUtil;
*/
public class AccumulatorEvent implements Serializable {
+ private static final long serialVersionUID = 8965894516006882735L;
+
private JobID jobID;
private Map<String, Accumulator<?, ?>> accumulators;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobKey.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobKey.java b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobKey.java
index c12fd23..e3d237d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobKey.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobKey.java
@@ -32,19 +32,16 @@ import java.util.Arrays;
*/
public final class BlobKey implements Serializable, Comparable<BlobKey> {
- /**
- * Array of hex characters to facilitate fast toString() method.
- */
+ private static final long serialVersionUID = 3847117712521785209L;
+
+ /** Array of hex characters to facilitate fast toString() method. */
private static final char[] HEX_ARRAY = "0123456789abcdef".toCharArray();
- /**
- * Size of the internal BLOB key in bytes.
- */
+ /** Size of the internal BLOB key in bytes. */
private static final int SIZE = 20;
- /**
- * The byte buffer storing the actual key data.
- */
+
+ /** The byte buffer storing the actual key data. */
private final byte[] key;
/**
@@ -60,8 +57,7 @@ public final class BlobKey implements Serializable, Comparable<BlobKey> {
* @param key
* the actual key data
*/
- BlobKey(final byte[] key) {
-
+ BlobKey(byte[] key) {
if (key.length != SIZE) {
throw new IllegalArgumentException("BLOB key must have a size of " + SIZE + " bytes");
}
@@ -70,8 +66,15 @@ public final class BlobKey implements Serializable, Comparable<BlobKey> {
}
/**
- * {@inheritDoc}
+ * Adds the BLOB key to the given {@link MessageDigest}.
+ *
+ * @param md
+ * the message digest to add the BLOB key to
*/
+ public void addToMessageDigest(MessageDigest md) {
+ md.update(this.key);
+ }
+
@Override
public boolean equals(final Object obj) {
@@ -84,17 +87,11 @@ public final class BlobKey implements Serializable, Comparable<BlobKey> {
return Arrays.equals(this.key, bk.key);
}
- /**
- * {@inheritDoc}
- */
@Override
public int hashCode() {
return Arrays.hashCode(this.key);
}
- /**
- * {@inheritDoc}
- */
@Override
public String toString() {
// from http://stackoverflow.com/questions/9655181/convert-from-byte-array-to-hex-string-in-java
@@ -108,6 +105,26 @@ public final class BlobKey implements Serializable, Comparable<BlobKey> {
return new String(hexChars);
}
+ @Override
+ public int compareTo(BlobKey o) {
+
+ final byte[] aarr = this.key;
+ final byte[] barr = o.key;
+ final int len = Math.min(aarr.length, barr.length);
+
+ for (int i = 0; i < len; ++i) {
+ final int a = (aarr[i] & 0xff);
+ final int b = (barr[i] & 0xff);
+ if (a != b) {
+ return a - b;
+ }
+ }
+
+ return aarr.length - barr.length;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
/**
* Auxiliary method to read a BLOB key from an input stream.
*
@@ -117,7 +134,7 @@ public final class BlobKey implements Serializable, Comparable<BlobKey> {
* @throws IOException
* throw if an I/O error occurs while reading from the input stream
*/
- static BlobKey readFromInputStream(final InputStream inputStream) throws IOException {
+ static BlobKey readFromInputStream(InputStream inputStream) throws IOException {
final byte[] key = new byte[BlobKey.SIZE];
@@ -142,39 +159,6 @@ public final class BlobKey implements Serializable, Comparable<BlobKey> {
* thrown if an I/O error occurs while writing the BLOB key
*/
void writeToOutputStream(final OutputStream outputStream) throws IOException {
-
outputStream.write(this.key);
}
-
- /**
- * {@inheritDoc}
- */
- @Override
- public int compareTo(final BlobKey o) {
-
- final byte[] aarr = this.key;
- final byte[] barr = o.key;
- final int len = Math.min(aarr.length, barr.length);
-
- for (int i = 0; i < len; ++i) {
- final int a = (aarr[i] & 0xff);
- final int b = (barr[i] & 0xff);
- if (a != b) {
- return a - b;
- }
- }
-
- return aarr.length - barr.length;
- }
-
- /**
- * Adds the BLOB key to the given {@link MessageDigest}.
- *
- * @param md
- * the message digest to add the BLOB key to
- */
- public void addToMessageDigest(final MessageDigest md) {
-
- md.update(this.key);
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/ChannelDeploymentDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/ChannelDeploymentDescriptor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/ChannelDeploymentDescriptor.java
index 2c90091..6fff4a5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/ChannelDeploymentDescriptor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/ChannelDeploymentDescriptor.java
@@ -29,6 +29,8 @@ import org.apache.flink.runtime.io.network.channels.ChannelID;
*/
public final class ChannelDeploymentDescriptor implements Serializable {
+ private static final long serialVersionUID = -4079084629425460213L;
+
/** The ID of the output channel. */
private final ChannelID outputChannelID;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/GateDeploymentDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/GateDeploymentDescriptor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/GateDeploymentDescriptor.java
index 1f73546..c676023 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/GateDeploymentDescriptor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/GateDeploymentDescriptor.java
@@ -29,6 +29,8 @@ import org.apache.flink.runtime.executiongraph.ExecutionEdge;
*/
public final class GateDeploymentDescriptor implements Serializable {
+ private static final long serialVersionUID = -8433936680266802364L;
+
/** The list of channel deployment descriptors attached to this gate. */
private final List<ChannelDeploymentDescriptor> channels;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptor.java
index 20157d9..75a3c9b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptor.java
@@ -34,6 +34,8 @@ import org.apache.flink.runtime.jobgraph.JobVertexID;
*/
public final class TaskDeploymentDescriptor implements Serializable {
+ private static final long serialVersionUID = -3233562176034358530L;
+
/** The ID of the job the tasks belongs to. */
private final JobID jobID;
@@ -192,6 +194,15 @@ public final class TaskDeploymentDescriptor implements Serializable {
public int getCurrentNumberOfSubtasks() {
return this.currentNumberOfSubtasks;
}
+
+ /**
+ * Gets the number of the slot into which the task is to be deployed.
+ *
+ * @return The number of the target slot.
+ */
+ public int getTargetSlotNumber() {
+ return targetSlotNumber;
+ }
/**
* Returns the configuration of the job the task belongs to.
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/DistributedFileStatus.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/DistributedFileStatus.java b/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/DistributedFileStatus.java
index 6b819dc..d8c9873 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/DistributedFileStatus.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/DistributedFileStatus.java
@@ -16,7 +16,6 @@
* limitations under the License.
*/
-
package org.apache.flink.runtime.fs.hdfs;
import org.apache.flink.core.fs.FileStatus;
@@ -25,7 +24,6 @@ import org.apache.flink.core.fs.Path;
/**
* Concrete implementation of the {@link FileStatus} interface for the
* Hadoop Distribution File System.
- *
*/
public final class DistributedFileStatus implements FileStatus {
@@ -43,13 +41,11 @@ public final class DistributedFileStatus implements FileStatus {
@Override
public long getLen() {
-
return fileStatus.getLen();
}
@Override
public long getBlockSize() {
-
long blocksize = fileStatus.getBlockSize();
if (blocksize > fileStatus.getLen()) {
return fileStatus.getLen();
@@ -60,19 +56,16 @@ public final class DistributedFileStatus implements FileStatus {
@Override
public long getAccessTime() {
-
return fileStatus.getAccessTime();
}
@Override
public long getModificationTime() {
-
return fileStatus.getModificationTime();
}
@Override
public short getReplication() {
-
return fileStatus.getReplication();
}
@@ -82,13 +75,12 @@ public final class DistributedFileStatus implements FileStatus {
@Override
public Path getPath() {
-
return new Path(fileStatus.getPath().toString());
}
+ @SuppressWarnings("deprecation")
@Override
public boolean isDir() {
-
return fileStatus.isDir();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/ConnectionInfoLookupResponse.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/ConnectionInfoLookupResponse.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/ConnectionInfoLookupResponse.java
index 547ce70..51a0f94 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/ConnectionInfoLookupResponse.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/ConnectionInfoLookupResponse.java
@@ -16,7 +16,6 @@
* limitations under the License.
*/
-
package org.apache.flink.runtime.io.network;
import java.io.Serializable;
@@ -25,6 +24,9 @@ import org.apache.flink.runtime.io.network.channels.ChannelID;
public class ConnectionInfoLookupResponse implements Serializable {
+ private static final long serialVersionUID = 3961171754642077522L;
+
+
private enum ReturnCode {
NOT_FOUND, FOUND_AND_RECEIVER_READY, FOUND_BUT_RECEIVER_NOT_READY, JOB_IS_ABORTING
};
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/RemoteReceiver.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/RemoteReceiver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/RemoteReceiver.java
index 78dcbae..436d07d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/RemoteReceiver.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/RemoteReceiver.java
@@ -32,6 +32,8 @@ import org.apache.flink.core.memory.DataOutputView;
*/
public final class RemoteReceiver implements IOReadableWritable, Serializable {
+ private static final long serialVersionUID = 4304924747853162443L;
+
/**
* The address of the connection to the remote TaskManager.
*/
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
index 3497824..a288357 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
@@ -43,10 +43,12 @@ import org.apache.flink.runtime.blob.BlobKey;
*/
public class JobGraph implements Serializable {
+ private static final long serialVersionUID = 1L;
+
// --------------------------------------------------------------------------------------------
// Members that define the structure / topology of the graph
// --------------------------------------------------------------------------------------------
-
+
/** List of task vertices included in this job graph. */
private final Map<JobVertexID, AbstractJobVertex> taskVertices = new LinkedHashMap<JobVertexID, AbstractJobVertex>();
@@ -54,7 +56,7 @@ public class JobGraph implements Serializable {
private final Configuration jobConfiguration = new Configuration();
/** Set of JAR files required to run this job. */
- private final transient List<Path> userJars = new ArrayList<Path>();
+ private final List<Path> userJars = new ArrayList<Path>();
/** Set of blob keys identifying the JAR files required to run this job. */
private final List<BlobKey> userJarBlobKeys = new ArrayList<BlobKey>();
@@ -357,7 +359,6 @@ public class JobGraph implements Serializable {
* @return set of BLOB keys referring to the JAR files required to run this job
*/
public List<BlobKey> getUserJarBlobKeys() {
-
return this.userJarBlobKeys;
}
@@ -369,15 +370,13 @@ public class JobGraph implements Serializable {
* @throws IOException
* thrown if an I/O error occurs during the upload
*/
- public void uploadRequiredJarFiles(final InetSocketAddress serverAddress) throws IOException {
-
+ public void uploadRequiredJarFiles(InetSocketAddress serverAddress) throws IOException {
if (this.userJars.isEmpty()) {
return;
}
BlobClient bc = null;
try {
-
bc = new BlobClient(serverAddress);
for (final Path jar : this.userJars) {
@@ -388,14 +387,15 @@ public class JobGraph implements Serializable {
is = fs.open(jar);
final BlobKey key = bc.put(is);
this.userJarBlobKeys.add(key);
- } finally {
+ }
+ finally {
if (is != null) {
is.close();
}
}
}
-
- } finally {
+ }
+ finally {
if (bc != null) {
bc.close();
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/SubSlot.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/SubSlot.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/SubSlot.java
index 0b19a5f..92d457b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/SubSlot.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/SubSlot.java
@@ -23,6 +23,9 @@ import org.apache.flink.runtime.instance.AllocatedSlot;
public class SubSlot extends AllocatedSlot {
+ private static final long serialVersionUID = 1361615219044538497L;
+
+
private final SharedSlot sharedSlot;
private final AbstractID groupId;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/web/WebInfoServer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/web/WebInfoServer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/web/WebInfoServer.java
index 7e67603..24dbaf7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/web/WebInfoServer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/web/WebInfoServer.java
@@ -85,15 +85,16 @@ public class WebInfoServer {
*/
public WebInfoServer(Configuration config, ActorRef jobmanager,
ActorRef archive, FiniteDuration timeout) throws IOException {
- this.port = config.getInteger(ConfigConstants.JOB_MANAGER_WEB_PORT_KEY,
- ConfigConstants.DEFAULT_JOB_MANAGER_WEB_FRONTEND_PORT);
-
- this.timeout = timeout;
-
+
// if no explicit configuration is given, use the global configuration
if (config == null) {
config = GlobalConfiguration.getConfiguration();
}
+
+ this.port = config.getInteger(ConfigConstants.JOB_MANAGER_WEB_PORT_KEY,
+ ConfigConstants.DEFAULT_JOB_MANAGER_WEB_FRONTEND_PORT);
+
+ this.timeout = timeout;
// get base path of Flink installation
final String basePath = config.getString(ConfigConstants.FLINK_BASE_DIR_PATH_KEY, "");
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/main/java/org/apache/flink/runtime/net/NetUtils.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/net/NetUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/net/NetUtils.java
index 2795158..ec2633c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/net/NetUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/net/NetUtils.java
@@ -16,18 +16,9 @@
* limitations under the License.
*/
-
-/**
- * This file is based on source code from the Hadoop Project (http://hadoop.apache.org/), licensed by the Apache
- * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
- * additional information regarding copyright ownership.
- */
-
package org.apache.flink.runtime.net;
import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
import java.net.Inet4Address;
import java.net.InetAddress;
import java.net.InetSocketAddress;
@@ -35,286 +26,15 @@ import java.net.NetworkInterface;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.net.UnknownHostException;
-import java.nio.channels.SocketChannel;
-import java.util.ArrayList;
-import java.util.Collection;
import java.util.Enumeration;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-
-import javax.net.SocketFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class NetUtils {
+
private static final Logger LOG = LoggerFactory.getLogger(NetUtils.class);
- private static Map<String, String> hostToResolved = new HashMap<String, String>();
-
- public static SocketFactory getSocketFactory() {
-
- return getDefaultSocketFactory();
- }
-
- public static SocketFactory getDefaultSocketFactory() {
-
- return SocketFactory.getDefault();
- }
-
- /**
- * Util method to build socket addr from either:
- * <host>:<post>
- * <fs>://<host>:<port>/<path>
- */
- public static InetSocketAddress createSocketAddr(String target) {
- return createSocketAddr(target, -1);
- }
-
- /**
- * Util method to build socket addr from either:
- * <host>
- * <host>:<post>
- * <fs>://<host>:<port>/<path>
- */
- public static InetSocketAddress createSocketAddr(String target, int defaultPort) {
- int colonIndex = target.indexOf(':');
- if (colonIndex < 0 && defaultPort == -1) {
- throw new RuntimeException("Not a host:port pair: " + target);
- }
- String hostname = "";
- int port = -1;
- if (!target.contains("/")) {
- if (colonIndex == -1) {
- hostname = target;
- } else {
- // must be the old style <host>:<port>
- hostname = target.substring(0, colonIndex);
- port = Integer.parseInt(target.substring(colonIndex + 1));
- }
- } else {
- // a new uri
- try {
- URI addr = new URI(target);
- hostname = addr.getHost();
- port = addr.getPort();
- } catch (URISyntaxException use) {
- LOG.error("Invalid URI syntax.", use);
- }
- }
-
- if (port == -1) {
- port = defaultPort;
- }
-
- if (getStaticResolution(hostname) != null) {
- hostname = getStaticResolution(hostname);
- }
- return new InetSocketAddress(hostname, port);
- }
-
- /**
- * Adds a static resolution for host. This can be used for setting up
- * hostnames with names that are fake to point to a well known host. For e.g.
- * in some testcases we require to have daemons with different hostnames
- * running on the same machine. In order to create connections to these
- * daemons, one can set up mappings from those hostnames to "localhost".
- * {@link NetUtils#getStaticResolution(String)} can be used to query for
- * the actual hostname.
- *
- * @param host
- * @param resolvedName
- */
- public static void addStaticResolution(String host, String resolvedName) {
- synchronized (hostToResolved) {
- hostToResolved.put(host, resolvedName);
- }
- }
-
- /**
- * Retrieves the resolved name for the passed host. The resolved name must
- * have been set earlier using {@link NetUtils#addStaticResolution(String, String)}
- *
- * @param host
- * @return the resolution
- */
- public static String getStaticResolution(String host) {
- synchronized (hostToResolved) {
- return hostToResolved.get(host);
- }
- }
-
- /**
- * This is used to get all the resolutions that were added using
- * {@link NetUtils#addStaticResolution(String, String)}. The return
- * value is a List each element of which contains an array of String
- * of the form String[0]=hostname, String[1]=resolved-hostname
- *
- * @return the list of resolutions
- */
- public static List<String[]> getAllStaticResolutions() {
- synchronized (hostToResolved) {
- Set<Entry<String, String>> entries = hostToResolved.entrySet();
- if (entries.size() == 0) {
- return null;
- }
- List<String[]> l = new ArrayList<String[]>(entries.size());
- for (Entry<String, String> e : entries) {
- l.add(new String[] { e.getKey(), e.getValue() });
- }
- return l;
- }
- }
-
- /**
- * Same as getInputStream(socket, socket.getSoTimeout()).<br>
- * <br>
- * From documentation for {@link #getInputStream(Socket, long)}:<br>
- * Returns InputStream for the socket. If the socket has an associated
- * SocketChannel then it returns a {@link SocketInputStream} with the given timeout. If the socket does not
- * have a channel, {@link Socket#getInputStream()} is returned. In the later
- * case, the timeout argument is ignored and the timeout set with {@link Socket#setSoTimeout(int)} applies for
- * reads.<br>
- * <br>
- * Any socket created using socket factories returned by {@link #NetUtils},
- * must use this interface instead of {@link Socket#getInputStream()}.
- *
- * @see #getInputStream(Socket, long)
- * @param socket
- * @return InputStream for reading from the socket.
- * @throws IOException
- */
- public static InputStream getInputStream(Socket socket) throws IOException {
- return getInputStream(socket, socket.getSoTimeout());
- }
-
- /**
- * Returns InputStream for the socket. If the socket has an associated
- * SocketChannel then it returns a {@link SocketInputStream} with the given timeout. If the socket does not
- * have a channel, {@link Socket#getInputStream()} is returned. In the later
- * case, the timeout argument is ignored and the timeout set with {@link Socket#setSoTimeout(int)} applies for
- * reads.<br>
- * <br>
- * Any socket created using socket factories returned by {@link #NetUtils},
- * must use this interface instead of {@link Socket#getInputStream()}.
- *
- * @see Socket#getChannel()
- * @param socket
- * @param timeout
- * timeout in milliseconds. This may not always apply. zero
- * for waiting as long as necessary.
- * @return InputStream for reading from the socket.
- * @throws IOException
- */
- public static InputStream getInputStream(Socket socket, long timeout) throws IOException {
- return (socket.getChannel() == null) ? socket.getInputStream() : new SocketInputStream(socket, timeout);
- }
-
- /**
- * Same as getOutputStream(socket, 0). Timeout of zero implies write will
- * wait until data is available.<br>
- * <br>
- * From documentation for {@link #getOutputStream(Socket, long)} : <br>
- * Returns OutputStream for the socket. If the socket has an associated
- * SocketChannel then it returns a {@link SocketOutputStream} with the given timeout. If the socket does not
- * have a channel, {@link Socket#getOutputStream()} is returned. In the later
- * case, the timeout argument is ignored and the write will wait until
- * data is available.<br>
- * <br>
- * Any socket created using socket factories returned by {@link #NetUtils},
- * must use this interface instead of {@link Socket#getOutputStream()}.
- *
- * @see #getOutputStream(Socket, long)
- * @param socket
- * @return OutputStream for writing to the socket.
- * @throws IOException
- */
- public static OutputStream getOutputStream(Socket socket) throws IOException {
- return getOutputStream(socket, 0);
- }
-
- /**
- * Returns OutputStream for the socket. If the socket has an associated
- * SocketChannel then it returns a {@link SocketOutputStream} with the given timeout. If the socket does not
- * have a channel, {@link Socket#getOutputStream()} is returned. In the later
- * case, the timeout argument is ignored and the write will wait until
- * data is available.<br>
- * <br>
- * Any socket created using socket factories returned by {@link #NetUtils},
- * must use this interface instead of {@link Socket#getOutputStream()}.
- *
- * @see Socket#getChannel()
- * @param socket
- * @param timeout
- * timeout in milliseconds. This may not always apply. zero
- * for waiting as long as necessary.
- * @return OutputStream for writing to the socket.
- * @throws IOException
- */
- public static OutputStream getOutputStream(Socket socket, long timeout) throws IOException {
- return (socket.getChannel() == null) ? socket.getOutputStream() : new SocketOutputStream(socket, timeout);
- }
-
- public static void connect(Socket socket, SocketAddress endpoint, int timeout) throws IOException {
- if (socket == null || endpoint == null || timeout < 0) {
- throw new IllegalArgumentException("Illegal argument for connect()");
- }
-
- SocketChannel ch = socket.getChannel();
-
- if (ch == null) {
- // let the default implementation handle it.
- socket.connect(endpoint, timeout);
- } else {
- SocketIOWithTimeout.connect(ch, endpoint, timeout);
- }
- }
-
- /**
- * Given a string representation of a host, return its ip address
- * in textual presentation.
- *
- * @param name
- * a string representation of a host:
- * either a textual representation its IP address or its host name
- * @return its IP address in the string format
- */
- public static String normalizeHostName(String name) {
- if (Character.digit(name.charAt(0), 16) != -1) { // it is an IP
- return name;
- } else {
- try {
- InetAddress ipAddress = InetAddress.getByName(name);
- return ipAddress.getHostAddress();
- } catch (UnknownHostException e) {
- return name;
- }
- }
- }
-
- /**
- * Given a collection of string representation of hosts, return a list of
- * corresponding IP addresses in the textual representation.
- *
- * @param names
- * a collection of string representations of hosts
- * @return a list of corresponding IP addresses in the string format
- * @see #normalizeHostName(String)
- */
- public static List<String> normalizeHostNames(Collection<String> names) {
- List<String> hostNames = new ArrayList<String>(names.size());
- for (String name : names) {
- hostNames.add(normalizeHostName(name));
- }
- return hostNames;
- }
-
/**
* The states of address detection mechanism.
* There is only a state transition if the current state failed to determine the address.
@@ -344,11 +64,14 @@ public class NetUtils {
while (true) {
Enumeration<NetworkInterface> e = NetworkInterface.getNetworkInterfaces();
+
while (e.hasMoreElements()) {
NetworkInterface n = e.nextElement();
Enumeration<InetAddress> ee = n.getInetAddresses();
+
while (ee.hasMoreElements()) {
InetAddress i = ee.nextElement();
+
switch (strategy) {
case ADDRESS:
if (hasCommonPrefix(jobManagerAddress.getAddress().getAddress(), i.getAddress())) {
@@ -358,6 +81,7 @@ public class NetUtils {
}
}
break;
+
case FAST_CONNECT:
case SLOW_CONNECT:
boolean correct = tryToConnect(i, jobManagerAddress, strategy.getTimeout());
@@ -366,17 +90,22 @@ public class NetUtils {
return i;
}
break;
+
case HEURISTIC:
- LOG.debug("ResolveAddress using heuristic strategy for " + i + " with" +
- "isLinkLocalAddress:" + i.isLinkLocalAddress() +" " +
- "isLoopbackAddress:" + i.isLoopbackAddress() + ".");
- if(!i.isLinkLocalAddress() && !i.isLoopbackAddress() && i instanceof Inet4Address){
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("ResolveAddress using heuristic strategy for " + i + " with" +
+ " isLinkLocalAddress:" + i.isLinkLocalAddress() +
+ " isLoopbackAddress:" + i.isLoopbackAddress() + ".");
+ }
+
+ if (!i.isLinkLocalAddress() && !i.isLoopbackAddress() && i instanceof Inet4Address){
LOG.warn("Hostname " + InetAddress.getLocalHost().getHostName() + " resolves to " +
"loopback address. Using instead " + i.getHostAddress() + " on network " +
"interface " + n.getName() + ".");
return i;
}
break;
+
default:
throw new RuntimeException("Unkown address detection strategy: " + strategy);
}
@@ -391,11 +120,11 @@ public class NetUtils {
strategy = AddressDetectionState.SLOW_CONNECT;
break;
case SLOW_CONNECT:
- if(!InetAddress.getLocalHost().isLoopbackAddress()){
+ if (!InetAddress.getLocalHost().isLoopbackAddress()) {
LOG.info("Heuristically taking " + InetAddress.getLocalHost() + " as own " +
"IP address.");
return InetAddress.getLocalHost();
- }else {
+ } else {
strategy = AddressDetectionState.HEURISTIC;
break;
}
@@ -437,7 +166,8 @@ public class NetUtils {
LOG.debug("Failed with exception", ex);
}
connectable = false;
- } finally {
+ }
+ finally {
if (socket != null) {
socket.close();
}
@@ -454,13 +184,18 @@ public class NetUtils {
if (port != 0) {
return port;
}
- } catch (IOException e) {
- LOG.debug("Unable to allocate port " + e.getMessage(), e);
- } finally {
+ }
+ catch (IOException e) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Unable to allocate port " + e.getMessage(), e);
+ }
+ }
+ finally {
if (serverSocket != null) {
try {
serverSocket.close();
} catch (Throwable t) {
+ // ignored
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/main/java/org/apache/flink/runtime/net/SocketIOWithTimeout.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/net/SocketIOWithTimeout.java b/flink-runtime/src/main/java/org/apache/flink/runtime/net/SocketIOWithTimeout.java
deleted file mode 100644
index daecfa5..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/net/SocketIOWithTimeout.java
+++ /dev/null
@@ -1,457 +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.
- */
-
-
-/**
- * This file is based on source code from the Hadoop Project (http://hadoop.apache.org/), licensed by the Apache
- * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
- * additional information regarding copyright ownership.
- */
-
-package org.apache.flink.runtime.net;
-
-import java.io.IOException;
-import java.io.InterruptedIOException;
-import java.net.SocketAddress;
-import java.net.SocketTimeoutException;
-import java.nio.ByteBuffer;
-import java.nio.channels.SelectableChannel;
-import java.nio.channels.SelectionKey;
-import java.nio.channels.Selector;
-import java.nio.channels.SocketChannel;
-import java.nio.channels.spi.SelectorProvider;
-import java.util.Iterator;
-import java.util.LinkedList;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * This supports input and output streams for a socket channels.
- * These streams can have a timeout.
- */
-abstract class SocketIOWithTimeout {
- // This is intentionally package private.
-
- static final Logger LOG = LoggerFactory.getLogger(SocketIOWithTimeout.class);
-
- private SelectableChannel channel;
-
- private long timeout;
-
- private boolean closed = false;
-
- private static SelectorPool selector = new SelectorPool();
-
- /*
- * A timeout value of 0 implies wait for ever.
- * We should have a value of timeout that implies zero wait.. i.e.
- * read or write returns immediately.
- * This will set channel to non-blocking.
- */
- SocketIOWithTimeout(SelectableChannel channel, long timeout)
- throws IOException {
- checkChannelValidity(channel);
-
- this.channel = channel;
- this.timeout = timeout;
- // Set non-blocking
- channel.configureBlocking(false);
- }
-
- void close() {
- closed = true;
- }
-
- boolean isOpen() {
- return !closed && channel.isOpen();
- }
-
- SelectableChannel getChannel() {
- return channel;
- }
-
- /**
- * Utility function to check if channel is ok.
- * Mainly to throw IOException instead of runtime exception
- * in case of mismatch. This mismatch can occur for many runtime
- * reasons.
- */
- static void checkChannelValidity(Object channel) throws IOException {
- if (channel == null) {
- /*
- * Most common reason is that original socket does not have a channel.
- * So making this an IOException rather than a RuntimeException.
- */
- throw new IOException("Channel is null. Check " + "how the channel or socket is created.");
- }
-
- if (!(channel instanceof SelectableChannel)) {
- throw new IOException("Channel should be a SelectableChannel");
- }
- }
-
- /**
- * Performs actual IO operations. This is not expected to block.
- *
- * @param buf
- * @return number of bytes (or some equivalent). 0 implies underlying
- * channel is drained completely. We will wait if more IO is
- * required.
- * @throws IOException
- */
- abstract int performIO(ByteBuffer buf) throws IOException;
-
- /**
- * Performs one IO and returns number of bytes read or written.
- * It waits up to the specified timeout. If the channel is
- * not read before the timeout, SocketTimeoutException is thrown.
- *
- * @param buf
- * buffer for IO
- * @param ops
- * Selection Ops used for waiting. Suggested values:
- * SelectionKey.OP_READ while reading and SelectionKey.OP_WRITE while
- * writing.
- * @return number of bytes read or written. negative implies end of stream.
- * @throws IOException
- */
- int doIO(ByteBuffer buf, int ops) throws IOException {
-
- /*
- * For now only one thread is allowed. If user want to read or write
- * from multiple threads, multiple streams could be created. In that
- * case multiple threads work as well as underlying channel supports it.
- */
- if (!buf.hasRemaining()) {
- throw new IllegalArgumentException("Buffer has no data left.");
- // or should we just return 0?
- }
-
- while (buf.hasRemaining()) {
- if (closed) {
- return -1;
- }
-
- try {
- int n = performIO(buf);
- if (n != 0) {
- // successful io or an error.
- return n;
- }
- } catch (IOException e) {
- if (!channel.isOpen()) {
- closed = true;
- }
- throw e;
- }
-
- // now wait for socket to be ready.
- int count = 0;
- try {
- count = selector.select(channel, ops, timeout);
- } catch (IOException e) { // unexpected IOException.
- closed = true;
- throw e;
- }
-
- if (count == 0) {
- throw new SocketTimeoutException(timeoutExceptionString(channel, timeout, ops));
- }
- // otherwise the socket should be ready for io.
- }
-
- return 0; // does not reach here.
- }
-
- /**
- * The contract is similar to {@link SocketChannel#connect(SocketAddress)} with a timeout.
- *
- * @see SocketChannel#connect(SocketAddress)
- * @param channel
- * - this should be a {@link SelectableChannel}
- * @param endpoint
- * @throws IOException
- */
- static void connect(SocketChannel channel, SocketAddress endpoint, int timeout) throws IOException {
-
- boolean blockingOn = channel.isBlocking();
- if (blockingOn) {
- channel.configureBlocking(false);
- }
-
- try {
- if (channel.connect(endpoint)) {
- return;
- }
-
- long timeoutLeft = timeout;
- long endTime = (timeout > 0) ? (System.currentTimeMillis() + timeout) : 0;
-
- while (true) {
- // we might have to call finishConnect() more than once
- // for some channels (with user level protocols)
-
- int ret = selector.select((SelectableChannel) channel, SelectionKey.OP_CONNECT, timeoutLeft);
-
- if (ret > 0 && channel.finishConnect()) {
- return;
- }
-
- if (ret == 0 || (timeout > 0 && (timeoutLeft = (endTime - System.currentTimeMillis())) <= 0)) {
- throw new SocketTimeoutException(timeoutExceptionString(channel, timeout, SelectionKey.OP_CONNECT));
- }
- }
- } catch (IOException e) {
- // javadoc for SocketChannel.connect() says channel should be closed.
- try {
- channel.close();
- } catch (IOException ignored) {
- }
- throw e;
- } finally {
- if (blockingOn && channel.isOpen()) {
- channel.configureBlocking(true);
- }
- }
- }
-
- /**
- * This is similar to {@link #doIO(ByteBuffer, int)} except that it
- * does not perform any I/O. It just waits for the channel to be ready
- * for I/O as specified in ops.
- *
- * @param ops
- * Selection Ops used for waiting
- * @throws SocketTimeoutException
- * if select on the channel times out.
- * @throws IOException
- * if any other I/O error occurs.
- */
- void waitForIO(int ops) throws IOException {
-
- if (selector.select(channel, ops, timeout) == 0) {
- throw new SocketTimeoutException(timeoutExceptionString(channel, timeout, ops));
- }
- }
-
- private static String timeoutExceptionString(SelectableChannel channel, long timeout, int ops) {
-
- String waitingFor;
- switch (ops) {
-
- case SelectionKey.OP_READ:
- waitingFor = "read";
- break;
-
- case SelectionKey.OP_WRITE:
- waitingFor = "write";
- break;
-
- case SelectionKey.OP_CONNECT:
- waitingFor = "connect";
- break;
-
- default:
- waitingFor = "" + ops;
- }
-
- return timeout + " millis timeout while " + "waiting for channel to be ready for " + waitingFor + ". ch : "
- + channel;
- }
-
- /**
- * This maintains a pool of selectors. These selectors are closed
- * once they are idle (unused) for a few seconds.
- */
- private static class SelectorPool {
-
- private static class SelectorInfo {
- Selector selector;
-
- long lastActivityTime;
-
- LinkedList<SelectorInfo> queue;
-
- void close() {
- if (selector != null) {
- try {
- selector.close();
- } catch (IOException e) {
- LOG.warn("Unexpected exception while closing selector : " + e.toString());
- }
- }
- }
- }
-
- private static class ProviderInfo {
- SelectorProvider provider;
-
- LinkedList<SelectorInfo> queue; // lifo
-
- ProviderInfo next;
- }
-
- private static final long IDLE_TIMEOUT = 10 * 1000; // 10 seconds.
-
- private ProviderInfo providerList = null;
-
- /**
- * Waits on the channel with the given timeout using one of the
- * cached selectors. It also removes any cached selectors that are
- * idle for a few seconds.
- *
- * @param channel
- * @param ops
- * @param timeout
- * @return
- * @throws IOException
- */
- int select(SelectableChannel channel, int ops, long timeout) throws IOException {
-
- SelectorInfo info = get(channel);
-
- SelectionKey key = null;
- int ret = 0;
-
- try {
- while (true) {
- long start = (timeout == 0) ? 0 : System.currentTimeMillis();
-
- key = channel.register(info.selector, ops);
- ret = info.selector.select(timeout);
-
- if (ret != 0) {
- return ret;
- }
-
- /*
- * Sometimes select() returns 0 much before timeout for
- * unknown reasons. So select again if required.
- */
- if (timeout > 0) {
- timeout -= System.currentTimeMillis() - start;
- if (timeout <= 0) {
- return 0;
- }
- }
-
- if (Thread.currentThread().isInterrupted()) {
- throw new InterruptedIOException("Interruped while waiting for " + "IO on channel " + channel
- + ". " + timeout + " millis timeout left.");
- }
- }
- } finally {
- if (key != null) {
- key.cancel();
- }
-
- // clear the canceled key.
- try {
- info.selector.selectNow();
- } catch (IOException e) {
- LOG.info("Unexpected Exception while clearing selector : " + e.toString());
- // don't put the selector back.
- info.close();
- return ret;
- }
-
- release(info);
- }
- }
-
- /**
- * Takes one selector from end of LRU list of free selectors.
- * If there are no selectors awailable, it creates a new selector.
- * Also invokes trimIdleSelectors().
- *
- * @param channel
- * @return
- * @throws IOException
- */
- private synchronized SelectorInfo get(SelectableChannel channel) throws IOException {
- SelectorInfo selInfo = null;
-
- SelectorProvider provider = channel.provider();
-
- // pick the list : rarely there is more than one provider in use.
- ProviderInfo pList = providerList;
- while (pList != null && pList.provider != provider) {
- pList = pList.next;
- }
- if (pList == null) {
- // LOG.info("Creating new ProviderInfo : " + provider.toString());
- pList = new ProviderInfo();
- pList.provider = provider;
- pList.queue = new LinkedList<SelectorInfo>();
- pList.next = providerList;
- providerList = pList;
- }
-
- LinkedList<SelectorInfo> queue = pList.queue;
-
- if (queue.isEmpty()) {
- Selector selector = provider.openSelector();
- selInfo = new SelectorInfo();
- selInfo.selector = selector;
- selInfo.queue = queue;
- } else {
- selInfo = queue.removeLast();
- }
-
- trimIdleSelectors(System.currentTimeMillis());
- return selInfo;
- }
-
- /**
- * puts selector back at the end of LRU list of free selectos.
- * Also invokes trimIdleSelectors().
- *
- * @param info
- */
- private synchronized void release(SelectorInfo info) {
- long now = System.currentTimeMillis();
- trimIdleSelectors(now);
- info.lastActivityTime = now;
- info.queue.addLast(info);
- }
-
- /**
- * Closes selectors that are idle for IDLE_TIMEOUT (10 sec). It does not
- * traverse the whole list, just over the one that have crossed
- * the timeout.
- */
- private void trimIdleSelectors(long now) {
- long cutoff = now - IDLE_TIMEOUT;
-
- for (ProviderInfo pList = providerList; pList != null; pList = pList.next) {
- if (pList.queue.isEmpty()) {
- continue;
- }
- for (Iterator<SelectorInfo> it = pList.queue.iterator(); it.hasNext();) {
- SelectorInfo info = it.next();
- if (info.lastActivityTime > cutoff) {
- break;
- }
- it.remove();
- info.close();
- }
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/main/java/org/apache/flink/runtime/net/SocketInputStream.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/net/SocketInputStream.java b/flink-runtime/src/main/java/org/apache/flink/runtime/net/SocketInputStream.java
deleted file mode 100644
index eb22ad0..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/net/SocketInputStream.java
+++ /dev/null
@@ -1,179 +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.
- */
-
-
-/**
- * This file is based on source code from the Hadoop Project (http://hadoop.apache.org/), licensed by the Apache
- * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
- * additional information regarding copyright ownership.
- */
-
-package org.apache.flink.runtime.net;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.net.Socket;
-import java.nio.ByteBuffer;
-import java.nio.channels.ReadableByteChannel;
-import java.nio.channels.SelectableChannel;
-import java.nio.channels.SelectionKey;
-
-/**
- * This implements an input stream that can have a timeout while reading.
- * This sets non-blocking flag on the socket channel.
- * So after create this object, read() on {@link Socket#getInputStream()} and write() on
- * {@link Socket#getOutputStream()} for the associated socket will throw
- * IllegalBlockingModeException.
- * Please use {@link SocketOutputStream} for writing.
- */
-public class SocketInputStream extends InputStream implements ReadableByteChannel {
-
- private Reader reader;
-
- private static class Reader extends SocketIOWithTimeout {
- ReadableByteChannel channel;
-
- Reader(ReadableByteChannel channel, long timeout)
- throws IOException {
- super((SelectableChannel) channel, timeout);
- this.channel = channel;
- }
-
- int performIO(ByteBuffer buf) throws IOException {
- return channel.read(buf);
- }
- }
-
- /**
- * Create a new input stream with the given timeout. If the timeout
- * is zero, it will be treated as infinite timeout. The socket's
- * channel will be configured to be non-blocking.
- *
- * @param channel
- * Channel for reading, should also be a {@link SelectableChannel}.
- * The channel will be configured to be non-blocking.
- * @param timeout
- * timeout in milliseconds. must not be negative.
- * @throws IOException
- */
- public SocketInputStream(ReadableByteChannel channel, long timeout)
- throws IOException {
- SocketIOWithTimeout.checkChannelValidity(channel);
- reader = new Reader(channel, timeout);
- }
-
- /**
- * Same as SocketInputStream(socket.getChannel(), timeout): <br>
- * <br>
- * Create a new input stream with the given timeout. If the timeout
- * is zero, it will be treated as infinite timeout. The socket's
- * channel will be configured to be non-blocking.
- *
- * @see SocketInputStream#SocketInputStream(ReadableByteChannel, long)
- * @param socket
- * should have a channel associated with it.
- * @param timeout
- * timeout timeout in milliseconds. must not be negative.
- * @throws IOException
- */
- public SocketInputStream(Socket socket, long timeout)
- throws IOException {
- this(socket.getChannel(), timeout);
- }
-
- /**
- * Same as SocketInputStream(socket.getChannel(), socket.getSoTimeout())
- * :<br>
- * <br>
- * Create a new input stream with the given timeout. If the timeout
- * is zero, it will be treated as infinite timeout. The socket's
- * channel will be configured to be non-blocking.
- *
- * @see SocketInputStream#SocketInputStream(ReadableByteChannel, long)
- * @param socket
- * should have a channel associated with it.
- * @throws IOException
- */
- public SocketInputStream(Socket socket)
- throws IOException {
- this(socket.getChannel(), socket.getSoTimeout());
- }
-
- @Override
- public int read() throws IOException {
- /*
- * Allocation can be removed if required.
- * probably no need to optimize or encourage single byte read.
- */
- byte[] buf = new byte[1];
- int ret = read(buf, 0, 1);
- if (ret > 0) {
- return (byte) buf[0];
- }
- if (ret != -1) {
- // unexpected
- throw new IOException("Could not read from stream");
- }
- return ret;
- }
-
- public int read(byte[] b, int off, int len) throws IOException {
- return read(ByteBuffer.wrap(b, off, len));
- }
-
- public synchronized void close() throws IOException {
- /*
- * close the channel since Socket.getInputStream().close()
- * closes the socket.
- */
- reader.channel.close();
- reader.close();
- }
-
- /**
- * Returns underlying channel used by inputstream.
- * This is useful in certain cases like channel for
- * {@link java.nio.channels.FileChannel#transferFrom(ReadableByteChannel, long, long)}.
- */
- public ReadableByteChannel getChannel() {
- return reader.channel;
- }
-
- // ReadableByteChannel interface
-
- public boolean isOpen() {
- return reader.isOpen();
- }
-
- public int read(ByteBuffer dst) throws IOException {
- return reader.doIO(dst, SelectionKey.OP_READ);
- }
-
- /**
- * waits for the underlying channel to be ready for reading.
- * The timeout specified for this stream applies to this wait.
- *
- * @throws SocketTimeoutException
- * if select on the channel times out.
- * @throws IOException
- * if any other I/O error occurs.
- */
- public void waitForReadable() throws IOException {
- reader.waitForIO(SelectionKey.OP_READ);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/main/java/org/apache/flink/runtime/net/SocketOutputStream.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/net/SocketOutputStream.java b/flink-runtime/src/main/java/org/apache/flink/runtime/net/SocketOutputStream.java
deleted file mode 100644
index 2df4bbe..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/net/SocketOutputStream.java
+++ /dev/null
@@ -1,222 +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.
- */
-
-
-/**
- * This file is based on source code from the Hadoop Project (http://hadoop.apache.org/), licensed by the Apache
- * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
- * additional information regarding copyright ownership.
- */
-
-package org.apache.flink.runtime.net;
-
-import java.io.EOFException;
-import java.io.IOException;
-import java.io.OutputStream;
-import java.net.Socket;
-import java.nio.ByteBuffer;
-import java.nio.channels.FileChannel;
-import java.nio.channels.SelectableChannel;
-import java.nio.channels.SelectionKey;
-import java.nio.channels.WritableByteChannel;
-
-/**
- * This implements an output stream that can have a timeout while writing.
- * This sets non-blocking flag on the socket channel.
- * So after creating this object , read() on {@link Socket#getInputStream()} and write() on
- * {@link Socket#getOutputStream()} on the associated socket will throw
- * llegalBlockingModeException.
- * Please use {@link SocketInputStream} for reading.
- */
-public class SocketOutputStream extends OutputStream implements WritableByteChannel {
-
- private Writer writer;
-
- private static class Writer extends SocketIOWithTimeout {
- WritableByteChannel channel;
-
- Writer(WritableByteChannel channel, long timeout)
- throws IOException {
- super((SelectableChannel) channel, timeout);
- this.channel = channel;
- }
-
- int performIO(ByteBuffer buf) throws IOException {
- return channel.write(buf);
- }
- }
-
- /**
- * Create a new ouput stream with the given timeout. If the timeout
- * is zero, it will be treated as infinite timeout. The socket's
- * channel will be configured to be non-blocking.
- *
- * @param channel
- * Channel for writing, should also be a {@link SelectableChannel}.
- * The channel will be configured to be non-blocking.
- * @param timeout
- * timeout in milliseconds. must not be negative.
- * @throws IOException
- */
- public SocketOutputStream(WritableByteChannel channel, long timeout)
- throws IOException {
- SocketIOWithTimeout.checkChannelValidity(channel);
- writer = new Writer(channel, timeout);
- }
-
- /**
- * Same as SocketOutputStream(socket.getChannel(), timeout):<br>
- * <br>
- * Create a new ouput stream with the given timeout. If the timeout
- * is zero, it will be treated as infinite timeout. The socket's
- * channel will be configured to be non-blocking.
- *
- * @see SocketOutputStream#SocketOutputStream(WritableByteChannel, long)
- * @param socket
- * should have a channel associated with it.
- * @param timeout
- * timeout timeout in milliseconds. must not be negative.
- * @throws IOException
- */
- public SocketOutputStream(Socket socket, long timeout)
- throws IOException {
- this(socket.getChannel(), timeout);
- }
-
- public void write(int b) throws IOException {
- /*
- * If we need to, we can optimize this allocation.
- * probably no need to optimize or encourage single byte writes.
- */
- final byte[] buf = new byte[1];
- buf[0] = (byte) b;
- write(buf, 0, 1);
- }
-
- public void write(byte[] b, int off, int len) throws IOException {
- final ByteBuffer buf = ByteBuffer.wrap(b, off, len);
- while (buf.hasRemaining()) {
- try {
- if (write(buf) < 0) {
- throw new IOException("The stream is closed");
- }
- } catch (IOException e) {
- /*
- * Unlike read, write can not inform user of partial writes.
- * So will close this if there was a partial write.
- */
- if (buf.capacity() > buf.remaining()) {
- writer.close();
- }
- throw e;
- }
- }
- }
-
- public synchronized void close() throws IOException {
- /*
- * close the channel since Socket.getOuputStream().close()
- * closes the socket.
- */
- writer.channel.close();
- writer.close();
- }
-
- /**
- * Returns underlying channel used by this stream.
- * This is useful in certain cases like channel for {@link FileChannel#transferTo(long, long, WritableByteChannel)}
- */
- public WritableByteChannel getChannel() {
- return writer.channel;
- }
-
- // WritableByteChannle interface
-
- public boolean isOpen() {
- return writer.isOpen();
- }
-
- public int write(ByteBuffer src) throws IOException {
- return writer.doIO(src, SelectionKey.OP_WRITE);
- }
-
- /**
- * waits for the underlying channel to be ready for writing.
- * The timeout specified for this stream applies to this wait.
- *
- * @throws SocketTimeoutException
- * if select on the channel times out.
- * @throws IOException
- * if any other I/O error occurs.
- */
- public void waitForWritable() throws IOException {
- writer.waitForIO(SelectionKey.OP_WRITE);
- }
-
- /**
- * Transfers data from FileChannel using {@link FileChannel#transferTo(long, long, WritableByteChannel)}.
- * Similar to readFully(), this waits till requested amount of
- * data is transfered.
- *
- * @param fileCh
- * FileChannel to transfer data from.
- * @param position
- * position within the channel where the transfer begins
- * @param count
- * number of bytes to transfer.
- * @throws EOFException
- * If end of input file is reached before requested number of
- * bytes are transfered.
- * @throws SocketTimeoutException
- * If this channel blocks transfer longer than timeout for
- * this stream.
- * @throws IOException
- * Includes any exception thrown by {@link FileChannel#transferTo(long, long, WritableByteChannel)}.
- */
- public void transferToFully(FileChannel fileCh, long position, int count) throws IOException {
-
- while (count > 0) {
- /*
- * Ideally we should wait after transferTo returns 0. But because of
- * a bug in JRE on Linux (http://bugs.sun.com/view_bug.do?bug_id=5103988),
- * which throws an exception instead of returning 0, we wait for the
- * channel to be writable before writing to it. If you ever see
- * IOException with message "Resource temporarily unavailable"
- * thrown here, please let us know.
- * Once we move to JAVA SE 7, wait should be moved to correct place.
- */
- waitForWritable();
- final int nTransfered = (int) fileCh.transferTo(position, count, getChannel());
-
- if (nTransfered == 0) {
- // check if end of file is reached.
- if (position >= fileCh.size()) {
- throw new EOFException("EOF Reached. file size is " + fileCh.size() + " and " + count
- + " more bytes left to be " + "transfered.");
- }
- // otherwise assume the socket is full.
- // waitForWritable(); // see comment above.
- } else if (nTransfered < 0) {
- throw new IOException("Unexpected return of " + nTransfered + " from transferTo()");
- } else {
- position += nTransfered;
- count -= nTransfered;
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/main/scala/org/apache/flink/runtime/ActorLogMessages.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/ActorLogMessages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/ActorLogMessages.scala
index 892e68d..c8ec180 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/ActorLogMessages.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/ActorLogMessages.scala
@@ -30,13 +30,17 @@ trait ActorLogMessages {
override def isDefinedAt(x: Any): Boolean = _receiveWithLogMessages.isDefinedAt(x)
override def apply(x: Any):Unit = {
- if(log.isDebugEnabled) {
- log.debug(s"Received message ${x} from ${self.sender}.")
+ if (!log.isDebugEnabled) {
+ _receiveWithLogMessages(x)
}
- val start = System.nanoTime()
- _receiveWithLogMessages(x)
- val duration = (System.nanoTime() - start) / 1000000
- if(log.isDebugEnabled) {
+ else {
+ log.debug(s"Received message ${x} from ${self.sender}.")
+
+ val start = System.nanoTime()
+
+ _receiveWithLogMessages(x)
+
+ val duration = (System.nanoTime() - start) / 1000000
log.debug(s"Handled message ${x} in ${duration} ms from ${self.sender}.")
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/serialization/IOReadableWritableSerializer.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/serialization/IOReadableWritableSerializer.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/serialization/IOReadableWritableSerializer.scala
index dff72bb..3a906b1 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/serialization/IOReadableWritableSerializer.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/serialization/IOReadableWritableSerializer.scala
@@ -24,7 +24,7 @@ import org.apache.flink.runtime.io.network.serialization.{DataInputDeserializer,
import org.apache.flink.util.InstantiationUtil
class IOReadableWritableSerializer extends JSerializer {
- val INITIAL_BUFFER_SIZE = 8096
+ val INITIAL_BUFFER_SIZE = 256
override protected def fromBinaryJava(bytes: Array[Byte], manifest: Class[_]): AnyRef = {
val in = new DataInputDeserializer(bytes, 0, bytes.length)
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/972a7b01/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/serialization/WritableSerializer.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/serialization/WritableSerializer.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/serialization/WritableSerializer.scala
index 0ac7ed6..be1ef17 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/serialization/WritableSerializer.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/serialization/WritableSerializer.scala
@@ -25,7 +25,7 @@ import org.apache.flink.util.InstantiationUtil
import org.apache.hadoop.io.Writable
class WritableSerializer extends JSerializer {
- val INITIAL_BUFFER_SIZE = 8096
+ val INITIAL_BUFFER_SIZE = 256
override protected def fromBinaryJava(bytes: Array[Byte], manifest: Class[_]): AnyRef = {
val in = new DataInputDeserializer(bytes, 0, bytes.length)
[4/4] incubator-flink git commit: [FLINK-1335] [docs] Add guide how
to set up Eclipse, and bring README.md up to speed.
Posted by se...@apache.org.
[FLINK-1335] [docs] Add guide how to set up Eclipse, and bring README.md up to speed.
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/2b86e953
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/2b86e953
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/2b86e953
Branch: refs/heads/master
Commit: 2b86e95368644794d3f7e319e53421f5e08e3a7f
Parents: cec30ff
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Jan 6 10:57:26 2015 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Jan 6 11:28:55 2015 +0100
----------------------------------------------------------------------
README.md | 71 ++++++++++++++++++++++++---
docs/internal_howto.md | 2 +
docs/internal_setup_eclipse.md | 98 +++++++++++++++++++++++++++++++++++++
3 files changed, 163 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/2b86e953/README.md
----------------------------------------------------------------------
diff --git a/README.md b/README.md
index 248d8d4..c0fae3d 100644
--- a/README.md
+++ b/README.md
@@ -1,18 +1,35 @@
# Apache Flink
+Apache Flink is an open source system for fast and versatile data analytics in clusters. Flink supports batch and streaming analytics,
+in one system. Analytical programs can be written in concise and elegant APIs in Java and Scala.
-Apache Flink is an open source system for expressive, declarative, fast, and efficient data analysis. Flink combines the scalability and programming flexibility of distributed MapReduce-like platforms with the efficiency, out-of-core execution, and query optimization capabilities found in parallel databases.
+```scala
+case class WordWithCount(word: String, count: Int)
+val text = env.readTextFile(path)
-Learn more about Flink at http://flink.incubator.apache.org/
+val counts = text.flatMap { _.split("\\W+") }
+ .map { WordWithCount(_, 1) }
+ .groupBy("word")
+ .sum("count")
+
+counts.writeAsCsv(outputPath)
+```
+
+Flink is highlighted by some unique features:
+
+* Hybrid batch/streaming runtime that supports batch processing and data streaming programs.
+* Custom memory management to guarantee efficient, adaptive, and highly robust switching between in-memory and data processing out-of-core algorithms.
+* Flexible and expressive windowing semantics for data stream programs
+* Built-in program optimizer that chooses the proper runtime operations for each program
+* Custom type analysis and serialization stack for high performance
-## Build Apache Flink
+Learn more about Flink at http://flink.incubator.apache.org/
-### Build From Source
+## Building Apache Flink from Source
-#### Requirements
* Unix-like environment (We use Linux, Mac OS X, Cygwin)
* git
* Maven (at least version 3.0.4)
@@ -27,8 +44,46 @@ mvn clean package -DskipTests # this will take up to 5 minutes
Flink is now installed in `flink-dist/target`
+## Developing Flink
+
+The Flink committers use the IntelliJ IDE and Eclipse to develop the Flink codebase.
+
+Minimal requirements for an IDE are:
+* Support for Java and Scala (also mixed projects)
+* Support for Maven with Java and Scala
+
+
+### IntelliJ
+
+The IntelliJ IDE supports Maven out of the box and offers a plugin for Scala development.
+
+* IntelliJ download: [https://www.jetbrains.com/idea/](https://www.jetbrains.com/idea/)
+* IntelliJ Scala Plugin: [http://plugins.jetbrains.com/plugin/?id=1347](http://plugins.jetbrains.com/plugin/?id=1347)
+
+### Eclipse Scala IDE
+
+We recommend using Scala IDE 3.0.3, based on Eclipse Kepler. While this is a slightly older version,
+we found it to be the verstion that works most robustly for a complex project like Flink.
+
+Further details, and a guide to newer Scala IDE versions can be found in the
+[How to setup Eclipse](https://github.com/StephanEwen/incubator-flink/blob/master/docs/internal_setup_eclipse.md) docs.
+
+**Note:** Before following this setup, make sure to run the build from the command line once
+(`mvn clean package -DskipTests`, see above)
+
+1. Download the Scala IDE (preferred) or install the plugin to Eclipse Kepler. See [here]([How to setup Eclipse](apache/incubator-flink/blob/master/docs/internal_setup_eclipse.md)
+ for download links and instructions.
+2. Add the "macroparadise" compiler plugin to the Scala compiler.
+ Open "Window" -> "Preferences" -> "Scala" -> "Compiler" -> "Advanced" and put into the "Xplugin" field the path to
+ the *macroparadise* jar file (typically "/home/*-your-user-*/.m2/repository/org/scalamacros/paradise_2.10.4/2.0.1/paradise_2.10.4-2.0.1.jar").
+ Note: If you do not have the jar file, you probably did not ran the command line build.
+3. Import the Flink Maven projects ("File" -> "Import" -> "Maven" -> "Existing Maven Projects")
+4. During the import, Eclipse will ask to automatically install additional Maven build helper plugins.
+5. Close the "flink-java8" project. Since Eclipse Kepler does not support Java 8, you cannot develop this project.
+
## Support
+
Don’t hesitate to ask!
Please contact the developers on our [mailing lists](http://flink.incubator.apache.org/community.html#mailing-lists) if you need help.
@@ -38,7 +93,8 @@ Please contact the developers on our [mailing lists](http://flink.incubator.apac
## Documentation
-The documentation of Apache Flink is located on the website: http://flink.incubator.apache.org or in the `docs/` directory of the source code.
+The documentation of Apache Flink is located on the website: [http://flink.incubator.apache.org](http://flink.incubator.apache.org)
+or in the `docs/` directory of the source code.
## Fork and Contribute
@@ -50,6 +106,5 @@ This article describes [how to contribute to Apache Flink](http://flink.incubato
## About
-Apache Flink is an effort undergoing incubation at The Apache Software Foundation (ASF), sponsored by the Apache Incubator PMC. Incubation is required of all newly accepted projects until a further review indicates that the infrastructure, communications, and decision making process have stabilized in a manner consistent with other successful ASF projects. While incubation status is not necessarily a reflection of the completeness or stability of the code, it does indicate that the project has yet to be fully endorsed by the ASF.
-
+Apache Flink is an open source project of The Apache Software Foundation (ASF).
The Apache Flink project originated from the [Stratosphere](http://stratosphere.eu) research project.
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/2b86e953/docs/internal_howto.md
----------------------------------------------------------------------
diff --git a/docs/internal_howto.md b/docs/internal_howto.md
index 415bda7..1aa1c9b 100644
--- a/docs/internal_howto.md
+++ b/docs/internal_howto.md
@@ -31,6 +31,8 @@ that are interested in the technology behind Flink.
- [How-to: Using logging in Flink](internal_logging.html)
+- [How-to: Setting up Eclipse to develop Flink](internal_setup_eclipse.html)
+
---
*This documentation is maintained by the contributors of the individual components.
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/2b86e953/docs/internal_setup_eclipse.md
----------------------------------------------------------------------
diff --git a/docs/internal_setup_eclipse.md b/docs/internal_setup_eclipse.md
new file mode 100644
index 0000000..3b6e6a5
--- /dev/null
+++ b/docs/internal_setup_eclipse.md
@@ -0,0 +1,98 @@
+---
+title: "How to set up Eclipse"
+---
+<!--
+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.
+-->
+
+A brief guide how to set up Eclipse for development of the Flink core.
+Flink uses mixed Scala/Java projects, which pose a challenge to some IDEs.
+Below is the setup guide that works best from our personal experience.
+
+For Eclipse users, we currently recomment the Scala IDE 3.0.3, as the most robust solution.
+
+
+## Eclipse Scala IDE 3.0.3
+
+**NOTE:** While this version of the Scala IDE is not the newest, we have found it to be the most reliably working
+version for complex projects like Flink. One restriction is, though, that it works only with Java 7, not with Java 8.
+
+**Note:** Before following this setup, make sure to run the build from the command line once
+(`mvn clean package -DskipTests`)
+
+1. Download the Scala IDE (preferred) or install the plugin to Eclipse Kepler. See section below for download links
+ and instructions.
+2. Add the "macroparadise" compiler plugin to the Scala compiler.
+ Open "Window" -> "Preferences" -> "Scala" -> "Compiler" -> "Advanced" and put into the "Xplugin" field the path to
+ the *macroparadise* jar file (typically "/home/*-your-user-*/.m2/repository/org/scalamacros/paradise_2.10.4/2.0.1/paradise_2.10.4-2.0.1.jar").
+ Note: If you do not have the jar file, you probably did not ran the command line build.
+3. Import the Flink Maven projects ("File" -> "Import" -> "Maven" -> "Existing Maven Projects")
+4. During the import, Eclipse will ask to automatically install additional Maven build helper plugins.
+5. Close the "flink-java8" project. Since Eclipse Kepler does not support Java 8, you cannot develop this project.
+
+
+### Download links for Scala IDE 3.0.3
+
+The Scala IDE 3.0.3 is a previous stable release, and download links are a bit hidden.
+
+The pre-packaged Scala IDE can be downloaded from the following links:
+
+* [Linux (64 bit)](http://downloads.typesafe.com/scalaide-pack/3.0.3.vfinal-210-20140327/scala-SDK-3.0.3-2.10-linux.gtk.x86_64.tar.gz)
+* [Linux (32 bit)](http://downloads.typesafe.com/scalaide-pack/3.0.3.vfinal-210-20140327/scala-SDK-3.0.3-2.10-linux.gtk.x86.tar.gz)
+* [MaxOS X Cocoa (64 bit)](http://downloads.typesafe.com/scalaide-pack/3.0.3.vfinal-210-20140327/scala-SDK-3.0.3-2.10-macosx.cocoa.x86_64.zip)
+* [MaxOS X Cocoa (32 bit)](http://downloads.typesafe.com/scalaide-pack/3.0.3.vfinal-210-20140327/scala-SDK-3.0.3-2.10-macosx.cocoa.x86.zip)
+* [Windows (64 bit)](http://downloads.typesafe.com/scalaide-pack/3.0.3.vfinal-210-20140327/scala-SDK-3.0.3-2.10-win32.win32.x86_64.zip)
+* [Windows (32 bit)](http://downloads.typesafe.com/scalaide-pack/3.0.3.vfinal-210-20140327/scala-SDK-3.0.3-2.10-win32.win32.x86.zip)
+
+Alternatively, you can download Eclipse Kepler from [https://eclipse.org/downloads/packages/release/Kepler/SR2](https://eclipse.org/downloads/packages/release/Kepler/SR2)
+and manually add the Scala and Maven plugins by plugin site at [http://scala-ide.org/download/prev-stable.html](http://scala-ide.org/download/prev-stable.html).
+
+* Either use the update site to install the plugin ("Help" -> "Install new Software")
+* Or download the [zip file](http://download.scala-ide.org/sdk/helium/e38/scala211/stable/update-site.zip), unpack it, and move the contents of the
+ "plugins" and "features" folders into the equally named folders of the Eclipse root directory
+
+
+## Eclipse Scala IDE 4.0.0
+
+**NOTE: From personal experience, the use of the Scala IDE 4.0.0 performs worse than previous versions for complex projects like Flink.**
+**Version 4.0.0 does not handle mixed Java/Scala projects as robustly and it frequently raises incorrect import and type errors.**
+
+*Note:* Before following this setup, make sure to run the build from the command line once
+(`mvn clean package -DskipTests`)
+
+1. Download the Scala IDE: [http://scala-ide.org/download/sdk.html](http://scala-ide.org/download/sdk.html)
+2. Import the Flink Maven projects (File -> Import -> Maven -> Existing Maven Projects)
+3. While importing the Flink project, the IDE may ask you to install an additional maven build helper plugin.
+4. After the import, you need to set the Scala version of your projects to Scala 2.10 (from the default 2.11).
+ To do that, select all projects that contain Scala code (marked by the small *S* on the project icon),
+ right click and select "Scala -> Set the Scala Installation" and pick "2.10.4".
+ Currently, the project to which that is relevant are "flink-runtime", "flink-scala", "flink-scala-examples",
+ "flink-streaming-example", "flink-streaming-scala", "flink-tests", "flink-test-utils", and "flink-yarn".
+5. Depending on your version of the Scala IDE, you may need to add the "macroparadise" compiler plugin to the
+ Scala compiler. Open "Window" -> "Preferences" -> "Scala" -> "Compiler" -> "Advanced" and put into the "Xplugin" field
+ the path to the *macroparadise* jar file (typically "/home/*-your-user-*/.m2/repository/org/scalamacros/paradise_2.10.4/2.0.1/paradise_2.10.4-2.0.1.jar")
+6. In order to compile the "flink-java-8" project, you may need to add a Java 8 execution environment.
+ See [this post](http://stackoverflow.com/questions/25391207/how-do-i-add-execution-environment-1-8-to-eclipse-luna)
+ for details.
+
+---
+
+*This documentation is maintained by the contributors of the individual components.
+We kindly ask anyone that adds and changes components to eventually provide a patch
+or pull request that updates these documents as well.*
+