You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@reef.apache.org by we...@apache.org on 2015/01/23 00:46:36 UTC
[03/51] [partial] incubator-reef git commit: [REEF-93] Move java
sources to lang/java
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tests/src/test/java/org/apache/reef/tests/statepassing/StatePassingTest.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/statepassing/StatePassingTest.java b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/statepassing/StatePassingTest.java
new file mode 100644
index 0000000..0a9bce5
--- /dev/null
+++ b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/statepassing/StatePassingTest.java
@@ -0,0 +1,71 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.tests.statepassing;
+
+import org.apache.reef.client.DriverConfiguration;
+import org.apache.reef.client.DriverLauncher;
+import org.apache.reef.client.LauncherStatus;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.exceptions.BindException;
+import org.apache.reef.tang.exceptions.InjectionException;
+import org.apache.reef.tests.TestEnvironment;
+import org.apache.reef.tests.TestEnvironmentFactory;
+import org.apache.reef.tests.library.driver.OnDriverStartedAllocateOne;
+import org.apache.reef.util.EnvironmentUtils;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Tests whether subsequent tasks can pass state from one to the other.
+ */
+public final class StatePassingTest {
+ private final TestEnvironment testEnvironment = TestEnvironmentFactory.getNewTestEnvironment();
+
+ @Before
+ public void setUp() throws Exception {
+ testEnvironment.setUp();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ this.testEnvironment.tearDown();
+ }
+
+ @Test
+ public void testStatePassing() throws BindException, InjectionException {
+
+ final Configuration runtimeConfiguration = this.testEnvironment.getRuntimeConfiguration();
+
+ final Configuration driverConfiguration = DriverConfiguration.CONF
+ .set(DriverConfiguration.GLOBAL_LIBRARIES, EnvironmentUtils.getClassLocation(this.getClass()))
+ .set(DriverConfiguration.DRIVER_IDENTIFIER, "StatePassingTest")
+ .set(DriverConfiguration.ON_DRIVER_STARTED, OnDriverStartedAllocateOne.class)
+ .set(DriverConfiguration.ON_EVALUATOR_ALLOCATED, StatePassingDriver.EvaluatorAllocatedHandler.class)
+ .set(DriverConfiguration.ON_CONTEXT_ACTIVE, StatePassingDriver.ContextActiveHandler.class)
+ .set(DriverConfiguration.ON_TASK_COMPLETED, StatePassingDriver.TaskCompletedHandler.class)
+ .build();
+
+ final LauncherStatus status = DriverLauncher.getLauncher(runtimeConfiguration)
+ .run(driverConfiguration, this.testEnvironment.getTestTimeout());
+
+ Assert.assertTrue("Job state after execution: " + status, status.isSuccess());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tests/src/test/java/org/apache/reef/tests/statepassing/package-info.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/statepassing/package-info.java b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/statepassing/package-info.java
new file mode 100644
index 0000000..7189809
--- /dev/null
+++ b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/statepassing/package-info.java
@@ -0,0 +1,22 @@
+/**
+ * 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.
+ */
+/**
+ * Tests whether subsequent tasks can pass state from one to the other.
+ */
+package org.apache.reef.tests.statepassing;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/ContextStartHandler1.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/ContextStartHandler1.java b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/ContextStartHandler1.java
new file mode 100644
index 0000000..9ca1bfb
--- /dev/null
+++ b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/ContextStartHandler1.java
@@ -0,0 +1,39 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.tests.subcontexts;
+
+import org.apache.reef.evaluator.context.events.ContextStart;
+import org.apache.reef.wake.EventHandler;
+
+import javax.inject.Inject;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+
+final class ContextStartHandler1 implements EventHandler<ContextStart> {
+
+ @Inject
+ ContextStartHandler1() {
+ }
+
+ @Override
+ public void onNext(final ContextStart contextStart) {
+ Logger.getLogger(this.getClass().getName()).log(Level.INFO, "ContextStart: " + contextStart);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/ContextStartHandler2.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/ContextStartHandler2.java b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/ContextStartHandler2.java
new file mode 100644
index 0000000..a298ad0
--- /dev/null
+++ b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/ContextStartHandler2.java
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.tests.subcontexts;
+
+import org.apache.reef.evaluator.context.events.ContextStart;
+import org.apache.reef.wake.EventHandler;
+
+import javax.inject.Inject;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+final class ContextStartHandler2 implements EventHandler<ContextStart> {
+
+ @Inject
+ ContextStartHandler2() {
+ }
+
+ @Override
+ public void onNext(final ContextStart contextStart) {
+ Logger.getLogger(this.getClass().getName()).log(Level.INFO, "ContextStart: " + contextStart);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/ContextStopHandler1.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/ContextStopHandler1.java b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/ContextStopHandler1.java
new file mode 100644
index 0000000..7d12bd2
--- /dev/null
+++ b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/ContextStopHandler1.java
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.tests.subcontexts;
+
+import org.apache.reef.evaluator.context.events.ContextStop;
+import org.apache.reef.wake.EventHandler;
+
+import javax.inject.Inject;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+final class ContextStopHandler1 implements EventHandler<ContextStop> {
+
+ @Inject
+ ContextStopHandler1() {
+ }
+
+ @Override
+ public void onNext(final ContextStop contextStop) {
+ Logger.getLogger(this.getClass().getName()).log(Level.INFO, "ContextStop: " + contextStop);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/ContextStopHandler2.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/ContextStopHandler2.java b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/ContextStopHandler2.java
new file mode 100644
index 0000000..d4ebd89
--- /dev/null
+++ b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/ContextStopHandler2.java
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.tests.subcontexts;
+
+import org.apache.reef.evaluator.context.events.ContextStop;
+import org.apache.reef.wake.EventHandler;
+
+import javax.inject.Inject;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+final class ContextStopHandler2 implements EventHandler<ContextStop> {
+
+ @Inject
+ ContextStopHandler2() {
+ }
+
+ @Override
+ public void onNext(final ContextStop contextStop) {
+ Logger.getLogger(this.getClass().getName()).log(Level.INFO, "ContextStop: " + contextStop);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/SubContextDriver.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/SubContextDriver.java b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/SubContextDriver.java
new file mode 100644
index 0000000..01eb271
--- /dev/null
+++ b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/SubContextDriver.java
@@ -0,0 +1,143 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.tests.subcontexts;
+
+import org.apache.reef.driver.context.ActiveContext;
+import org.apache.reef.driver.context.ClosedContext;
+import org.apache.reef.driver.context.ContextConfiguration;
+import org.apache.reef.driver.evaluator.AllocatedEvaluator;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.annotations.Unit;
+import org.apache.reef.tang.exceptions.BindException;
+import org.apache.reef.wake.EventHandler;
+
+import javax.inject.Inject;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+@Unit
+final class SubContextDriver {
+
+ private static final Logger LOG = Logger.getLogger(SubContextDriver.class.getName());
+
+ private static String CONTEXT_1_IDENTIFIER = "CONTEXT_1";
+ private static String CONTEXT_2_IDENTIFIER = "CONTEXT_2";
+
+ private State state = State.INIT; // lock: this
+
+ @Inject
+ SubContextDriver() {
+ }
+
+ private enum State {
+ INIT,
+ CONTEXT_1_SUBMITTED,
+ CONTEXT_2_SUBMITTED,
+ CONTEXT_2_CLOSED,
+ }
+
+ final class EvaluatorAllocatedHandler implements EventHandler<AllocatedEvaluator> {
+
+ @Override
+ public void onNext(final AllocatedEvaluator allocatedEvaluator) {
+
+ LOG.log(Level.FINE, "Submitting root context");
+
+ try {
+
+ final Configuration contextConfiguration = ContextConfiguration.CONF
+ .set(ContextConfiguration.ON_CONTEXT_STARTED, ContextStartHandler1.class)
+ .set(ContextConfiguration.ON_CONTEXT_STOP, ContextStopHandler1.class)
+ .set(ContextConfiguration.IDENTIFIER, CONTEXT_1_IDENTIFIER)
+ .build();
+
+ allocatedEvaluator.submitContext(contextConfiguration);
+
+ synchronized (SubContextDriver.this) {
+ SubContextDriver.this.state = State.CONTEXT_1_SUBMITTED;
+ }
+
+ } catch (final BindException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ final class ContextActiveHandler implements EventHandler<ActiveContext> {
+
+ @Override
+ public void onNext(final ActiveContext activeContext) {
+
+ LOG.log(Level.FINE, "Received ActiveContext: {0}", activeContext);
+
+ if (activeContext.getId().equals(CONTEXT_1_IDENTIFIER)) {
+
+ synchronized (SubContextDriver.this) {
+ assert (SubContextDriver.this.state == State.CONTEXT_1_SUBMITTED);
+ }
+
+ LOG.log(Level.FINE, "Submitting sub context");
+
+ final Configuration contextConfiguration = ContextConfiguration.CONF
+ .set(ContextConfiguration.ON_CONTEXT_STARTED, ContextStartHandler2.class)
+ .set(ContextConfiguration.ON_CONTEXT_STOP, ContextStopHandler2.class)
+ .set(ContextConfiguration.IDENTIFIER, CONTEXT_2_IDENTIFIER)
+ .build();
+
+ activeContext.submitContext(contextConfiguration);
+
+ synchronized (SubContextDriver.this) {
+ SubContextDriver.this.state = State.CONTEXT_2_SUBMITTED;
+ }
+
+
+ } else if (activeContext.getId().equals(CONTEXT_2_IDENTIFIER)) {
+ LOG.log(Level.INFO, "Received sub context. Closing");
+ activeContext.close();
+ }
+ }
+ }
+
+ final class ContextClosedHandler implements EventHandler<ClosedContext> {
+
+ @Override
+ public void onNext(final ClosedContext closedContext) {
+
+ LOG.log(Level.FINE, "Received ClosedContext: {0}", closedContext);
+
+ if (closedContext.getId().equals(CONTEXT_2_IDENTIFIER)) {
+
+ synchronized (SubContextDriver.this) {
+ assert (SubContextDriver.this.state == State.CONTEXT_2_SUBMITTED);
+ }
+
+ closedContext.getParentContext().close();
+ SubContextDriver.this.state = State.CONTEXT_2_CLOSED;
+
+ } else if (closedContext.getId().equals(CONTEXT_1_IDENTIFIER)) {
+
+ synchronized (SubContextDriver.this) {
+ assert (SubContextDriver.this.state == State.CONTEXT_2_CLOSED);
+ }
+
+ throw new IllegalStateException("Received a closed context for the root context");
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/SubContextTest.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/SubContextTest.java b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/SubContextTest.java
new file mode 100644
index 0000000..53d2735
--- /dev/null
+++ b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/SubContextTest.java
@@ -0,0 +1,93 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.tests.subcontexts;
+
+import org.apache.reef.client.DriverConfiguration;
+import org.apache.reef.client.DriverLauncher;
+import org.apache.reef.client.LauncherStatus;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.exceptions.BindException;
+import org.apache.reef.tang.exceptions.InjectionException;
+import org.apache.reef.tests.TestEnvironment;
+import org.apache.reef.tests.TestEnvironmentFactory;
+import org.apache.reef.tests.library.driver.OnDriverStartedAllocateOne;
+import org.apache.reef.util.EnvironmentUtils;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Tests the creation of context stacks on an Evaluator.
+ */
+public final class SubContextTest {
+ private final TestEnvironment testEnvironment = TestEnvironmentFactory.getNewTestEnvironment();
+
+ @Before
+ public void setUp() throws Exception {
+ testEnvironment.setUp();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ this.testEnvironment.tearDown();
+ }
+
+ @Test
+ public void testSubContexts() throws BindException, InjectionException {
+
+ final Configuration runtimeConfiguration = this.testEnvironment.getRuntimeConfiguration();
+
+ final Configuration driverConfiguration = DriverConfiguration.CONF
+ .set(DriverConfiguration.GLOBAL_LIBRARIES, EnvironmentUtils.getClassLocation(this.getClass()))
+ .set(DriverConfiguration.DRIVER_IDENTIFIER, "TEST_SubContextTest_testSubContexts")
+ .set(DriverConfiguration.ON_DRIVER_STARTED, OnDriverStartedAllocateOne.class)
+ .set(DriverConfiguration.ON_EVALUATOR_ALLOCATED, SubContextDriver.EvaluatorAllocatedHandler.class)
+ .set(DriverConfiguration.ON_CONTEXT_ACTIVE, SubContextDriver.ContextActiveHandler.class)
+ .set(DriverConfiguration.ON_CONTEXT_CLOSED, SubContextDriver.ContextClosedHandler.class)
+ .build();
+
+ final LauncherStatus status = DriverLauncher.getLauncher(runtimeConfiguration)
+ .run(driverConfiguration, this.testEnvironment.getTestTimeout());
+
+ Assert.assertTrue("Job state after execution: " + status, status.isSuccess());
+ }
+
+ /**
+ * Same as testSubContexts(), but using default ClosedContext handler.
+ */
+ @Test
+ public void testChainClose() throws BindException, InjectionException {
+
+ final Configuration runtimeConfiguration = this.testEnvironment.getRuntimeConfiguration();
+
+ final Configuration driverConfiguration = DriverConfiguration.CONF
+ .set(DriverConfiguration.GLOBAL_LIBRARIES, EnvironmentUtils.getClassLocation(this.getClass()))
+ .set(DriverConfiguration.DRIVER_IDENTIFIER, "TEST_SubContextTest_testChainClose")
+ .set(DriverConfiguration.ON_DRIVER_STARTED, OnDriverStartedAllocateOne.class)
+ .set(DriverConfiguration.ON_EVALUATOR_ALLOCATED, SubContextDriver.EvaluatorAllocatedHandler.class)
+ .set(DriverConfiguration.ON_CONTEXT_ACTIVE, SubContextDriver.ContextActiveHandler.class)
+ .build();
+
+ final LauncherStatus status = DriverLauncher.getLauncher(runtimeConfiguration)
+ .run(driverConfiguration, this.testEnvironment.getTestTimeout());
+
+ Assert.assertTrue("Job state after execution: " + status, status.isSuccess());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/package-info.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/package-info.java b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/package-info.java
new file mode 100644
index 0000000..4382e15
--- /dev/null
+++ b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/subcontexts/package-info.java
@@ -0,0 +1,22 @@
+/**
+ * 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.
+ */
+/**
+ * Tests the creation of context stacks on an Evaluator.
+ */
+package org.apache.reef.tests.subcontexts;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskcounting/TaskCountingDriver.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskcounting/TaskCountingDriver.java b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskcounting/TaskCountingDriver.java
new file mode 100644
index 0000000..947416b
--- /dev/null
+++ b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskcounting/TaskCountingDriver.java
@@ -0,0 +1,109 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.tests.taskcounting;
+
+import org.apache.reef.driver.evaluator.AllocatedEvaluator;
+import org.apache.reef.driver.task.CompletedTask;
+import org.apache.reef.driver.task.RunningTask;
+import org.apache.reef.driver.task.TaskConfiguration;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.annotations.Unit;
+import org.apache.reef.tests.library.exceptions.DriverSideFailure;
+import org.apache.reef.tests.library.tasks.NoopTask;
+import org.apache.reef.wake.EventHandler;
+import org.apache.reef.wake.time.event.StopTime;
+
+import javax.inject.Inject;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+
+@Unit
+final class TaskCountingDriver {
+
+ private final Set<String> expectedRunningTaskIds = new HashSet<>();
+ private AtomicInteger numberOfTaskSubmissions = new AtomicInteger(1000);
+
+ @Inject
+ TaskCountingDriver() {
+ }
+
+ private final Configuration getTaskConfiguration(final String taskId) {
+ return TaskConfiguration.CONF
+ .set(TaskConfiguration.IDENTIFIER, taskId)
+ .set(TaskConfiguration.TASK, NoopTask.class)
+ .build();
+ }
+
+ final class EvaluatorAllocatedHandler implements EventHandler<AllocatedEvaluator> {
+
+ @Override
+ public void onNext(final AllocatedEvaluator allocatedEvaluator) {
+ synchronized (expectedRunningTaskIds) {
+ final String taskId = "Task-" + numberOfTaskSubmissions.getAndDecrement();
+ final Configuration taskConfiguration = getTaskConfiguration(taskId);
+ allocatedEvaluator.submitTask(taskConfiguration);
+ expectedRunningTaskIds.add(taskId);
+ }
+ }
+ }
+
+ final class TaskRunningHandler implements EventHandler<RunningTask> {
+
+ @Override
+ public void onNext(final RunningTask runningTask) {
+ synchronized (expectedRunningTaskIds) {
+ final boolean isExpected = expectedRunningTaskIds.remove(runningTask.getId());
+ if (!isExpected) {
+ throw new DriverSideFailure("Unexpected RunningTask: " + runningTask.getId());
+ }
+ }
+ }
+ }
+
+ final class TaskCompletedHandler implements EventHandler<CompletedTask> {
+
+ @Override
+ public void onNext(final CompletedTask completedTask) {
+ synchronized (expectedRunningTaskIds) {
+ final int nextTaskNumber = numberOfTaskSubmissions.getAndDecrement();
+ if (nextTaskNumber > 0) {
+ final String taskId = "Task-" + nextTaskNumber;
+ completedTask.getActiveContext().submitTask(getTaskConfiguration(taskId));
+ expectedRunningTaskIds.add(taskId);
+ } else {
+ completedTask.getActiveContext().close();
+ }
+ }
+ }
+ }
+
+ final class DriverStopHandler implements EventHandler<StopTime> {
+
+ @Override
+ public void onNext(final StopTime stopTime) {
+ synchronized (expectedRunningTaskIds) {
+ if (!expectedRunningTaskIds.isEmpty()) {
+ throw new DriverSideFailure("Still expecting RunningTasks");
+ }
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskcounting/TaskCountingTest.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskcounting/TaskCountingTest.java b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskcounting/TaskCountingTest.java
new file mode 100644
index 0000000..0509185
--- /dev/null
+++ b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskcounting/TaskCountingTest.java
@@ -0,0 +1,70 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.tests.taskcounting;
+
+import org.apache.reef.client.DriverConfiguration;
+import org.apache.reef.client.DriverLauncher;
+import org.apache.reef.client.LauncherStatus;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.exceptions.InjectionException;
+import org.apache.reef.tests.TestEnvironment;
+import org.apache.reef.tests.TestEnvironmentFactory;
+import org.apache.reef.tests.library.driver.OnDriverStartedAllocateOne;
+import org.apache.reef.util.EnvironmentUtils;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Tests the submission of multiple tasks to an Evaluator in sequence.
+ */
+public final class TaskCountingTest {
+ private final TestEnvironment testEnvironment = TestEnvironmentFactory.getNewTestEnvironment();
+
+ @Before
+ public void setUp() throws Exception {
+ this.testEnvironment.setUp();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ this.testEnvironment.tearDown();
+ }
+
+ private Configuration getDriverConfiguration() {
+ return DriverConfiguration.CONF
+ .set(DriverConfiguration.GLOBAL_LIBRARIES, EnvironmentUtils.getClassLocation(this.getClass()))
+ .set(DriverConfiguration.DRIVER_IDENTIFIER, "TaskCounting")
+ .set(DriverConfiguration.ON_DRIVER_STARTED, OnDriverStartedAllocateOne.class)
+ .set(DriverConfiguration.ON_EVALUATOR_ALLOCATED, TaskCountingDriver.EvaluatorAllocatedHandler.class)
+ .set(DriverConfiguration.ON_TASK_RUNNING, TaskCountingDriver.TaskRunningHandler.class)
+ .set(DriverConfiguration.ON_TASK_COMPLETED, TaskCountingDriver.TaskCompletedHandler.class)
+ .set(DriverConfiguration.ON_DRIVER_STOP, TaskCountingDriver.DriverStopHandler.class)
+ .build();
+ }
+
+ @Test
+ public void testTaskCounting() throws InjectionException {
+ final LauncherStatus state = DriverLauncher.getLauncher(this.testEnvironment.getRuntimeConfiguration())
+ .run(getDriverConfiguration());
+
+ Assert.assertTrue("Job state after execution: " + state, state.isSuccess());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskcounting/package-info.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskcounting/package-info.java b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskcounting/package-info.java
new file mode 100644
index 0000000..c552fbd
--- /dev/null
+++ b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskcounting/package-info.java
@@ -0,0 +1,22 @@
+/**
+ * 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.
+ */
+/**
+ * Tests the submission of multiple tasks to an Evaluator in sequence.
+ */
+package org.apache.reef.tests.taskcounting;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskresubmit/TaskResubmitDriver.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskresubmit/TaskResubmitDriver.java b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskresubmit/TaskResubmitDriver.java
new file mode 100644
index 0000000..35ee102
--- /dev/null
+++ b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskresubmit/TaskResubmitDriver.java
@@ -0,0 +1,85 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.tests.taskresubmit;
+
+import org.apache.reef.driver.context.ActiveContext;
+import org.apache.reef.driver.evaluator.AllocatedEvaluator;
+import org.apache.reef.driver.task.FailedTask;
+import org.apache.reef.driver.task.TaskConfiguration;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.annotations.Unit;
+import org.apache.reef.tests.TestUtils;
+import org.apache.reef.tests.fail.task.FailTaskCall;
+import org.apache.reef.tests.library.exceptions.SimulatedTaskFailure;
+import org.apache.reef.tests.library.exceptions.TaskSideFailure;
+import org.apache.reef.wake.EventHandler;
+
+import javax.inject.Inject;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+@Unit
+class TaskResubmitDriver {
+
+ private static final Logger LOG = Logger.getLogger(TaskResubmitDriver.class.getName());
+
+ private int failuresSeen = 0;
+
+ @Inject
+ TaskResubmitDriver() {
+ }
+
+ private static Configuration getTaskConfiguration() {
+ return TaskConfiguration.CONF
+ .set(TaskConfiguration.TASK, FailTaskCall.class)
+ .set(TaskConfiguration.IDENTIFIER, "FailTask")
+ .build();
+ }
+
+ final class EvaluatorAllocatedHandler implements EventHandler<AllocatedEvaluator> {
+
+ @Override
+ public void onNext(final AllocatedEvaluator allocatedEvaluator) {
+ allocatedEvaluator.submitTask(getTaskConfiguration());
+ }
+ }
+
+ final class TaskFailedHandler implements EventHandler<FailedTask> {
+
+ @Override
+ public void onNext(final FailedTask failedTask) {
+
+ LOG.log(Level.INFO, "FailedTask: {0}", failedTask);
+
+ final Throwable ex = failedTask.getReason().get();
+ if (!TestUtils.hasCause(ex, SimulatedTaskFailure.class)) {
+ final String msg = "Expected SimulatedTaskFailure from " + failedTask.getId();
+ LOG.log(Level.SEVERE, msg, ex);
+ throw new TaskSideFailure(msg, ex);
+ }
+
+ final ActiveContext activeContext = failedTask.getActiveContext().get();
+ if (++TaskResubmitDriver.this.failuresSeen <= 1) { // resubmit the task
+ activeContext.submitTask(getTaskConfiguration());
+ } else { // Close the context
+ activeContext.close();
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskresubmit/TaskResubmitTest.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskresubmit/TaskResubmitTest.java b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskresubmit/TaskResubmitTest.java
new file mode 100644
index 0000000..d53ea46
--- /dev/null
+++ b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskresubmit/TaskResubmitTest.java
@@ -0,0 +1,72 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.tests.taskresubmit;
+
+import org.apache.reef.client.DriverConfiguration;
+import org.apache.reef.client.DriverLauncher;
+import org.apache.reef.client.LauncherStatus;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.exceptions.BindException;
+import org.apache.reef.tang.exceptions.InjectionException;
+import org.apache.reef.tests.TestEnvironment;
+import org.apache.reef.tests.TestEnvironmentFactory;
+import org.apache.reef.tests.library.driver.OnDriverStartedAllocateOne;
+import org.apache.reef.util.EnvironmentUtils;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+
+/**
+ * Tests whether (failed) tasks can be resubmitted with identical Configurations, including Task Ids.
+ */
+public final class TaskResubmitTest {
+
+ private final TestEnvironment testEnvironment = TestEnvironmentFactory.getNewTestEnvironment();
+
+ @Before
+ public void setUp() throws Exception {
+ this.testEnvironment.setUp();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ this.testEnvironment.tearDown();
+ }
+
+ @Test
+ public void testTaskResubmission() throws BindException, InjectionException {
+
+ final Configuration runtimeConfiguration = this.testEnvironment.getRuntimeConfiguration();
+
+ final Configuration driverConfiguration = DriverConfiguration.CONF
+ .set(DriverConfiguration.GLOBAL_LIBRARIES, EnvironmentUtils.getClassLocation(this.getClass()))
+ .set(DriverConfiguration.DRIVER_IDENTIFIER, "TEST_TaskResubmitTest")
+ .set(DriverConfiguration.ON_DRIVER_STARTED, OnDriverStartedAllocateOne.class)
+ .set(DriverConfiguration.ON_EVALUATOR_ALLOCATED, TaskResubmitDriver.EvaluatorAllocatedHandler.class)
+ .set(DriverConfiguration.ON_TASK_FAILED, TaskResubmitDriver.TaskFailedHandler.class)
+ .build();
+
+ final LauncherStatus state = DriverLauncher.getLauncher(runtimeConfiguration)
+ .run(driverConfiguration, this.testEnvironment.getTestTimeout());
+
+ Assert.assertTrue("Job state after execution: " + state, state.isSuccess());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskresubmit/package-info.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskresubmit/package-info.java b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskresubmit/package-info.java
new file mode 100644
index 0000000..a37cc1d
--- /dev/null
+++ b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/taskresubmit/package-info.java
@@ -0,0 +1,22 @@
+/**
+ * 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.
+ */
+/**
+ * Tests whether (failed) tasks can be resubmitted with identical Configurations, including Task Ids.
+ */
+package org.apache.reef.tests.taskresubmit;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-utils-hadoop/pom.xml
----------------------------------------------------------------------
diff --git a/lang/java/reef-utils-hadoop/pom.xml b/lang/java/reef-utils-hadoop/pom.xml
new file mode 100644
index 0000000..d1e4375
--- /dev/null
+++ b/lang/java/reef-utils-hadoop/pom.xml
@@ -0,0 +1,45 @@
+<?xml version="1.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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.reef</groupId>
+ <artifactId>reef-project</artifactId>
+ <version>0.11.0-incubating-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>reef-utils-hadoop</artifactId>
+ <name>REEF Utils for Hadoop</name>
+ <description>Utilities for using REEF on Hadoop.</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>reef-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-utils-hadoop/src/main/java/org/apache/reef/util/HadoopEnvironment.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-utils-hadoop/src/main/java/org/apache/reef/util/HadoopEnvironment.java b/lang/java/reef-utils-hadoop/src/main/java/org/apache/reef/util/HadoopEnvironment.java
new file mode 100644
index 0000000..513a10a
--- /dev/null
+++ b/lang/java/reef-utils-hadoop/src/main/java/org/apache/reef/util/HadoopEnvironment.java
@@ -0,0 +1,37 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.util;
+
+/**
+ * Constants for the various Hadoop environment variables.
+ */
+public class HadoopEnvironment {
+
+ public static final String HADOOP_CONF_DIR = "HADOOP_CONF_DIR";
+ public static final String HADOOP_HOME = "HADOOP_HOME";
+ public static final String HADOOP_COMMON_HOME = "HADOOP_COMMON_HOME";
+ public static final String HADOOP_YARN_HOME = "HADOOP_YARN_HOME";
+ public static final String HADOOP_HDFS_HOME = "HADOOP_HDFS_HOME";
+ public static final String HADOOP_MAPRED_HOME = "HADOOP_MAPRED_HOME";
+
+ private HadoopEnvironment() {
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-utils-hadoop/src/main/java/org/apache/reef/util/logging/DFSHandler.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-utils-hadoop/src/main/java/org/apache/reef/util/logging/DFSHandler.java b/lang/java/reef-utils-hadoop/src/main/java/org/apache/reef/util/logging/DFSHandler.java
new file mode 100644
index 0000000..64f024c
--- /dev/null
+++ b/lang/java/reef-utils-hadoop/src/main/java/org/apache/reef/util/logging/DFSHandler.java
@@ -0,0 +1,146 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.util.logging;
+
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import javax.inject.Inject;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.UnsupportedEncodingException;
+import java.util.logging.*;
+
+/**
+ * A java.util.logging.Handler that logs to (H)DFS.
+ */
+public final class DFSHandler extends Handler {
+
+ private static final String CLASS_NAME = DFSHandler.class.getName();
+ public static final String DFS_PATH_OPTION = CLASS_NAME + ".folder";
+ public static final String FORMATTER_OPTION = CLASS_NAME + ".formatter";
+ private final StreamHandler streamHandler;
+ private final OutputStream logOutputStream;
+
+ @Inject
+ public DFSHandler() throws IOException {
+ final LogManager logManager = LogManager.getLogManager();
+ final String fileName = logManager.getProperty(DFS_PATH_OPTION) + "/log.txt";
+ logOutputStream = FileSystem.get(new Configuration()).create(new Path(fileName), true);
+ final Formatter logFormatter = getInstance(logManager.getProperty(FORMATTER_OPTION), new SimpleFormatter());
+ this.streamHandler = new StreamHandler(logOutputStream, logFormatter);
+ }
+
+ private static final <T> T getInstance(final String className, final T defaultValue) {
+ try {
+ final Class aClass = ClassLoader.getSystemClassLoader().loadClass(className);
+ return (T) aClass.newInstance();
+ } catch (final Exception e) {
+ return defaultValue;
+ }
+
+ }
+
+ @Override
+ public Formatter getFormatter() {
+ return this.streamHandler.getFormatter();
+ }
+
+ @Override
+ public void setFormatter(final Formatter formatter) throws SecurityException {
+ this.streamHandler.setFormatter(formatter);
+ }
+
+ @Override
+ public String getEncoding() {
+ return this.streamHandler.getEncoding();
+ }
+
+ @Override
+ public void setEncoding(final String s) throws SecurityException, UnsupportedEncodingException {
+ this.streamHandler.setEncoding(s);
+ }
+
+ @Override
+ public Filter getFilter() {
+ return this.streamHandler.getFilter();
+ }
+
+ @Override
+ public void setFilter(final Filter filter) throws SecurityException {
+ this.streamHandler.setFilter(filter);
+ }
+
+ @Override
+ public ErrorManager getErrorManager() {
+ return this.streamHandler.getErrorManager();
+ }
+
+ @Override
+ public void setErrorManager(final ErrorManager errorManager) {
+ this.streamHandler.setErrorManager(errorManager);
+ }
+
+ @Override
+ protected void reportError(final String s, final Exception e, final int i) {
+ super.reportError(s, e, i);
+ }
+
+ @Override
+ public synchronized Level getLevel() {
+ return this.streamHandler.getLevel();
+ }
+
+ @Override
+ public synchronized void setLevel(final Level level) throws SecurityException {
+ this.streamHandler.setLevel(level);
+ }
+
+ @Override
+ public boolean isLoggable(final LogRecord logRecord) {
+ return this.streamHandler.isLoggable(logRecord);
+ }
+
+ @Override
+ public void publish(final LogRecord logRecord) {
+ this.streamHandler.publish(logRecord);
+ }
+
+ @Override
+ public void flush() {
+ this.streamHandler.flush();
+ try {
+ this.logOutputStream.flush();
+ } catch (final IOException e) {
+ // Eating it as it has nowhere to go.
+ }
+ }
+
+ @Override
+ public void close() throws SecurityException {
+ this.streamHandler.close();
+ try {
+ this.logOutputStream.close();
+ } catch (final IOException e) {
+ // Eating it as it has nowhere to go.
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-utils/pom.xml
----------------------------------------------------------------------
diff --git a/lang/java/reef-utils/pom.xml b/lang/java/reef-utils/pom.xml
new file mode 100644
index 0000000..62f4530
--- /dev/null
+++ b/lang/java/reef-utils/pom.xml
@@ -0,0 +1,45 @@
+<?xml version="1.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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.reef</groupId>
+ <artifactId>reef-project</artifactId>
+ <version>0.11.0-incubating-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>reef-utils</artifactId>
+ <name>REEF Utils</name>
+ <description>Utilities used across REEF modules.</description>
+
+ <!-- This module shouldn't have many dependencies to make sure it is broadly usable across reef subprojects -->
+ <dependencies>
+ <dependency>
+ <groupId>net.jcip</groupId>
+ <artifactId>jcip-annotations</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-utils/src/main/java/org/apache/reef/util/Optional.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-utils/src/main/java/org/apache/reef/util/Optional.java b/lang/java/reef-utils/src/main/java/org/apache/reef/util/Optional.java
new file mode 100644
index 0000000..6dc1a6e
--- /dev/null
+++ b/lang/java/reef-utils/src/main/java/org/apache/reef/util/Optional.java
@@ -0,0 +1,128 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.util;
+
+import net.jcip.annotations.Immutable;
+import net.jcip.annotations.ThreadSafe;
+
+import java.io.Serializable;
+
+/**
+ * Represents an optional value. Loosely based on
+ * <a href="http://download.java.net/jdk8/docs/api/java/util/Optional.html"></a>The Java 8 version</a>, but filtered for
+ * Java 7 compatibility.
+ */
+@Immutable
+@ThreadSafe
+public final class Optional<T> implements Serializable {
+
+ private static final long serialVersionUID = 42L;
+
+ private final T value;
+ private final String valueStr;
+ private final int valueHash;
+
+ private Optional(final T value) {
+ this.value = value;
+ this.valueStr = "Optional:{" + value + '}';
+ this.valueHash = value.hashCode();
+ }
+
+ private Optional() {
+ this.value = null;
+ this.valueStr = "OptionalvNothing";
+ this.valueHash = 0;
+ }
+
+ /**
+ * @return An Optional with the given value.
+ * @throws NullPointerException if the value is null
+ */
+ public static <T> Optional<T> of(final T value) throws NullPointerException {
+ if (null == value) {
+ throw new NullPointerException("Passed a null value. Use ofNullable() instead");
+ }
+ return new Optional<>(value);
+ }
+
+ /**
+ * @return an Optional with no value.
+ */
+ public static <T> Optional<T> empty() {
+ return new Optional<>();
+ }
+
+ /**
+ * @return An optional representing the given value, or an empty Optional.
+ */
+ public static <T> Optional<T> ofNullable(final T value) {
+ if (null == value) {
+ return Optional.empty();
+ } else {
+ return Optional.of(value);
+ }
+ }
+
+ /**
+ * @return the value represented or null, if isPresent() is false.
+ */
+ public T get() {
+ return this.value;
+ }
+
+ /**
+ * @param other
+ * @return the value of this Optional or other, if no value exists.
+ */
+ public T orElse(final T other) {
+ if (isPresent()) {
+ return this.get();
+ } else {
+ return other;
+ }
+ }
+
+ /**
+ * @return true if there is a value, false otherwise.
+ */
+ public boolean isPresent() {
+ return null != this.value;
+ }
+
+ @Override
+ public boolean equals(final Object obj) {
+
+ if (this == obj) return true;
+
+ if (obj == null || getClass() != obj.getClass()) return false;
+
+ final Optional that = (Optional) obj;
+ return this.value == that.value || (this.value != null && this.value.equals(that.value));
+ }
+
+ @Override
+ public int hashCode() {
+ return this.valueHash;
+ }
+
+ @Override
+ public String toString() {
+ return this.valueStr;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-utils/src/main/java/org/apache/reef/util/package-info.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-utils/src/main/java/org/apache/reef/util/package-info.java b/lang/java/reef-utils/src/main/java/org/apache/reef/util/package-info.java
new file mode 100644
index 0000000..1058756
--- /dev/null
+++ b/lang/java/reef-utils/src/main/java/org/apache/reef/util/package-info.java
@@ -0,0 +1,22 @@
+/**
+ * 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.
+ */
+/**
+ * Utilities used across reef modules. Subpackages are structred following the java.utils package.
+ */
+package org.apache.reef.util;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-utils/src/test/java/org/apache/reef/util/OptionalTest.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-utils/src/test/java/org/apache/reef/util/OptionalTest.java b/lang/java/reef-utils/src/test/java/org/apache/reef/util/OptionalTest.java
new file mode 100644
index 0000000..f643ab0
--- /dev/null
+++ b/lang/java/reef-utils/src/test/java/org/apache/reef/util/OptionalTest.java
@@ -0,0 +1,99 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.util;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.ArrayList;
+
+/**
+ * Tests for Optional.
+ */
+public class OptionalTest {
+
+ @Test
+ public void testEmpty() {
+ Assert.assertFalse("An empty Optional should return false to isPresent()",
+ Optional.empty().isPresent());
+ }
+
+ @Test
+ public void testOf() {
+ Assert.assertTrue("Optional.of() needs to return an Optional where isPresent() returns true",
+ Optional.of(2).isPresent());
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void testOfNull() {
+ final Optional<Integer> o = Optional.of(null);
+ }
+
+ @Test
+ public void testOfRandom() {
+ final double value = Math.random();
+ final Optional<Double> o = Optional.of(value);
+ Assert.assertEquals(value, (double) o.get(), 1e-12);
+ }
+
+ @Test
+ public void testOfNullable() {
+ Assert.assertFalse(Optional.ofNullable(null).isPresent());
+ Assert.assertTrue(Optional.ofNullable(1).isPresent());
+ Assert.assertEquals(Optional.ofNullable(1).get(), Integer.valueOf(1));
+ }
+
+ @Test
+ public void testOrElse() {
+ Assert.assertEquals(Optional.empty().orElse(2), 2);
+ Assert.assertEquals(Optional.of(1).orElse(2), Integer.valueOf(1));
+ }
+
+ @Test
+ public void testEquals() {
+ Assert.assertEquals(Optional.empty(), Optional.empty());
+ Assert.assertEquals(Optional.empty(), Optional.ofNullable(null));
+ Assert.assertEquals(Optional.of(1), Optional.of(1));
+ Assert.assertEquals(Optional.of("one"), Optional.of("one"));
+ Assert.assertFalse(Optional.of("one").equals(Optional.of("two")));
+ }
+
+ @Test
+ public void testEqualsCornerCases() {
+
+ // We lose type coercion:
+ Assert.assertFalse(Optional.of(1L).equals(Optional.of(1)));
+ Assert.assertTrue(1L == 1);
+ Assert.assertTrue(new Integer(1) == 1L);
+
+ // .equals() isn't typesafe, so we lose compile-time type checking:
+ Assert.assertFalse(Optional.of(1L).equals(1));
+
+ Assert.assertFalse(Optional.empty().equals(null));
+ Assert.assertFalse(Optional.of(3).equals(3));
+ Assert.assertFalse(Optional.of("one").equals(1));
+
+ // Assert.assertFalse("one" == 1); // incompatible operands; does not compile.
+
+ Assert.assertFalse(Optional.of(new ArrayList<>()).equals(Optional.of(new Object[]{})));
+
+ // Incompatible operands; does not compile, though == between objects is almost always a typo:
+ // Assert.assertFalse(new java.util.ArrayList() == new java.awt.List());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-wake/.gitattributes
----------------------------------------------------------------------
diff --git a/lang/java/reef-wake/.gitattributes b/lang/java/reef-wake/.gitattributes
new file mode 100644
index 0000000..db5b15f
--- /dev/null
+++ b/lang/java/reef-wake/.gitattributes
@@ -0,0 +1,3 @@
+# Commit text files using LF endings
+*.java text eol=lf whitespace=trailing-space,space-before-tab,tab-in-indent,blank-at-eof
+* text=auto
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-wake/.gitignore
----------------------------------------------------------------------
diff --git a/lang/java/reef-wake/.gitignore b/lang/java/reef-wake/.gitignore
new file mode 100644
index 0000000..a7927e6
--- /dev/null
+++ b/lang/java/reef-wake/.gitignore
@@ -0,0 +1,16 @@
+tmp
+bin
+tang.conf
+.DS_Store
+target
+generated
+.settings
+.classpath
+.project
+.sw[op]
+*.sw[op]
+.externalToolBuilders
+nbactions.xml
+.idea
+*.iml
+
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-wake/README.md
----------------------------------------------------------------------
diff --git a/lang/java/reef-wake/README.md b/lang/java/reef-wake/README.md
new file mode 100644
index 0000000..0b56cf0
--- /dev/null
+++ b/lang/java/reef-wake/README.md
@@ -0,0 +1,94 @@
+Wake
+====
+Wake is an event-driven framework based on ideas from SEDA, Click, Akka and Rx. It is *general purpose* in the sense that it is designed to support computationally intensive applications as well as high performance networking, storage, and legacy I/O systems. We implemented Wake to support high-performance, scalable analytical processing systems ("big data" applications), and have used it to implement control plane logic (which requires high fanout and low latency) and the data plane (which requires high-throughput processing as well).
+
+
+Background
+----------
+Wake applications consist of asynchronous *event handlers* that run inside of *stages*. Stages provide scheduling primitives such as thread pool sizing and performance isolation between event handlers. In addition to event handler and stage APIs, Wake includes profiling tools and a rich standard library of primitives for system builders.
+
+Event driven processing frameworks improve upon the performance of threaded architectures in two ways: (1) Event handlers often have lower memory and context switching overhead than threaded solutions, and (2) event driven systems allow applications to allocate and monitor computational and I/O resources in an extremely fine-grained fashion. Modern threading packages have done much to address the first concern, and have significantly lowered concurrency control and other implementation overheads in recent years. However, fine grained resource allocation remains a challenge in threaded systems, and is Wake's primary advantage over threading.
+
+Early event driven systems such as SEDA executed each event handler in a dedicated thread pool called a stage. This isolated low-latency event handlers (such as cache lookups) from expensive high-latency operations, such as disk I/O. With a single thread pool, high-latency I/O operations can easily monopolize the thread pool, causing all of the CPUs to block on disk I/O, even when there is computation to be scheduled. With separate thread pools, the operating system schedules I/O requests and computation separately, guaranteeing that runnable computations will not block on I/O requests.
+
+This is in contrast to event-driven systems such as the Click modular router that were designed to maximize throughput for predictable, low latency event-handlers. When possible, Click aggressively chains event handlers together, reducing the cost of an event dispatch to that of a function call, and allowing the compiler to perform optimizations such as inlining and constant propagation across event handlers.
+
+Wake allows developers to trade off between these two extremes by explicitly partitioning their event handlers into stages. Within a stage, event handlers engage in *thread-sharing* by simply calling each other directly. When an event crosses a stage boundary, it is placed in a queue of similar events. The queue is then drained by the threads managed by the receiving stage.
+
+Although event handling systems improve upon threaded performance in theory, they are notoriously difficult to reason about. We kept this in mind while designing Wake, and have gone to great pains to ensure that its APIs are simple and easy to implement without sacrificing our performance goals.
+
+Other event driven systems provide support for so-called *push-based* and *pull-based* event handlers. In push-based systems, event sources invoke event handlers that are exposed by the events' destinations, while pull-based APIs have the destination code invoke iterators to obtain the next available event from the source.
+
+Wake is completely push based. This eliminates the need for push and pull based variants of event handling logic, and also allowed us to unify all error handling in Wake into a single API. It is always possible to convert between push and pull based APIs by inserting a queue and a thread boundary between the push and pull based code. Wake supports libraries and applications that use this trick, since operating systems and legacy code sometimes expose pull-based APIs.
+
+Systems such as Rx allow event handlers to be dynamically registered and torn down at runtime, allowing applications to evolve over time. This leads to complicated setup and teardown protocols, where event handlers need to reason about the state of upstream and downstream handlers, both during setup and teardown, but also when routing messages at runtime. It also encourages design patterns such as dynamic event dispatching that break standard compiler optimizations. In contrast, Wake applications consist of immutable graphs of event handlers that are built up from sink to source. This ensures that, once an event handler has been instantiated, all downstream handlers are ready to receive messages.
+
+Wake is designed to work with Tang, a dependency injection system that focuses on configuration and debuggability. This makes it extremely easy to wire up complicated graphs of event handling logic. In addition to making it easy to build up event-driven applications, Tang provides a range of static analysis tools and provides a simple aspect-style programming facility that supports Wake's latency and throughput profilers.
+
+
+Core API
+--------
+
+### Event Handlers
+
+Wake provides two APIs for event handler implementations. The first is the [EventHandler](wake/src/main/java/org/apache/reef/wake/EventHandler.java) interface:
+```java
+public interface EventHandler<T> {
+ void onNext(T value);
+}
+```
+Callers of `onNext()` should assume that it is asynchronous, and that it always succeeds. Unrecoverable errors should be reported by throwing a runtime exception (which should not be caught, and will instead take down the process). Recoverable errors are reported by invoking an event handler that contains the appropriate error handling logic.
+
+The latter approach can be implemented by registering separate event handlers for each type of error. However, for convenience, it is formalized in Wake's simplified version of the Rx [Observer](wake/src/main/java/org/apache/reef/wake/rx/Observer.java) interface:
+```java
+public interface Observer<T> {
+ void onNext(final T value);
+ void onError(final Exception error);
+ void onCompleted();
+}
+```
+The `Observer` is designed for stateful event handlers that need to be explicitly torn down at exit, or when errors occor. Such event handlers may maintain open network sockets, write to disk, buffer output, and so on. As with `onNext()`, neither `onError()` nor `onCompleted()` throw exceptions. Instead, callers should assume that they are asynchronously invoked.
+
+`EventHandler` and `Observer` implementations should be threadsafe and handle concurrent invocations of `onNext()`. However, it is illegal to call `onCompleted()` or `onError()` in race with any calls to `onNext()`, and the call to `onCompleted()` or `onError()` must be the last call made to the object. Therefore, implementations of `onCompleted()` and `onError()` can assume they have a lock on `this`, and that `this` has not been torn down and is still in a valid state.
+
+We chose these invariants because they are simple and easy to enforce. In most cases, application logic simply limits calls to `onCompleted()` and `onError()` to other implementations of `onError()` and `onCompleted()`, and relies upon Wake (and any intervening application logic) to obey the same protocol.
+
+### Stages
+
+Wake Stages are responsible for resource management. The base [Stage](wake/src/main/java/org/apache/reef/wake/Stage.java) interface is fairly simple:
+
+```java
+public interface Stage extends AutoCloseable { }
+```
+
+The only method it contains is `close()` from auto-closable. This reflects the fact that Wake stages can either contain `EventHandler`s, as [EStage](wake/src/main/java/org/apache/reef/wake/EStage.java) implementations do:
+```java
+public interface EStage<T> extends EventHandler<T>, Stage { }
+```
+or they can contain `Observable`s, as [RxStage](wake/src/main/java/org/apache/reef/wake/rx/RxStage.java) implementations do:
+```java
+public interface RxStage<T> extends Observer<T>, Stage { }
+```
+In both cases, the stage simply exposes the same API as the event handler that it manages. This allows code that produces events to treat downstream stages and raw `EventHandlers` / `Observers` interchangebly. Recall that Wake implements thread sharing by allowing EventHandlers and Observers to directly invoke each other. Since Stages implement the same interface as raw EventHandlers and Observers, this pushes the placement of thread boundaries and other scheduling tradeoffs to the code that is instantiating the application. In turn, this simplifies testing and improves the reusability of code written on top of Wake.
+
+#### `close()` vs. `onCompleted()`
+
+It may seem strange that Wake RxStage exposes two shutdown methods: `close()` and `onCompleted()`. Since `onCompleted()` is part of the Observer API, it may be implemented in an asynchronous fashion. This makes it difficult for applications to cleanly shut down, since, even after `onCompleted()` has returned, resources may still be held by the downstream code.
+
+In contrast, `close()` is synchronous, and is not allowed to return until all queued events have been processed, and any resources held by the Stage implementation have been released. The upshot is that shutdown sequences in Wake work as follows: Once the upstream event sources are done calling `onNext()` (and all calls to `onNext()` have returned), `onCompleted()` or `onError()` is called exactly once per stage. After the `onCompleted()` or `onError()` call to a given stage has returned, `close()` must be called. Once `close()` returns, all resources have been released, and the JVM may safely exit, or the code that is invoking Wake may proceed under the assumption that no resources or memory have been leaked. Note that, depending on the implementation of the downstream Stage, there may be a delay between the return of calls such as `onNext()` or `onCompleted()` and their execution. Therefore, it is possible that the stage will continue to schedule `onNext()` calls after `clos
e()` has been invoked. It is illegal for stages to drop events on shutdown, so the stage will execute the requests in its queue before it releases resources and returns from `close()`.
+
+`Observer` implementations do not expose a `close()` method, and generally do not invoke `close()`. Instead, when `onCompleted()` is invoked, it should arrange for `onCompleted()` to be called on any `Observer` instances that `this` directly invokes, free any resources it is holding, and then return. Since the downstream `onCompleted()` calls are potentially asynchronous, it cannot assume that downstream cleanup completes before it returns.
+
+In a thread pool `Stage`, the final `close()` call will block until there are no more outstanding events queued in the stage. Once `close()` has been called (and returns) on each stage, no events are left in any queues, and no `Observer` or `EventHandler` objects are holding resources or scheduled on any cores, so shutdown is compelete.
+
+Helper libraries
+----------------
+
+Wake includes a number of standard library packages:
+
+ - ```org.apache.reef.wake.time``` allows events to be scheduled in the future, and notifies the application when it starts and when it is being torn down.
+ - ```org.apache.reef.wake.remote``` provides networking primitives, including hooks into netty (a high-performance event-based networking library for Java).
+ - ```org.apache.reef.wake.metrics``` provides implementations of standard latency and throughput instrumentation.
+ - ```org.apache.reef.wake.profiler``` provides a graphical profiler that automatically instruments Tang-based Wake applications.
+
+
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-wake/pom.xml
----------------------------------------------------------------------
diff --git a/lang/java/reef-wake/pom.xml b/lang/java/reef-wake/pom.xml
new file mode 100644
index 0000000..88eeea5
--- /dev/null
+++ b/lang/java/reef-wake/pom.xml
@@ -0,0 +1,38 @@
+<?xml version="1.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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+ <packaging>pom</packaging>
+ <name>REEF Wake Project</name>
+ <artifactId>wake-project</artifactId>
+
+ <parent>
+ <groupId>org.apache.reef</groupId>
+ <artifactId>reef-project</artifactId>
+ <version>0.11.0-incubating-SNAPSHOT</version>
+ </parent>
+
+ <modules>
+ <module>wake</module>
+ </modules>
+
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-wake/wake/pom.xml
----------------------------------------------------------------------
diff --git a/lang/java/reef-wake/wake/pom.xml b/lang/java/reef-wake/wake/pom.xml
new file mode 100644
index 0000000..75fc953
--- /dev/null
+++ b/lang/java/reef-wake/wake/pom.xml
@@ -0,0 +1,123 @@
+<?xml version="1.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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+ <artifactId>wake</artifactId>
+ <name>REEF Wake</name>
+
+ <parent>
+ <groupId>org.apache.reef</groupId>
+ <artifactId>wake-project</artifactId>
+ <version>0.11.0-incubating-SNAPSHOT</version>
+ </parent>
+
+ <build>
+
+ <resources>
+ <resource>
+ <targetPath>META-INF/conf</targetPath>
+ <filtering>false</filtering>
+ <directory>${basedir}/src/main/conf</directory>
+ <includes>
+ <include>*.xml</include>
+ <include>*.properties</include>
+ </includes>
+ <excludes>
+ </excludes>
+ </resource>
+ </resources>
+
+ <plugins>
+ <plugin>
+ <artifactId>maven-antrun-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>generate-sources</id>
+ <phase>generate-sources</phase>
+ <configuration>
+ <tasks>
+ <mkdir dir="target/generated-sources/proto"/>
+ <exec executable="protoc">
+ <arg value="--proto_path=src/main/proto/"/>
+ <arg value="--java_out=target/generated-sources/proto"/>
+ <arg value="src/main/proto/RemoteProtocol.proto"/>
+ </exec>
+ <exec executable="protoc">
+ <arg value="--proto_path=src/test/proto/"/>
+ <arg value="--java_out=target/generated-sources/proto"/>
+ <arg value="src/test/proto/TestProtocol.proto"/>
+ </exec>
+ </tasks>
+ <sourceRoot>target/generated-sources/proto</sourceRoot>
+ </configuration>
+ <goals>
+ <goal>run</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>add-source</id>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>add-source</goal>
+ </goals>
+ <configuration>
+ <sources>
+ <source>target/generated-sources/proto</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+ <dependencies>
+ <dependency>
+ <groupId>cglib</groupId>
+ <artifactId>cglib</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-all</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.google.protobuf</groupId>
+ <artifactId>protobuf-java</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>tang</artifactId>
+ </dependency>
+ </dependencies>
+
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/AbstractEStage.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/AbstractEStage.java b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/AbstractEStage.java
new file mode 100644
index 0000000..30cf265
--- /dev/null
+++ b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/AbstractEStage.java
@@ -0,0 +1,91 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.wake;
+
+import org.apache.reef.wake.metrics.Meter;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * An {@link EStage} that implements metering
+ *
+ * @param <T> type
+ */
+public abstract class AbstractEStage<T> implements EStage<T> {
+
+ protected final AtomicBoolean closed;
+ protected final String name;
+ protected final Meter inMeter;
+
+ /**
+ * outputs share a single meter
+ */
+ protected final Meter outMeter;
+
+ /**
+ * Constructs an abstract estage
+ *
+ * @parm stageName the stage name
+ */
+ public AbstractEStage(String stageName) {
+ this.closed = new AtomicBoolean(false);
+ this.name = stageName;
+ this.inMeter = new Meter(stageName + "_in");
+ this.outMeter = new Meter(stageName + "_out");
+ }
+
+ /**
+ * Gets the input meter of this stage
+ *
+ * @return the input meter
+ */
+ public Meter getInMeter() {
+ return inMeter;
+ }
+
+ /**
+ * Gets the output meter of this stage
+ *
+ * @return the output meter
+ */
+ public Meter getOutMeter() {
+ return outMeter;
+ }
+
+ /**
+ * Updates the input meter.
+ * <p/>
+ * Stages that want to meter their
+ * input must call this each time an event is input.
+ */
+ protected void beforeOnNext() {
+ inMeter.mark(1);
+ }
+
+ /**
+ * Updates the output meter.
+ * <p/>
+ * Stages that want to meter their
+ * output must call this each time an event is output.
+ */
+ protected void afterOnNext() {
+ outMeter.mark(1);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/53ea32cc/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/ComparableIdentifier.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/ComparableIdentifier.java b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/ComparableIdentifier.java
new file mode 100644
index 0000000..a428266
--- /dev/null
+++ b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/ComparableIdentifier.java
@@ -0,0 +1,27 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.wake;
+
+/**
+ * Identifier that can be totally ordered.
+ *
+ * @param <T> type
+ */
+public interface ComparableIdentifier extends Identifier, Comparable<Identifier> {
+}