You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by zj...@apache.org on 2014/11/11 03:43:38 UTC
[2/2] tez git commit: Revert "TEZ-1642. TestAMRecovery sometimes
fail. (zjffdu)"
Revert "TEZ-1642. TestAMRecovery sometimes fail. (zjffdu)"
This reverts commit 088358973205295da006eb2e8dbcb57d87858818.
Project: http://git-wip-us.apache.org/repos/asf/tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/ca4980ad
Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/ca4980ad
Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/ca4980ad
Branch: refs/heads/branch-0.5
Commit: ca4980ad4a1acf9ed736a6406317de406ccc9f44
Parents: 30fb853
Author: Jeff Zhang <zj...@apache.org>
Authored: Tue Nov 11 10:43:16 2014 +0800
Committer: Jeff Zhang <zj...@apache.org>
Committed: Tue Nov 11 10:43:16 2014 +0800
----------------------------------------------------------------------
CHANGES.txt | 1 -
.../org/apache/tez/test/TestAMRecovery.java | 62 ++++++--------------
2 files changed, 19 insertions(+), 44 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tez/blob/ca4980ad/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 8e1da18..4bcaf25 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -12,7 +12,6 @@ ALL CHANGES:
TEZ-1749. Increase test timeout for TestLocalMode.testMultipleClientsWithSession
TEZ-1750. Add a DAGScheduler which schedules tasks only when sources have been scheduled.
TEZ-1761. TestRecoveryParser::testGetLastInProgressDAG fails in similar manner to TEZ-1686.
- TEZ-1642. TestAMRecovery sometimes fail.
Release 0.5.2: 2014-11-07
http://git-wip-us.apache.org/repos/asf/tez/blob/ca4980ad/tez-tests/src/test/java/org/apache/tez/test/TestAMRecovery.java
----------------------------------------------------------------------
diff --git a/tez-tests/src/test/java/org/apache/tez/test/TestAMRecovery.java b/tez-tests/src/test/java/org/apache/tez/test/TestAMRecovery.java
index 867c7ab..42d6f5c 100644
--- a/tez-tests/src/test/java/org/apache/tez/test/TestAMRecovery.java
+++ b/tez-tests/src/test/java/org/apache/tez/test/TestAMRecovery.java
@@ -104,7 +104,7 @@ public class TestAMRecovery {
}
if (miniTezCluster == null) {
miniTezCluster =
- new MiniTezCluster(TestAMRecovery.class.getName(), 2, 1, 1);
+ new MiniTezCluster(TestAMRecovery.class.getName(), 1, 1, 1);
Configuration miniTezconf = new Configuration(conf);
miniTezconf.setInt(YarnConfiguration.RM_AM_MAX_ATTEMPTS, MAX_AM_ATTEMPT);
miniTezconf.set("fs.defaultFS", remoteFs.getUri().toString()); // use HDFS
@@ -127,7 +127,6 @@ public class TestAMRecovery {
try {
LOG.info("Stopping MiniTezCluster");
miniTezCluster.stop();
- miniTezCluster = null;
} catch (Exception e) {
e.printStackTrace();
}
@@ -166,7 +165,6 @@ public class TestAMRecovery {
tezConf.setBoolean(
RecoveryService.TEZ_AM_RECOVERY_HANDLE_REMAINING_EVENT_WHEN_STOPPED,
true);
- tezConf.setInt(TezConfiguration.TEZ_AM_RM_HEARTBEAT_INTERVAL_MS_MAX, 100);
tezSession = TezClient.create("TestDAGRecovery", tezConf);
tezSession.start();
}
@@ -184,16 +182,6 @@ public class TestAMRecovery {
tezSession = null;
}
- private void printHistoryEvents(List<HistoryEvent> historyEvents, int attemptId) {
- LOG.info("RecoveryLogs from attempt:" + attemptId);
- for(HistoryEvent historyEvent : historyEvents) {
- LOG.info("Parsed event from recovery stream"
- + ", eventType=" + historyEvent.getEventType()
- + ", event=" + historyEvent);
- }
- LOG.info("");
- }
-
/**
* Fine-grained recovery task-level, In a vertex (v1), task 0 is done task 1
* is running. History flush happens. AM dies. Once AM is recovered, task 0 is
@@ -207,16 +195,14 @@ public class TestAMRecovery {
createDAG(ControlledInputReadyVertexManager.class,
DataMovementType.BROADCAST, true);
TezCounters counters = runDAGAndVerify(dag, DAGStatus.State.SUCCEEDED);
- List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
- List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
- printHistoryEvents(historyEvents1, 1);
- printHistoryEvents(historyEvents2, 2);
-
assertEquals(5, counters.findCounter(DAGCounter.TOTAL_LAUNCHED_TASKS).getValue());
assertEquals(1, counters.findCounter(DAGCounter.NUM_KILLED_TASKS).getValue());
assertEquals(4, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
assertEquals(2, counters.findCounter(TestCounter.Counter_1).getValue());
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+
// task_0 of v1 is finished in attempt 1, task_1 of v1 is not finished in
// attempt 1
assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
@@ -241,16 +227,14 @@ public class TestAMRecovery {
createDAG(ControlledInputReadyVertexManager.class,
DataMovementType.BROADCAST, false);
TezCounters counters = runDAGAndVerify(dag, DAGStatus.State.SUCCEEDED);
- List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
- List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
- printHistoryEvents(historyEvents1, 1);
- printHistoryEvents(historyEvents2, 2);
-
assertEquals(4, counters.findCounter(DAGCounter.TOTAL_LAUNCHED_TASKS).getValue());
assertEquals(0, counters.findCounter(DAGCounter.NUM_KILLED_TASKS).getValue());
assertEquals(4, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
assertEquals(2, counters.findCounter(TestCounter.Counter_1).getValue());
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+
// task_0 of v1 is finished in attempt 1, task_1 of v1 is not finished in
// attempt 1
assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
@@ -275,16 +259,14 @@ public class TestAMRecovery {
createDAG(ControlledInputReadyVertexManager.class,
DataMovementType.ONE_TO_ONE, true);
TezCounters counters = runDAGAndVerify(dag, DAGStatus.State.SUCCEEDED);
- List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
- List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
- printHistoryEvents(historyEvents1, 1);
- printHistoryEvents(historyEvents2, 2);
-
assertEquals(5, counters.findCounter(DAGCounter.TOTAL_LAUNCHED_TASKS).getValue());
assertEquals(1, counters.findCounter(DAGCounter.NUM_KILLED_TASKS).getValue());
assertEquals(4, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
assertEquals(2, counters.findCounter(TestCounter.Counter_1).getValue());
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+
// task_0 of v1 is finished in attempt 1, task_1 of v1 is not finished in
// attempt 1
assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
@@ -310,16 +292,14 @@ public class TestAMRecovery {
createDAG(ControlledInputReadyVertexManager.class,
DataMovementType.ONE_TO_ONE, false);
TezCounters counters = runDAGAndVerify(dag, DAGStatus.State.SUCCEEDED);
- List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
- List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
- printHistoryEvents(historyEvents1, 1);
- printHistoryEvents(historyEvents2, 2);
-
assertEquals(4, counters.findCounter(DAGCounter.TOTAL_LAUNCHED_TASKS).getValue());
assertEquals(0, counters.findCounter(DAGCounter.NUM_KILLED_TASKS).getValue());
assertEquals(4, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
assertEquals(2, counters.findCounter(TestCounter.Counter_1).getValue());
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+
// task_0 of v1 is finished in attempt 1, task_1 of v1 is not finished in
// attempt 1
assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
@@ -345,16 +325,14 @@ public class TestAMRecovery {
createDAG(ControlledShuffleVertexManager.class,
DataMovementType.SCATTER_GATHER, true);
TezCounters counters = runDAGAndVerify(dag, DAGStatus.State.SUCCEEDED);
- List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
- List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
- printHistoryEvents(historyEvents1, 1);
- printHistoryEvents(historyEvents2, 2);
-
assertEquals(5, counters.findCounter(DAGCounter.TOTAL_LAUNCHED_TASKS).getValue());
assertEquals(1, counters.findCounter(DAGCounter.NUM_KILLED_TASKS).getValue());
assertEquals(4, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
assertEquals(2, counters.findCounter(TestCounter.Counter_1).getValue());
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+
// task_0 of v1 is finished in attempt 1, task_1 of v1 is not finished in
// attempt 1
assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
@@ -380,16 +358,14 @@ public class TestAMRecovery {
createDAG(ControlledShuffleVertexManager.class,
DataMovementType.SCATTER_GATHER, false);
TezCounters counters = runDAGAndVerify(dag, DAGStatus.State.SUCCEEDED);
- List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
- List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
- printHistoryEvents(historyEvents1, 1);
- printHistoryEvents(historyEvents2, 2);
-
assertEquals(4, counters.findCounter(DAGCounter.TOTAL_LAUNCHED_TASKS).getValue());
assertEquals(0, counters.findCounter(DAGCounter.NUM_KILLED_TASKS).getValue());
assertEquals(4, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
assertEquals(2, counters.findCounter(TestCounter.Counter_1).getValue());
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+
// task_0 of v1 is finished in attempt 1, task_1 of v1 is not finished in
// attempt 1
assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());