You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@tez.apache.org by "okumin (via GitHub)" <gi...@apache.org> on 2024/06/12 08:16:05 UTC
[PR] [WIP] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
okumin opened a new pull request, #361:
URL: https://github.com/apache/tez/pull/361
Work in progress. The current solution could break something
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "okumin (via GitHub)" <gi...@apache.org>.
okumin commented on PR #361:
URL: https://github.com/apache/tez/pull/361#issuecomment-2560503721
Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "abstractdog (via GitHub)" <gi...@apache.org>.
abstractdog commented on code in PR #361:
URL: https://github.com/apache/tez/pull/361#discussion_r1894897289
##########
tez-tests/src/test/java/org/apache/tez/test/TestAMRecoveryAggregationBroadcast.java:
##########
@@ -0,0 +1,513 @@
+/**
+ * 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.tez.test;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.apache.tez.client.TezClient;
+import org.apache.tez.client.TezClientUtils;
+import org.apache.tez.common.Preconditions;
+import org.apache.tez.common.TezCommonUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.DAGCounter;
+import org.apache.tez.common.counters.TezCounters;
+import org.apache.tez.dag.api.DAG;
+import org.apache.tez.dag.api.DataSinkDescriptor;
+import org.apache.tez.dag.api.DataSourceDescriptor;
+import org.apache.tez.dag.api.Edge;
+import org.apache.tez.dag.api.EdgeProperty;
+import org.apache.tez.dag.api.ProcessorDescriptor;
+import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.api.TezConstants;
+import org.apache.tez.dag.api.UserPayload;
+import org.apache.tez.dag.api.Vertex;
+import org.apache.tez.dag.api.client.DAGClient;
+import org.apache.tez.dag.api.client.DAGStatus;
+import org.apache.tez.dag.api.client.DAGStatus.State;
+import org.apache.tez.dag.api.client.StatusGetOpts;
+import org.apache.tez.dag.api.oldrecords.TaskAttemptState;
+import org.apache.tez.dag.app.RecoveryParser;
+import org.apache.tez.dag.history.HistoryEvent;
+import org.apache.tez.dag.history.HistoryEventType;
+import org.apache.tez.dag.history.events.TaskAttemptFinishedEvent;
+import org.apache.tez.dag.history.recovery.RecoveryService;
+import org.apache.tez.mapreduce.input.MRInput;
+import org.apache.tez.mapreduce.output.MROutput;
+import org.apache.tez.mapreduce.processor.SimpleMRProcessor;
+import org.apache.tez.runtime.api.ProcessorContext;
+import org.apache.tez.runtime.library.api.KeyValueReader;
+import org.apache.tez.runtime.library.api.KeyValueWriter;
+import org.apache.tez.runtime.library.api.KeyValuesReader;
+import org.apache.tez.runtime.library.conf.OrderedPartitionedKVEdgeConfig;
+import org.apache.tez.runtime.library.conf.UnorderedKVEdgeConfig;
+import org.apache.tez.runtime.library.partitioner.HashPartitioner;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TestAMRecoveryAggregationBroadcast {
+ private static final Logger LOG = LoggerFactory.getLogger(TestAMRecoveryAggregationBroadcast.class);
+ private static final String INPUT1 = "Input";
+ private static final String INPUT2 = "Input";
+ private static final String OUTPUT = "Output";
+ private static final String TABLE_SCAN = "TableScan";
+ private static final String AGGREGATION = "Aggregation";
+ private static final String MAP_JOIN = "MapJoin";
+ private static final String TEST_ROOT_DIR = "target" + Path.SEPARATOR
+ + TestAMRecoveryAggregationBroadcast.class.getName() + "-tmpDir";
+ private static final Path INPUT_FILE = new Path(TEST_ROOT_DIR, "input.csv");
+ private static final Path OUT_PATH = new Path(TEST_ROOT_DIR, "out-groups");
+ private static final String EXPECTED_OUTPUT = "1-5\n1-5\n1-5\n1-5\n1-5\n"
+ + "2-4\n2-4\n2-4\n2-4\n" + "3-3\n3-3\n3-3\n" + "4-2\n4-2\n" + "5-1\n";
+ private static final String TABLE_SCAN_SLEEP = "tez.test.table.scan.sleep";
+ private static final String AGGREGATION_SLEEP = "tez.test.aggregation.sleep";
+ private static final String MAP_JOIN_SLEEP = "tez.test.map.join.sleep";
+
+ private static Configuration dfsConf;
+ private static MiniDFSCluster dfsCluster;
+ private static MiniTezCluster tezCluster;
+ private static FileSystem remoteFs;
+
+ private TezConfiguration tezConf;
+ private TezClient tezSession;
+
+ @BeforeClass
+ public static void setupAll() {
+ try {
+ dfsConf = new Configuration();
+ dfsConf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, TEST_ROOT_DIR);
+ dfsCluster = new MiniDFSCluster.Builder(dfsConf).numDataNodes(3)
+ .format(true).racks(null).build();
+ remoteFs = dfsCluster.getFileSystem();
+ createSampleFile();
+ } catch (IOException io) {
+ throw new RuntimeException("problem starting mini dfs cluster", io);
+ }
+
+ if (tezCluster == null) {
+ tezCluster = new MiniTezCluster(TestAMRecoveryAggregationBroadcast.class.getName(), 1, 1, 1);
+ Configuration conf = new Configuration(dfsConf);
+ conf.set("fs.defaultFS", remoteFs.getUri().toString()); // use HDFS
+ conf.setInt("yarn.nodemanager.delete.debug-delay-sec", 20000);
+ conf.setLong(TezConfiguration.TEZ_AM_SLEEP_TIME_BEFORE_EXIT_MILLIS, 500);
+ tezCluster.init(conf);
+ tezCluster.start();
+ }
+ }
+
+ private static void createSampleFile() throws IOException {
+ FSDataOutputStream out = remoteFs.create(INPUT_FILE);
+ BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(out));
+ // 1, 1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 4, 4, 5
+ for (int i = 1; i <= 5; i++) {
+ for (int j = 0; j <= 5 - i; j++) {
+ writer.write(String.valueOf(i));
+ writer.newLine();
+ }
+ }
+ writer.close();
+ }
+
+ @AfterClass
+ public static void tearDownAll() {
+ if (tezCluster != null) {
+ tezCluster.stop();
+ tezCluster = null;
+ }
+ if (dfsCluster != null) {
+ dfsCluster.shutdown(true);
+ dfsCluster = null;
+ }
+ }
+
+ @Before
+ public void setup() throws Exception {
+ Path remoteStagingDir = remoteFs.makeQualified(new Path(TEST_ROOT_DIR, String
+ .valueOf(new Random().nextInt(100000))));
+ TezClientUtils.ensureStagingDirExists(dfsConf, remoteStagingDir);
+
+ tezConf = new TezConfiguration(tezCluster.getConfig());
+ tezConf.setInt(TezConfiguration.DAG_RECOVERY_MAX_UNFLUSHED_EVENTS, 0);
+ tezConf.set(TezConfiguration.TEZ_AM_LOG_LEVEL, "INFO");
+ tezConf.set(TezConfiguration.TEZ_AM_STAGING_DIR, remoteStagingDir.toString());
+ tezConf.setInt(TezConfiguration.TEZ_AM_RESOURCE_MEMORY_MB, 500);
+ tezConf.set(TezConfiguration.TEZ_AM_LAUNCH_CMD_OPTS, " -Xmx256m");
+ tezConf.setBoolean(TezConfiguration.TEZ_AM_STAGING_SCRATCH_DATA_AUTO_DELETE, false);
+ tezConf.setBoolean(RecoveryService.TEZ_TEST_RECOVERY_DRAIN_EVENTS_WHEN_STOPPED, true);
+
+ tezSession = TezClient.create("TestAMRecoveryAggregationBroadcast", tezConf);
+ tezSession.start();
+ }
+
+ @After
+ public void teardown() throws InterruptedException {
+ if (tezSession != null) {
+ try {
+ LOG.info("Stopping Tez Session");
+ tezSession.stop();
+ } catch (Exception e) {
+ LOG.error("Failed to stop Tez session", e);
+ }
+ }
+ tezSession = null;
+ }
+
+ @Test(timeout = 120000)
+ public void testSucceed() throws Exception {
+ DAG dag = createDAG("Succeed");
+ TezCounters counters = runDAGAndVerify(dag, false);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ // No retry happens
+ assertEquals(Collections.emptyList(), readRecoveryLog(2));
+ }
+
+ @Test(timeout = 120000)
+ public void testTableScanTemporalFailure() throws Exception {
+ tezConf.setBoolean(TABLE_SCAN_SLEEP, true);
+ DAG dag = createDAG("TableScanTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ @Test(timeout = 120000)
+ public void testAggregationTemporalFailure() throws Exception {
+ tezConf.setBoolean(AGGREGATION_SLEEP, true);
+ DAG dag = createDAG("AggregationTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ @Test(timeout = 120000)
+ public void testMapJoinTemporalFailure() throws Exception {
+ tezConf.setBoolean(MAP_JOIN_SLEEP, true);
+ DAG dag = createDAG("MapJoinTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ /**
+ * v1 scans lines and emit (line -> 1), imitating a simple Map vertex.
+ * v2 aggregates the lines and emit (line -> # of duplicated values), imitating an aggregation.
+ * v3 joins the output of v2 with another input. v2 broadcasts its output.
+ * (input1)
+ * \
+ * v1
+ * \
+ * v2 (input2)
+ * \ /
+ * v3
+ */
+ private DAG createDAG(String dagName) throws Exception {
+ UserPayload payload =TezUtils.createUserPayloadFromConf(tezConf);
+ DataSourceDescriptor dataSource = MRInput
+ .createConfigBuilder(new Configuration(tezConf), TextInputFormat.class,
+ INPUT_FILE.toString())
+ .build();
+ // each line -> 1
+ Vertex tableScanVertex = Vertex
+ .create(TABLE_SCAN, ProcessorDescriptor.create(TableScanProcessor.class.getName())
+ .setUserPayload(payload))
+ .addDataSource(INPUT1, dataSource);
+
+ // key -> num keys
+ Vertex aggregationVertex = Vertex
+ .create(AGGREGATION, ProcessorDescriptor
+ .create(AggregationProcessor.class.getName()).setUserPayload(payload), 1);
+
+ DataSinkDescriptor dataSink = MROutput
+ .createConfigBuilder(new Configuration(tezConf), TextOutputFormat.class,
+ OUT_PATH.toString())
+ .build();
+ // Broadcast Hash Join
+ Vertex mapJoinVertex = Vertex
+ .create(MAP_JOIN, ProcessorDescriptor.create(MapJoinProcessor.class.getName())
+ .setUserPayload(payload))
+ .addDataSource(INPUT2, dataSource)
+ .addDataSink(OUTPUT, dataSink);
+
+ EdgeProperty orderedEdge = OrderedPartitionedKVEdgeConfig
+ .newBuilder(Text.class.getName(), IntWritable.class.getName(), HashPartitioner.class.getName())
+ .setFromConfiguration(tezConf)
+ .build()
+ .createDefaultEdgeProperty();
+ EdgeProperty broadcastEdge = UnorderedKVEdgeConfig
+ .newBuilder(Text.class.getName(), IntWritable.class.getName())
+ .setFromConfiguration(tezConf)
+ .build()
+ .createDefaultBroadcastEdgeProperty();
+
+ DAG dag = DAG.create("TestAMRecoveryAggregationBroadcast_" + dagName);
+ dag.addVertex(tableScanVertex)
+ .addVertex(aggregationVertex)
+ .addVertex(mapJoinVertex)
+ .addEdge(Edge.create(tableScanVertex, aggregationVertex, orderedEdge))
+ .addEdge(Edge.create(aggregationVertex, mapJoinVertex, broadcastEdge));
+ return dag;
+ }
+
+ TezCounters runDAGAndVerify(DAG dag, boolean killAM) throws Exception {
+ tezSession.waitTillReady();
+ DAGClient dagClient = tezSession.submitDAG(dag);
+
+ if (killAM) {
+ TimeUnit.SECONDS.sleep(10);
+ YarnClient yarnClient = YarnClient.createYarnClient();
+ yarnClient.init(tezConf);
+ yarnClient.start();
+ ApplicationAttemptId id = ApplicationAttemptId.newInstance(
+ tezSession.getAppMasterApplicationId(), 1);
Review Comment:
nit: line break is not needed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "okumin (via GitHub)" <gi...@apache.org>.
okumin commented on code in PR #361:
URL: https://github.com/apache/tez/pull/361#discussion_r1895638991
##########
tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java:
##########
@@ -2843,7 +2855,7 @@ private void assignVertexManager() throws TezException {
if (recoveryData != null && recoveryData.shouldSkipInit()
Review Comment:
Done. I'm ready to update more if you have a different thought or additional requests.
https://github.com/apache/tez/pull/361/commits/c8c18fcb49de1d0cadd368ed8db73750f9d154f7
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "okumin (via GitHub)" <gi...@apache.org>.
okumin commented on code in PR #361:
URL: https://github.com/apache/tez/pull/361#discussion_r1894906774
##########
tez-tests/src/test/java/org/apache/tez/test/TestAMRecoveryAggregationBroadcast.java:
##########
@@ -0,0 +1,513 @@
+/**
+ * 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.tez.test;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.apache.tez.client.TezClient;
+import org.apache.tez.client.TezClientUtils;
+import org.apache.tez.common.Preconditions;
+import org.apache.tez.common.TezCommonUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.DAGCounter;
+import org.apache.tez.common.counters.TezCounters;
+import org.apache.tez.dag.api.DAG;
+import org.apache.tez.dag.api.DataSinkDescriptor;
+import org.apache.tez.dag.api.DataSourceDescriptor;
+import org.apache.tez.dag.api.Edge;
+import org.apache.tez.dag.api.EdgeProperty;
+import org.apache.tez.dag.api.ProcessorDescriptor;
+import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.api.TezConstants;
+import org.apache.tez.dag.api.UserPayload;
+import org.apache.tez.dag.api.Vertex;
+import org.apache.tez.dag.api.client.DAGClient;
+import org.apache.tez.dag.api.client.DAGStatus;
+import org.apache.tez.dag.api.client.DAGStatus.State;
+import org.apache.tez.dag.api.client.StatusGetOpts;
+import org.apache.tez.dag.api.oldrecords.TaskAttemptState;
+import org.apache.tez.dag.app.RecoveryParser;
+import org.apache.tez.dag.history.HistoryEvent;
+import org.apache.tez.dag.history.HistoryEventType;
+import org.apache.tez.dag.history.events.TaskAttemptFinishedEvent;
+import org.apache.tez.dag.history.recovery.RecoveryService;
+import org.apache.tez.mapreduce.input.MRInput;
+import org.apache.tez.mapreduce.output.MROutput;
+import org.apache.tez.mapreduce.processor.SimpleMRProcessor;
+import org.apache.tez.runtime.api.ProcessorContext;
+import org.apache.tez.runtime.library.api.KeyValueReader;
+import org.apache.tez.runtime.library.api.KeyValueWriter;
+import org.apache.tez.runtime.library.api.KeyValuesReader;
+import org.apache.tez.runtime.library.conf.OrderedPartitionedKVEdgeConfig;
+import org.apache.tez.runtime.library.conf.UnorderedKVEdgeConfig;
+import org.apache.tez.runtime.library.partitioner.HashPartitioner;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TestAMRecoveryAggregationBroadcast {
+ private static final Logger LOG = LoggerFactory.getLogger(TestAMRecoveryAggregationBroadcast.class);
+ private static final String INPUT1 = "Input";
+ private static final String INPUT2 = "Input";
+ private static final String OUTPUT = "Output";
+ private static final String TABLE_SCAN = "TableScan";
+ private static final String AGGREGATION = "Aggregation";
+ private static final String MAP_JOIN = "MapJoin";
+ private static final String TEST_ROOT_DIR = "target" + Path.SEPARATOR
+ + TestAMRecoveryAggregationBroadcast.class.getName() + "-tmpDir";
+ private static final Path INPUT_FILE = new Path(TEST_ROOT_DIR, "input.csv");
+ private static final Path OUT_PATH = new Path(TEST_ROOT_DIR, "out-groups");
+ private static final String EXPECTED_OUTPUT = "1-5\n1-5\n1-5\n1-5\n1-5\n"
+ + "2-4\n2-4\n2-4\n2-4\n" + "3-3\n3-3\n3-3\n" + "4-2\n4-2\n" + "5-1\n";
+ private static final String TABLE_SCAN_SLEEP = "tez.test.table.scan.sleep";
+ private static final String AGGREGATION_SLEEP = "tez.test.aggregation.sleep";
+ private static final String MAP_JOIN_SLEEP = "tez.test.map.join.sleep";
+
+ private static Configuration dfsConf;
+ private static MiniDFSCluster dfsCluster;
+ private static MiniTezCluster tezCluster;
+ private static FileSystem remoteFs;
+
+ private TezConfiguration tezConf;
+ private TezClient tezSession;
+
+ @BeforeClass
+ public static void setupAll() {
+ try {
+ dfsConf = new Configuration();
+ dfsConf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, TEST_ROOT_DIR);
+ dfsCluster = new MiniDFSCluster.Builder(dfsConf).numDataNodes(3)
+ .format(true).racks(null).build();
+ remoteFs = dfsCluster.getFileSystem();
+ createSampleFile();
+ } catch (IOException io) {
+ throw new RuntimeException("problem starting mini dfs cluster", io);
+ }
+
+ if (tezCluster == null) {
+ tezCluster = new MiniTezCluster(TestAMRecoveryAggregationBroadcast.class.getName(), 1, 1, 1);
+ Configuration conf = new Configuration(dfsConf);
+ conf.set("fs.defaultFS", remoteFs.getUri().toString()); // use HDFS
+ conf.setInt("yarn.nodemanager.delete.debug-delay-sec", 20000);
+ conf.setLong(TezConfiguration.TEZ_AM_SLEEP_TIME_BEFORE_EXIT_MILLIS, 500);
+ tezCluster.init(conf);
+ tezCluster.start();
+ }
+ }
+
+ private static void createSampleFile() throws IOException {
+ FSDataOutputStream out = remoteFs.create(INPUT_FILE);
+ BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(out));
+ // 1, 1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 4, 4, 5
+ for (int i = 1; i <= 5; i++) {
+ for (int j = 0; j <= 5 - i; j++) {
+ writer.write(String.valueOf(i));
+ writer.newLine();
+ }
+ }
+ writer.close();
+ }
+
+ @AfterClass
+ public static void tearDownAll() {
+ if (tezCluster != null) {
+ tezCluster.stop();
+ tezCluster = null;
+ }
+ if (dfsCluster != null) {
+ dfsCluster.shutdown(true);
+ dfsCluster = null;
+ }
+ }
+
+ @Before
+ public void setup() throws Exception {
+ Path remoteStagingDir = remoteFs.makeQualified(new Path(TEST_ROOT_DIR, String
+ .valueOf(new Random().nextInt(100000))));
+ TezClientUtils.ensureStagingDirExists(dfsConf, remoteStagingDir);
+
+ tezConf = new TezConfiguration(tezCluster.getConfig());
+ tezConf.setInt(TezConfiguration.DAG_RECOVERY_MAX_UNFLUSHED_EVENTS, 0);
+ tezConf.set(TezConfiguration.TEZ_AM_LOG_LEVEL, "INFO");
+ tezConf.set(TezConfiguration.TEZ_AM_STAGING_DIR, remoteStagingDir.toString());
+ tezConf.setInt(TezConfiguration.TEZ_AM_RESOURCE_MEMORY_MB, 500);
+ tezConf.set(TezConfiguration.TEZ_AM_LAUNCH_CMD_OPTS, " -Xmx256m");
+ tezConf.setBoolean(TezConfiguration.TEZ_AM_STAGING_SCRATCH_DATA_AUTO_DELETE, false);
+ tezConf.setBoolean(RecoveryService.TEZ_TEST_RECOVERY_DRAIN_EVENTS_WHEN_STOPPED, true);
+
+ tezSession = TezClient.create("TestAMRecoveryAggregationBroadcast", tezConf);
+ tezSession.start();
+ }
+
+ @After
+ public void teardown() throws InterruptedException {
+ if (tezSession != null) {
+ try {
+ LOG.info("Stopping Tez Session");
+ tezSession.stop();
+ } catch (Exception e) {
+ LOG.error("Failed to stop Tez session", e);
+ }
+ }
+ tezSession = null;
+ }
+
+ @Test(timeout = 120000)
+ public void testSucceed() throws Exception {
+ DAG dag = createDAG("Succeed");
+ TezCounters counters = runDAGAndVerify(dag, false);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ // No retry happens
+ assertEquals(Collections.emptyList(), readRecoveryLog(2));
+ }
+
+ @Test(timeout = 120000)
+ public void testTableScanTemporalFailure() throws Exception {
+ tezConf.setBoolean(TABLE_SCAN_SLEEP, true);
+ DAG dag = createDAG("TableScanTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ @Test(timeout = 120000)
+ public void testAggregationTemporalFailure() throws Exception {
+ tezConf.setBoolean(AGGREGATION_SLEEP, true);
+ DAG dag = createDAG("AggregationTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ @Test(timeout = 120000)
+ public void testMapJoinTemporalFailure() throws Exception {
+ tezConf.setBoolean(MAP_JOIN_SLEEP, true);
+ DAG dag = createDAG("MapJoinTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ /**
+ * v1 scans lines and emit (line -> 1), imitating a simple Map vertex.
+ * v2 aggregates the lines and emit (line -> # of duplicated values), imitating an aggregation.
+ * v3 joins the output of v2 with another input. v2 broadcasts its output.
+ * (input1)
+ * \
+ * v1
+ * \
+ * v2 (input2)
+ * \ /
+ * v3
+ */
+ private DAG createDAG(String dagName) throws Exception {
+ UserPayload payload =TezUtils.createUserPayloadFromConf(tezConf);
+ DataSourceDescriptor dataSource = MRInput
+ .createConfigBuilder(new Configuration(tezConf), TextInputFormat.class,
+ INPUT_FILE.toString())
+ .build();
+ // each line -> 1
+ Vertex tableScanVertex = Vertex
+ .create(TABLE_SCAN, ProcessorDescriptor.create(TableScanProcessor.class.getName())
+ .setUserPayload(payload))
+ .addDataSource(INPUT1, dataSource);
+
+ // key -> num keys
+ Vertex aggregationVertex = Vertex
+ .create(AGGREGATION, ProcessorDescriptor
+ .create(AggregationProcessor.class.getName()).setUserPayload(payload), 1);
+
+ DataSinkDescriptor dataSink = MROutput
+ .createConfigBuilder(new Configuration(tezConf), TextOutputFormat.class,
+ OUT_PATH.toString())
+ .build();
+ // Broadcast Hash Join
+ Vertex mapJoinVertex = Vertex
+ .create(MAP_JOIN, ProcessorDescriptor.create(MapJoinProcessor.class.getName())
+ .setUserPayload(payload))
+ .addDataSource(INPUT2, dataSource)
+ .addDataSink(OUTPUT, dataSink);
+
+ EdgeProperty orderedEdge = OrderedPartitionedKVEdgeConfig
+ .newBuilder(Text.class.getName(), IntWritable.class.getName(), HashPartitioner.class.getName())
+ .setFromConfiguration(tezConf)
+ .build()
+ .createDefaultEdgeProperty();
+ EdgeProperty broadcastEdge = UnorderedKVEdgeConfig
+ .newBuilder(Text.class.getName(), IntWritable.class.getName())
+ .setFromConfiguration(tezConf)
+ .build()
+ .createDefaultBroadcastEdgeProperty();
+
+ DAG dag = DAG.create("TestAMRecoveryAggregationBroadcast_" + dagName);
+ dag.addVertex(tableScanVertex)
+ .addVertex(aggregationVertex)
+ .addVertex(mapJoinVertex)
+ .addEdge(Edge.create(tableScanVertex, aggregationVertex, orderedEdge))
+ .addEdge(Edge.create(aggregationVertex, mapJoinVertex, broadcastEdge));
+ return dag;
+ }
+
+ TezCounters runDAGAndVerify(DAG dag, boolean killAM) throws Exception {
+ tezSession.waitTillReady();
+ DAGClient dagClient = tezSession.submitDAG(dag);
+
+ if (killAM) {
+ TimeUnit.SECONDS.sleep(10);
+ YarnClient yarnClient = YarnClient.createYarnClient();
+ yarnClient.init(tezConf);
+ yarnClient.start();
+ ApplicationAttemptId id = ApplicationAttemptId.newInstance(
+ tezSession.getAppMasterApplicationId(), 1);
+ yarnClient.failApplicationAttempt(id);
+ yarnClient.close();
+ }
+ DAGStatus dagStatus = dagClient
+ .waitForCompletionWithStatusUpdates(EnumSet.of(StatusGetOpts.GET_COUNTERS));
+ LOG.info("Diagnosis: " + dagStatus.getDiagnostics());
+ Assert.assertEquals(State.SUCCEEDED, dagStatus.getState());
+
+ FSDataInputStream in = remoteFs.open(new Path(OUT_PATH, "part-v002-o000-r-00000"));
+ ByteBuffer buf = ByteBuffer.allocate(100);
+ in.read(buf);
+ buf.flip();
+ Assert.assertEquals(EXPECTED_OUTPUT, StandardCharsets.UTF_8.decode(buf).toString());
+ return dagStatus.getDAGCounters();
+ }
+
+ private List<HistoryEvent> readRecoveryLog(int attemptNum) throws IOException {
+ ApplicationId appId = tezSession.getAppMasterApplicationId();
+ Path tezSystemStagingDir = TezCommonUtils.getTezSystemStagingPath(tezConf, appId.toString());
+ Path recoveryDataDir = TezCommonUtils.getRecoveryPath(tezSystemStagingDir, tezConf);
+ FileSystem fs = tezSystemStagingDir.getFileSystem(tezConf);
+ List<HistoryEvent> historyEvents = new ArrayList<>();
+ Path currentAttemptRecoveryDataDir = TezCommonUtils.getAttemptRecoveryPath(recoveryDataDir, attemptNum);
+ Path recoveryFilePath =
+ new Path(currentAttemptRecoveryDataDir, appId.toString().replace("application", "dag")
+ + "_1" + TezConstants.DAG_RECOVERY_RECOVER_FILE_SUFFIX);
+ if (fs.exists(recoveryFilePath)) {
+ LOG.info("Read recovery file:" + recoveryFilePath);
+ historyEvents.addAll(RecoveryParser.parseDAGRecoveryFile(fs.open(recoveryFilePath)));
+ }
+ printHistoryEvents(historyEvents, attemptNum);
+ return historyEvents;
+ }
+
+ 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("");
Review Comment:
I ported it from [TestAMRecovery](https://github.com/apache/tez/blob/1084699ddb1baf6e4a6df97d988ac696d33f41e8/tez-tests/src/test/java/org/apache/tez/test/TestAMRecovery.java#L506). I guess we wanted a new line to visibly split huge output.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "abstractdog (via GitHub)" <gi...@apache.org>.
abstractdog commented on code in PR #361:
URL: https://github.com/apache/tez/pull/361#discussion_r1894896489
##########
tez-tests/src/test/java/org/apache/tez/test/TestAMRecoveryAggregationBroadcast.java:
##########
@@ -0,0 +1,513 @@
+/**
+ * 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.tez.test;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.apache.tez.client.TezClient;
+import org.apache.tez.client.TezClientUtils;
+import org.apache.tez.common.Preconditions;
+import org.apache.tez.common.TezCommonUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.DAGCounter;
+import org.apache.tez.common.counters.TezCounters;
+import org.apache.tez.dag.api.DAG;
+import org.apache.tez.dag.api.DataSinkDescriptor;
+import org.apache.tez.dag.api.DataSourceDescriptor;
+import org.apache.tez.dag.api.Edge;
+import org.apache.tez.dag.api.EdgeProperty;
+import org.apache.tez.dag.api.ProcessorDescriptor;
+import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.api.TezConstants;
+import org.apache.tez.dag.api.UserPayload;
+import org.apache.tez.dag.api.Vertex;
+import org.apache.tez.dag.api.client.DAGClient;
+import org.apache.tez.dag.api.client.DAGStatus;
+import org.apache.tez.dag.api.client.DAGStatus.State;
+import org.apache.tez.dag.api.client.StatusGetOpts;
+import org.apache.tez.dag.api.oldrecords.TaskAttemptState;
+import org.apache.tez.dag.app.RecoveryParser;
+import org.apache.tez.dag.history.HistoryEvent;
+import org.apache.tez.dag.history.HistoryEventType;
+import org.apache.tez.dag.history.events.TaskAttemptFinishedEvent;
+import org.apache.tez.dag.history.recovery.RecoveryService;
+import org.apache.tez.mapreduce.input.MRInput;
+import org.apache.tez.mapreduce.output.MROutput;
+import org.apache.tez.mapreduce.processor.SimpleMRProcessor;
+import org.apache.tez.runtime.api.ProcessorContext;
+import org.apache.tez.runtime.library.api.KeyValueReader;
+import org.apache.tez.runtime.library.api.KeyValueWriter;
+import org.apache.tez.runtime.library.api.KeyValuesReader;
+import org.apache.tez.runtime.library.conf.OrderedPartitionedKVEdgeConfig;
+import org.apache.tez.runtime.library.conf.UnorderedKVEdgeConfig;
+import org.apache.tez.runtime.library.partitioner.HashPartitioner;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TestAMRecoveryAggregationBroadcast {
+ private static final Logger LOG = LoggerFactory.getLogger(TestAMRecoveryAggregationBroadcast.class);
+ private static final String INPUT1 = "Input";
+ private static final String INPUT2 = "Input";
+ private static final String OUTPUT = "Output";
+ private static final String TABLE_SCAN = "TableScan";
+ private static final String AGGREGATION = "Aggregation";
+ private static final String MAP_JOIN = "MapJoin";
+ private static final String TEST_ROOT_DIR = "target" + Path.SEPARATOR
+ + TestAMRecoveryAggregationBroadcast.class.getName() + "-tmpDir";
+ private static final Path INPUT_FILE = new Path(TEST_ROOT_DIR, "input.csv");
+ private static final Path OUT_PATH = new Path(TEST_ROOT_DIR, "out-groups");
+ private static final String EXPECTED_OUTPUT = "1-5\n1-5\n1-5\n1-5\n1-5\n"
+ + "2-4\n2-4\n2-4\n2-4\n" + "3-3\n3-3\n3-3\n" + "4-2\n4-2\n" + "5-1\n";
+ private static final String TABLE_SCAN_SLEEP = "tez.test.table.scan.sleep";
+ private static final String AGGREGATION_SLEEP = "tez.test.aggregation.sleep";
+ private static final String MAP_JOIN_SLEEP = "tez.test.map.join.sleep";
+
+ private static Configuration dfsConf;
+ private static MiniDFSCluster dfsCluster;
+ private static MiniTezCluster tezCluster;
+ private static FileSystem remoteFs;
+
+ private TezConfiguration tezConf;
+ private TezClient tezSession;
+
+ @BeforeClass
+ public static void setupAll() {
+ try {
+ dfsConf = new Configuration();
+ dfsConf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, TEST_ROOT_DIR);
+ dfsCluster = new MiniDFSCluster.Builder(dfsConf).numDataNodes(3)
+ .format(true).racks(null).build();
+ remoteFs = dfsCluster.getFileSystem();
+ createSampleFile();
+ } catch (IOException io) {
+ throw new RuntimeException("problem starting mini dfs cluster", io);
+ }
+
+ if (tezCluster == null) {
+ tezCluster = new MiniTezCluster(TestAMRecoveryAggregationBroadcast.class.getName(), 1, 1, 1);
+ Configuration conf = new Configuration(dfsConf);
+ conf.set("fs.defaultFS", remoteFs.getUri().toString()); // use HDFS
+ conf.setInt("yarn.nodemanager.delete.debug-delay-sec", 20000);
+ conf.setLong(TezConfiguration.TEZ_AM_SLEEP_TIME_BEFORE_EXIT_MILLIS, 500);
+ tezCluster.init(conf);
+ tezCluster.start();
+ }
+ }
+
+ private static void createSampleFile() throws IOException {
+ FSDataOutputStream out = remoteFs.create(INPUT_FILE);
+ BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(out));
+ // 1, 1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 4, 4, 5
+ for (int i = 1; i <= 5; i++) {
+ for (int j = 0; j <= 5 - i; j++) {
+ writer.write(String.valueOf(i));
+ writer.newLine();
+ }
+ }
+ writer.close();
+ }
+
+ @AfterClass
+ public static void tearDownAll() {
+ if (tezCluster != null) {
+ tezCluster.stop();
+ tezCluster = null;
+ }
+ if (dfsCluster != null) {
+ dfsCluster.shutdown(true);
+ dfsCluster = null;
+ }
+ }
+
+ @Before
+ public void setup() throws Exception {
+ Path remoteStagingDir = remoteFs.makeQualified(new Path(TEST_ROOT_DIR, String
+ .valueOf(new Random().nextInt(100000))));
+ TezClientUtils.ensureStagingDirExists(dfsConf, remoteStagingDir);
+
+ tezConf = new TezConfiguration(tezCluster.getConfig());
+ tezConf.setInt(TezConfiguration.DAG_RECOVERY_MAX_UNFLUSHED_EVENTS, 0);
+ tezConf.set(TezConfiguration.TEZ_AM_LOG_LEVEL, "INFO");
+ tezConf.set(TezConfiguration.TEZ_AM_STAGING_DIR, remoteStagingDir.toString());
+ tezConf.setInt(TezConfiguration.TEZ_AM_RESOURCE_MEMORY_MB, 500);
+ tezConf.set(TezConfiguration.TEZ_AM_LAUNCH_CMD_OPTS, " -Xmx256m");
+ tezConf.setBoolean(TezConfiguration.TEZ_AM_STAGING_SCRATCH_DATA_AUTO_DELETE, false);
+ tezConf.setBoolean(RecoveryService.TEZ_TEST_RECOVERY_DRAIN_EVENTS_WHEN_STOPPED, true);
+
+ tezSession = TezClient.create("TestAMRecoveryAggregationBroadcast", tezConf);
+ tezSession.start();
+ }
+
+ @After
+ public void teardown() throws InterruptedException {
+ if (tezSession != null) {
+ try {
+ LOG.info("Stopping Tez Session");
+ tezSession.stop();
+ } catch (Exception e) {
+ LOG.error("Failed to stop Tez session", e);
+ }
+ }
+ tezSession = null;
+ }
+
+ @Test(timeout = 120000)
+ public void testSucceed() throws Exception {
+ DAG dag = createDAG("Succeed");
+ TezCounters counters = runDAGAndVerify(dag, false);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ // No retry happens
+ assertEquals(Collections.emptyList(), readRecoveryLog(2));
+ }
+
+ @Test(timeout = 120000)
+ public void testTableScanTemporalFailure() throws Exception {
+ tezConf.setBoolean(TABLE_SCAN_SLEEP, true);
+ DAG dag = createDAG("TableScanTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ @Test(timeout = 120000)
+ public void testAggregationTemporalFailure() throws Exception {
+ tezConf.setBoolean(AGGREGATION_SLEEP, true);
+ DAG dag = createDAG("AggregationTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ @Test(timeout = 120000)
+ public void testMapJoinTemporalFailure() throws Exception {
+ tezConf.setBoolean(MAP_JOIN_SLEEP, true);
+ DAG dag = createDAG("MapJoinTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ /**
+ * v1 scans lines and emit (line -> 1), imitating a simple Map vertex.
+ * v2 aggregates the lines and emit (line -> # of duplicated values), imitating an aggregation.
+ * v3 joins the output of v2 with another input. v2 broadcasts its output.
+ * (input1)
+ * \
+ * v1
+ * \
+ * v2 (input2)
+ * \ /
+ * v3
+ */
+ private DAG createDAG(String dagName) throws Exception {
+ UserPayload payload =TezUtils.createUserPayloadFromConf(tezConf);
Review Comment:
nit: space after =
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "okumin (via GitHub)" <gi...@apache.org>.
okumin commented on code in PR #361:
URL: https://github.com/apache/tez/pull/361#discussion_r1894651483
##########
tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java:
##########
@@ -2817,6 +2817,18 @@ private VertexState setupVertex() {
return VertexState.INITED;
}
+ private boolean canSkipInitializingParents() {
+ // Both cases use RootInputVertexManager. RootInputVertexManager can start tasks even though
+ // any parents are not fully initialized.
+ if (vertexPlan.hasVertexManagerPlugin()) {
+ final VertexManagerPluginDescriptor pluginDesc = DagTypeConverters
+ .convertVertexManagerPluginDescriptorFromDAGPlan(vertexPlan.getVertexManagerPlugin());
+ return pluginDesc.getClassName().equals(RootInputVertexManager.class.getName());
+ } else {
+ return inputsWithInitializers != null;
Review Comment:
Based on the comment above, I guess it would try to care about this pattern. Honestly, I lost a part of my memory about this PR.
https://github.com/apache/tez/blob/ca1511998e183c8e10da79bef20463d40a39726a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java#L2898-L2917
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "okumin (via GitHub)" <gi...@apache.org>.
okumin commented on code in PR #361:
URL: https://github.com/apache/tez/pull/361#discussion_r1894652069
##########
tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java:
##########
@@ -2843,7 +2855,7 @@ private void assignVertexManager() throws TezException {
if (recoveryData != null && recoveryData.shouldSkipInit()
Review Comment:
I can do it once we agree on the course of TEZ-4569. Do you prefer to move the whole condition in this `if` branch?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "abstractdog (via GitHub)" <gi...@apache.org>.
abstractdog commented on code in PR #361:
URL: https://github.com/apache/tez/pull/361#discussion_r1894902637
##########
tez-tests/src/test/java/org/apache/tez/test/TestAMRecoveryAggregationBroadcast.java:
##########
@@ -0,0 +1,513 @@
+/**
+ * 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.tez.test;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.apache.tez.client.TezClient;
+import org.apache.tez.client.TezClientUtils;
+import org.apache.tez.common.Preconditions;
+import org.apache.tez.common.TezCommonUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.DAGCounter;
+import org.apache.tez.common.counters.TezCounters;
+import org.apache.tez.dag.api.DAG;
+import org.apache.tez.dag.api.DataSinkDescriptor;
+import org.apache.tez.dag.api.DataSourceDescriptor;
+import org.apache.tez.dag.api.Edge;
+import org.apache.tez.dag.api.EdgeProperty;
+import org.apache.tez.dag.api.ProcessorDescriptor;
+import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.api.TezConstants;
+import org.apache.tez.dag.api.UserPayload;
+import org.apache.tez.dag.api.Vertex;
+import org.apache.tez.dag.api.client.DAGClient;
+import org.apache.tez.dag.api.client.DAGStatus;
+import org.apache.tez.dag.api.client.DAGStatus.State;
+import org.apache.tez.dag.api.client.StatusGetOpts;
+import org.apache.tez.dag.api.oldrecords.TaskAttemptState;
+import org.apache.tez.dag.app.RecoveryParser;
+import org.apache.tez.dag.history.HistoryEvent;
+import org.apache.tez.dag.history.HistoryEventType;
+import org.apache.tez.dag.history.events.TaskAttemptFinishedEvent;
+import org.apache.tez.dag.history.recovery.RecoveryService;
+import org.apache.tez.mapreduce.input.MRInput;
+import org.apache.tez.mapreduce.output.MROutput;
+import org.apache.tez.mapreduce.processor.SimpleMRProcessor;
+import org.apache.tez.runtime.api.ProcessorContext;
+import org.apache.tez.runtime.library.api.KeyValueReader;
+import org.apache.tez.runtime.library.api.KeyValueWriter;
+import org.apache.tez.runtime.library.api.KeyValuesReader;
+import org.apache.tez.runtime.library.conf.OrderedPartitionedKVEdgeConfig;
+import org.apache.tez.runtime.library.conf.UnorderedKVEdgeConfig;
+import org.apache.tez.runtime.library.partitioner.HashPartitioner;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TestAMRecoveryAggregationBroadcast {
Review Comment:
checked, unit test hangs without the fix, passes with the fix, nice one, looks good overall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "abstractdog (via GitHub)" <gi...@apache.org>.
abstractdog commented on code in PR #361:
URL: https://github.com/apache/tez/pull/361#discussion_r1894630065
##########
tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java:
##########
@@ -2843,7 +2855,7 @@ private void assignVertexManager() throws TezException {
if (recoveryData != null && recoveryData.shouldSkipInit()
Review Comment:
this whole condition reached a complexity where we should move it into a separate method like:
```
canSkipVertexInitOnRecovery(recoveryData)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "okumin (via GitHub)" <gi...@apache.org>.
okumin commented on code in PR #361:
URL: https://github.com/apache/tez/pull/361#discussion_r1894651483
##########
tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java:
##########
@@ -2817,6 +2817,18 @@ private VertexState setupVertex() {
return VertexState.INITED;
}
+ private boolean canSkipInitializingParents() {
+ // Both cases use RootInputVertexManager. RootInputVertexManager can start tasks even though
+ // any parents are not fully initialized.
+ if (vertexPlan.hasVertexManagerPlugin()) {
+ final VertexManagerPluginDescriptor pluginDesc = DagTypeConverters
+ .convertVertexManagerPluginDescriptorFromDAGPlan(vertexPlan.getVertexManagerPlugin());
+ return pluginDesc.getClassName().equals(RootInputVertexManager.class.getName());
+ } else {
+ return inputsWithInitializers != null;
Review Comment:
Based on the comment above, I guess it would try to care about this pattern. Honestly, I don't have the perfect memory of this PR.
https://github.com/apache/tez/blob/ca1511998e183c8e10da79bef20463d40a39726a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java#L2912-L2917
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "okumin (via GitHub)" <gi...@apache.org>.
okumin commented on PR #361:
URL: https://github.com/apache/tez/pull/361#issuecomment-2559785639
@abstractdog Sorry, my hand-made refactoring included a mistake. I copy-pasted the method names used in the original condition. I appreciate it if you could double-check it.
https://github.com/apache/tez/pull/361/commits/bdece70288d20463b131d852511128103f093cdf
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "abstractdog (via GitHub)" <gi...@apache.org>.
abstractdog commented on code in PR #361:
URL: https://github.com/apache/tez/pull/361#discussion_r1894897462
##########
tez-tests/src/test/java/org/apache/tez/test/TestAMRecoveryAggregationBroadcast.java:
##########
@@ -0,0 +1,513 @@
+/**
+ * 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.tez.test;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.apache.tez.client.TezClient;
+import org.apache.tez.client.TezClientUtils;
+import org.apache.tez.common.Preconditions;
+import org.apache.tez.common.TezCommonUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.DAGCounter;
+import org.apache.tez.common.counters.TezCounters;
+import org.apache.tez.dag.api.DAG;
+import org.apache.tez.dag.api.DataSinkDescriptor;
+import org.apache.tez.dag.api.DataSourceDescriptor;
+import org.apache.tez.dag.api.Edge;
+import org.apache.tez.dag.api.EdgeProperty;
+import org.apache.tez.dag.api.ProcessorDescriptor;
+import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.api.TezConstants;
+import org.apache.tez.dag.api.UserPayload;
+import org.apache.tez.dag.api.Vertex;
+import org.apache.tez.dag.api.client.DAGClient;
+import org.apache.tez.dag.api.client.DAGStatus;
+import org.apache.tez.dag.api.client.DAGStatus.State;
+import org.apache.tez.dag.api.client.StatusGetOpts;
+import org.apache.tez.dag.api.oldrecords.TaskAttemptState;
+import org.apache.tez.dag.app.RecoveryParser;
+import org.apache.tez.dag.history.HistoryEvent;
+import org.apache.tez.dag.history.HistoryEventType;
+import org.apache.tez.dag.history.events.TaskAttemptFinishedEvent;
+import org.apache.tez.dag.history.recovery.RecoveryService;
+import org.apache.tez.mapreduce.input.MRInput;
+import org.apache.tez.mapreduce.output.MROutput;
+import org.apache.tez.mapreduce.processor.SimpleMRProcessor;
+import org.apache.tez.runtime.api.ProcessorContext;
+import org.apache.tez.runtime.library.api.KeyValueReader;
+import org.apache.tez.runtime.library.api.KeyValueWriter;
+import org.apache.tez.runtime.library.api.KeyValuesReader;
+import org.apache.tez.runtime.library.conf.OrderedPartitionedKVEdgeConfig;
+import org.apache.tez.runtime.library.conf.UnorderedKVEdgeConfig;
+import org.apache.tez.runtime.library.partitioner.HashPartitioner;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TestAMRecoveryAggregationBroadcast {
+ private static final Logger LOG = LoggerFactory.getLogger(TestAMRecoveryAggregationBroadcast.class);
+ private static final String INPUT1 = "Input";
+ private static final String INPUT2 = "Input";
+ private static final String OUTPUT = "Output";
+ private static final String TABLE_SCAN = "TableScan";
+ private static final String AGGREGATION = "Aggregation";
+ private static final String MAP_JOIN = "MapJoin";
+ private static final String TEST_ROOT_DIR = "target" + Path.SEPARATOR
+ + TestAMRecoveryAggregationBroadcast.class.getName() + "-tmpDir";
+ private static final Path INPUT_FILE = new Path(TEST_ROOT_DIR, "input.csv");
+ private static final Path OUT_PATH = new Path(TEST_ROOT_DIR, "out-groups");
+ private static final String EXPECTED_OUTPUT = "1-5\n1-5\n1-5\n1-5\n1-5\n"
+ + "2-4\n2-4\n2-4\n2-4\n" + "3-3\n3-3\n3-3\n" + "4-2\n4-2\n" + "5-1\n";
+ private static final String TABLE_SCAN_SLEEP = "tez.test.table.scan.sleep";
+ private static final String AGGREGATION_SLEEP = "tez.test.aggregation.sleep";
+ private static final String MAP_JOIN_SLEEP = "tez.test.map.join.sleep";
+
+ private static Configuration dfsConf;
+ private static MiniDFSCluster dfsCluster;
+ private static MiniTezCluster tezCluster;
+ private static FileSystem remoteFs;
+
+ private TezConfiguration tezConf;
+ private TezClient tezSession;
+
+ @BeforeClass
+ public static void setupAll() {
+ try {
+ dfsConf = new Configuration();
+ dfsConf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, TEST_ROOT_DIR);
+ dfsCluster = new MiniDFSCluster.Builder(dfsConf).numDataNodes(3)
+ .format(true).racks(null).build();
+ remoteFs = dfsCluster.getFileSystem();
+ createSampleFile();
+ } catch (IOException io) {
+ throw new RuntimeException("problem starting mini dfs cluster", io);
+ }
+
+ if (tezCluster == null) {
+ tezCluster = new MiniTezCluster(TestAMRecoveryAggregationBroadcast.class.getName(), 1, 1, 1);
+ Configuration conf = new Configuration(dfsConf);
+ conf.set("fs.defaultFS", remoteFs.getUri().toString()); // use HDFS
+ conf.setInt("yarn.nodemanager.delete.debug-delay-sec", 20000);
+ conf.setLong(TezConfiguration.TEZ_AM_SLEEP_TIME_BEFORE_EXIT_MILLIS, 500);
+ tezCluster.init(conf);
+ tezCluster.start();
+ }
+ }
+
+ private static void createSampleFile() throws IOException {
+ FSDataOutputStream out = remoteFs.create(INPUT_FILE);
+ BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(out));
+ // 1, 1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 4, 4, 5
+ for (int i = 1; i <= 5; i++) {
+ for (int j = 0; j <= 5 - i; j++) {
+ writer.write(String.valueOf(i));
+ writer.newLine();
+ }
+ }
+ writer.close();
+ }
+
+ @AfterClass
+ public static void tearDownAll() {
+ if (tezCluster != null) {
+ tezCluster.stop();
+ tezCluster = null;
+ }
+ if (dfsCluster != null) {
+ dfsCluster.shutdown(true);
+ dfsCluster = null;
+ }
+ }
+
+ @Before
+ public void setup() throws Exception {
+ Path remoteStagingDir = remoteFs.makeQualified(new Path(TEST_ROOT_DIR, String
+ .valueOf(new Random().nextInt(100000))));
+ TezClientUtils.ensureStagingDirExists(dfsConf, remoteStagingDir);
+
+ tezConf = new TezConfiguration(tezCluster.getConfig());
+ tezConf.setInt(TezConfiguration.DAG_RECOVERY_MAX_UNFLUSHED_EVENTS, 0);
+ tezConf.set(TezConfiguration.TEZ_AM_LOG_LEVEL, "INFO");
+ tezConf.set(TezConfiguration.TEZ_AM_STAGING_DIR, remoteStagingDir.toString());
+ tezConf.setInt(TezConfiguration.TEZ_AM_RESOURCE_MEMORY_MB, 500);
+ tezConf.set(TezConfiguration.TEZ_AM_LAUNCH_CMD_OPTS, " -Xmx256m");
+ tezConf.setBoolean(TezConfiguration.TEZ_AM_STAGING_SCRATCH_DATA_AUTO_DELETE, false);
+ tezConf.setBoolean(RecoveryService.TEZ_TEST_RECOVERY_DRAIN_EVENTS_WHEN_STOPPED, true);
+
+ tezSession = TezClient.create("TestAMRecoveryAggregationBroadcast", tezConf);
+ tezSession.start();
+ }
+
+ @After
+ public void teardown() throws InterruptedException {
+ if (tezSession != null) {
+ try {
+ LOG.info("Stopping Tez Session");
+ tezSession.stop();
+ } catch (Exception e) {
+ LOG.error("Failed to stop Tez session", e);
+ }
+ }
+ tezSession = null;
+ }
+
+ @Test(timeout = 120000)
+ public void testSucceed() throws Exception {
+ DAG dag = createDAG("Succeed");
+ TezCounters counters = runDAGAndVerify(dag, false);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ // No retry happens
+ assertEquals(Collections.emptyList(), readRecoveryLog(2));
+ }
+
+ @Test(timeout = 120000)
+ public void testTableScanTemporalFailure() throws Exception {
+ tezConf.setBoolean(TABLE_SCAN_SLEEP, true);
+ DAG dag = createDAG("TableScanTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ @Test(timeout = 120000)
+ public void testAggregationTemporalFailure() throws Exception {
+ tezConf.setBoolean(AGGREGATION_SLEEP, true);
+ DAG dag = createDAG("AggregationTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ @Test(timeout = 120000)
+ public void testMapJoinTemporalFailure() throws Exception {
+ tezConf.setBoolean(MAP_JOIN_SLEEP, true);
+ DAG dag = createDAG("MapJoinTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ /**
+ * v1 scans lines and emit (line -> 1), imitating a simple Map vertex.
+ * v2 aggregates the lines and emit (line -> # of duplicated values), imitating an aggregation.
+ * v3 joins the output of v2 with another input. v2 broadcasts its output.
+ * (input1)
+ * \
+ * v1
+ * \
+ * v2 (input2)
+ * \ /
+ * v3
+ */
+ private DAG createDAG(String dagName) throws Exception {
+ UserPayload payload =TezUtils.createUserPayloadFromConf(tezConf);
+ DataSourceDescriptor dataSource = MRInput
+ .createConfigBuilder(new Configuration(tezConf), TextInputFormat.class,
+ INPUT_FILE.toString())
+ .build();
+ // each line -> 1
+ Vertex tableScanVertex = Vertex
+ .create(TABLE_SCAN, ProcessorDescriptor.create(TableScanProcessor.class.getName())
+ .setUserPayload(payload))
+ .addDataSource(INPUT1, dataSource);
+
+ // key -> num keys
+ Vertex aggregationVertex = Vertex
+ .create(AGGREGATION, ProcessorDescriptor
+ .create(AggregationProcessor.class.getName()).setUserPayload(payload), 1);
+
+ DataSinkDescriptor dataSink = MROutput
+ .createConfigBuilder(new Configuration(tezConf), TextOutputFormat.class,
+ OUT_PATH.toString())
+ .build();
+ // Broadcast Hash Join
+ Vertex mapJoinVertex = Vertex
+ .create(MAP_JOIN, ProcessorDescriptor.create(MapJoinProcessor.class.getName())
+ .setUserPayload(payload))
+ .addDataSource(INPUT2, dataSource)
+ .addDataSink(OUTPUT, dataSink);
+
+ EdgeProperty orderedEdge = OrderedPartitionedKVEdgeConfig
+ .newBuilder(Text.class.getName(), IntWritable.class.getName(), HashPartitioner.class.getName())
+ .setFromConfiguration(tezConf)
+ .build()
+ .createDefaultEdgeProperty();
+ EdgeProperty broadcastEdge = UnorderedKVEdgeConfig
+ .newBuilder(Text.class.getName(), IntWritable.class.getName())
+ .setFromConfiguration(tezConf)
+ .build()
+ .createDefaultBroadcastEdgeProperty();
+
+ DAG dag = DAG.create("TestAMRecoveryAggregationBroadcast_" + dagName);
+ dag.addVertex(tableScanVertex)
+ .addVertex(aggregationVertex)
+ .addVertex(mapJoinVertex)
+ .addEdge(Edge.create(tableScanVertex, aggregationVertex, orderedEdge))
+ .addEdge(Edge.create(aggregationVertex, mapJoinVertex, broadcastEdge));
+ return dag;
+ }
+
+ TezCounters runDAGAndVerify(DAG dag, boolean killAM) throws Exception {
+ tezSession.waitTillReady();
+ DAGClient dagClient = tezSession.submitDAG(dag);
+
+ if (killAM) {
+ TimeUnit.SECONDS.sleep(10);
+ YarnClient yarnClient = YarnClient.createYarnClient();
+ yarnClient.init(tezConf);
+ yarnClient.start();
+ ApplicationAttemptId id = ApplicationAttemptId.newInstance(
+ tezSession.getAppMasterApplicationId(), 1);
+ yarnClient.failApplicationAttempt(id);
+ yarnClient.close();
+ }
+ DAGStatus dagStatus = dagClient
+ .waitForCompletionWithStatusUpdates(EnumSet.of(StatusGetOpts.GET_COUNTERS));
Review Comment:
nit: line break is not needed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "abstractdog (via GitHub)" <gi...@apache.org>.
abstractdog commented on code in PR #361:
URL: https://github.com/apache/tez/pull/361#discussion_r1894896068
##########
tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java:
##########
@@ -2817,6 +2817,18 @@ private VertexState setupVertex() {
return VertexState.INITED;
}
+ private boolean canSkipInitializingParents() {
+ // Both cases use RootInputVertexManager. RootInputVertexManager can start tasks even though
+ // any parents are not fully initialized.
+ if (vertexPlan.hasVertexManagerPlugin()) {
+ final VertexManagerPluginDescriptor pluginDesc = DagTypeConverters
+ .convertVertexManagerPluginDescriptorFromDAGPlan(vertexPlan.getVertexManagerPlugin());
+ return pluginDesc.getClassName().equals(RootInputVertexManager.class.getName());
+ } else {
+ return inputsWithInitializers != null;
Review Comment:
okay, that makes sense, are you fine with further refactoring and reusing the conditions where possible? something like below:
```
private boolean isRootInputVertexManagerUsed() {
return isRootInputVertexManagerPluginUsed() || hasInputInitializers();
}
private boolean isRootInputVertexManagerPluginUsed(){
return vertexPlan.hasVertexManagerPlugin() && DagTypeConverters
.convertVertexManagerPluginDescriptorFromDAGPlan(vertexPlan.getVertexManagerPlugin()).getClassName().equals(RootInputVertexManager.class.getName());
}
private boolean hasInputInitializers(){
return inputsWithInitializers != null;
}
```
hasInputInitializers might be used in 2 places for cross reference
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "tez-yetus (via GitHub)" <gi...@apache.org>.
tez-yetus commented on PR #361:
URL: https://github.com/apache/tez/pull/361#issuecomment-2163421264
:confetti_ball: **+1 overall**
| Vote | Subsystem | Runtime | Comment |
|:----:|----------:|--------:|:--------|
| +0 :ok: | reexec | 0m 32s | Docker mode activated. |
||| _ Prechecks _ |
| +1 :green_heart: | dupname | 0m 0s | No case conflicting files found. |
| +1 :green_heart: | @author | 0m 0s | The patch does not contain any @author tags. |
| +1 :green_heart: | test4tests | 0m 0s | The patch appears to include 1 new or modified test files. |
||| _ master Compile Tests _ |
| +0 :ok: | mvndep | 7m 2s | Maven dependency ordering for branch |
| +1 :green_heart: | mvninstall | 16m 6s | master passed |
| +1 :green_heart: | compile | 1m 21s | master passed with JDK Ubuntu-11.0.23+9-post-Ubuntu-1ubuntu122.04.1 |
| +1 :green_heart: | compile | 1m 12s | master passed with JDK Private Build-1.8.0_412-8u412-ga-1~22.04.1-b08 |
| +1 :green_heart: | checkstyle | 1m 33s | master passed |
| +1 :green_heart: | javadoc | 1m 4s | master passed with JDK Ubuntu-11.0.23+9-post-Ubuntu-1ubuntu122.04.1 |
| +1 :green_heart: | javadoc | 0m 44s | master passed with JDK Private Build-1.8.0_412-8u412-ga-1~22.04.1-b08 |
| +0 :ok: | spotbugs | 0m 48s | Used deprecated FindBugs config; considering switching to SpotBugs. |
| +1 :green_heart: | findbugs | 2m 49s | master passed |
||| _ Patch Compile Tests _ |
| +0 :ok: | mvndep | 0m 10s | Maven dependency ordering for patch |
| +1 :green_heart: | mvninstall | 0m 48s | the patch passed |
| +1 :green_heart: | compile | 0m 52s | the patch passed with JDK Ubuntu-11.0.23+9-post-Ubuntu-1ubuntu122.04.1 |
| +1 :green_heart: | javac | 0m 52s | the patch passed |
| +1 :green_heart: | compile | 0m 48s | the patch passed with JDK Private Build-1.8.0_412-8u412-ga-1~22.04.1-b08 |
| +1 :green_heart: | javac | 0m 48s | the patch passed |
| -0 :warning: | checkstyle | 0m 12s | tez-tests: The patch generated 1 new + 0 unchanged - 0 fixed = 1 total (was 0) |
| +1 :green_heart: | whitespace | 0m 0s | The patch has no whitespace issues. |
| +1 :green_heart: | javadoc | 0m 20s | the patch passed with JDK Ubuntu-11.0.23+9-post-Ubuntu-1ubuntu122.04.1 |
| +1 :green_heart: | javadoc | 0m 20s | the patch passed with JDK Private Build-1.8.0_412-8u412-ga-1~22.04.1-b08 |
| +1 :green_heart: | findbugs | 1m 59s | the patch passed |
||| _ Other Tests _ |
| +1 :green_heart: | unit | 5m 11s | tez-dag in the patch passed. |
| +1 :green_heart: | unit | 41m 25s | tez-tests in the patch passed. |
| +1 :green_heart: | asflicense | 0m 31s | The patch does not generate ASF License warnings. |
| | | 86m 47s | |
| Subsystem | Report/Notes |
|----------:|:-------------|
| Docker | ClientAPI=1.45 ServerAPI=1.45 base: https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-361/2/artifact/out/Dockerfile |
| GITHUB PR | https://github.com/apache/tez/pull/361 |
| Optional Tests | dupname asflicense javac javadoc unit spotbugs findbugs checkstyle compile |
| uname | Linux 1a7a0c602b4f 5.15.0-106-generic #116-Ubuntu SMP Wed Apr 17 09:17:56 UTC 2024 x86_64 x86_64 x86_64 GNU/Linux |
| Build tool | maven |
| Personality | personality/tez.sh |
| git revision | master / e08d0279c |
| Default Java | Private Build-1.8.0_412-8u412-ga-1~22.04.1-b08 |
| Multi-JDK versions | /usr/lib/jvm/java-11-openjdk-amd64:Ubuntu-11.0.23+9-post-Ubuntu-1ubuntu122.04.1 /usr/lib/jvm/java-8-openjdk-amd64:Private Build-1.8.0_412-8u412-ga-1~22.04.1-b08 |
| checkstyle | https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-361/2/artifact/out/diff-checkstyle-tez-tests.txt |
| Test Results | https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-361/2/testReport/ |
| Max. process+thread count | 1162 (vs. ulimit of 5500) |
| modules | C: tez-dag tez-tests U: . |
| Console output | https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-361/2/console |
| versions | git=2.34.1 maven=3.6.3 findbugs=3.0.1 |
| Powered by | Apache Yetus 0.12.0 https://yetus.apache.org |
This message was automatically generated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "okumin (via GitHub)" <gi...@apache.org>.
okumin commented on code in PR #361:
URL: https://github.com/apache/tez/pull/361#discussion_r1894906774
##########
tez-tests/src/test/java/org/apache/tez/test/TestAMRecoveryAggregationBroadcast.java:
##########
@@ -0,0 +1,513 @@
+/**
+ * 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.tez.test;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.apache.tez.client.TezClient;
+import org.apache.tez.client.TezClientUtils;
+import org.apache.tez.common.Preconditions;
+import org.apache.tez.common.TezCommonUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.DAGCounter;
+import org.apache.tez.common.counters.TezCounters;
+import org.apache.tez.dag.api.DAG;
+import org.apache.tez.dag.api.DataSinkDescriptor;
+import org.apache.tez.dag.api.DataSourceDescriptor;
+import org.apache.tez.dag.api.Edge;
+import org.apache.tez.dag.api.EdgeProperty;
+import org.apache.tez.dag.api.ProcessorDescriptor;
+import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.api.TezConstants;
+import org.apache.tez.dag.api.UserPayload;
+import org.apache.tez.dag.api.Vertex;
+import org.apache.tez.dag.api.client.DAGClient;
+import org.apache.tez.dag.api.client.DAGStatus;
+import org.apache.tez.dag.api.client.DAGStatus.State;
+import org.apache.tez.dag.api.client.StatusGetOpts;
+import org.apache.tez.dag.api.oldrecords.TaskAttemptState;
+import org.apache.tez.dag.app.RecoveryParser;
+import org.apache.tez.dag.history.HistoryEvent;
+import org.apache.tez.dag.history.HistoryEventType;
+import org.apache.tez.dag.history.events.TaskAttemptFinishedEvent;
+import org.apache.tez.dag.history.recovery.RecoveryService;
+import org.apache.tez.mapreduce.input.MRInput;
+import org.apache.tez.mapreduce.output.MROutput;
+import org.apache.tez.mapreduce.processor.SimpleMRProcessor;
+import org.apache.tez.runtime.api.ProcessorContext;
+import org.apache.tez.runtime.library.api.KeyValueReader;
+import org.apache.tez.runtime.library.api.KeyValueWriter;
+import org.apache.tez.runtime.library.api.KeyValuesReader;
+import org.apache.tez.runtime.library.conf.OrderedPartitionedKVEdgeConfig;
+import org.apache.tez.runtime.library.conf.UnorderedKVEdgeConfig;
+import org.apache.tez.runtime.library.partitioner.HashPartitioner;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TestAMRecoveryAggregationBroadcast {
+ private static final Logger LOG = LoggerFactory.getLogger(TestAMRecoveryAggregationBroadcast.class);
+ private static final String INPUT1 = "Input";
+ private static final String INPUT2 = "Input";
+ private static final String OUTPUT = "Output";
+ private static final String TABLE_SCAN = "TableScan";
+ private static final String AGGREGATION = "Aggregation";
+ private static final String MAP_JOIN = "MapJoin";
+ private static final String TEST_ROOT_DIR = "target" + Path.SEPARATOR
+ + TestAMRecoveryAggregationBroadcast.class.getName() + "-tmpDir";
+ private static final Path INPUT_FILE = new Path(TEST_ROOT_DIR, "input.csv");
+ private static final Path OUT_PATH = new Path(TEST_ROOT_DIR, "out-groups");
+ private static final String EXPECTED_OUTPUT = "1-5\n1-5\n1-5\n1-5\n1-5\n"
+ + "2-4\n2-4\n2-4\n2-4\n" + "3-3\n3-3\n3-3\n" + "4-2\n4-2\n" + "5-1\n";
+ private static final String TABLE_SCAN_SLEEP = "tez.test.table.scan.sleep";
+ private static final String AGGREGATION_SLEEP = "tez.test.aggregation.sleep";
+ private static final String MAP_JOIN_SLEEP = "tez.test.map.join.sleep";
+
+ private static Configuration dfsConf;
+ private static MiniDFSCluster dfsCluster;
+ private static MiniTezCluster tezCluster;
+ private static FileSystem remoteFs;
+
+ private TezConfiguration tezConf;
+ private TezClient tezSession;
+
+ @BeforeClass
+ public static void setupAll() {
+ try {
+ dfsConf = new Configuration();
+ dfsConf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, TEST_ROOT_DIR);
+ dfsCluster = new MiniDFSCluster.Builder(dfsConf).numDataNodes(3)
+ .format(true).racks(null).build();
+ remoteFs = dfsCluster.getFileSystem();
+ createSampleFile();
+ } catch (IOException io) {
+ throw new RuntimeException("problem starting mini dfs cluster", io);
+ }
+
+ if (tezCluster == null) {
+ tezCluster = new MiniTezCluster(TestAMRecoveryAggregationBroadcast.class.getName(), 1, 1, 1);
+ Configuration conf = new Configuration(dfsConf);
+ conf.set("fs.defaultFS", remoteFs.getUri().toString()); // use HDFS
+ conf.setInt("yarn.nodemanager.delete.debug-delay-sec", 20000);
+ conf.setLong(TezConfiguration.TEZ_AM_SLEEP_TIME_BEFORE_EXIT_MILLIS, 500);
+ tezCluster.init(conf);
+ tezCluster.start();
+ }
+ }
+
+ private static void createSampleFile() throws IOException {
+ FSDataOutputStream out = remoteFs.create(INPUT_FILE);
+ BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(out));
+ // 1, 1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 4, 4, 5
+ for (int i = 1; i <= 5; i++) {
+ for (int j = 0; j <= 5 - i; j++) {
+ writer.write(String.valueOf(i));
+ writer.newLine();
+ }
+ }
+ writer.close();
+ }
+
+ @AfterClass
+ public static void tearDownAll() {
+ if (tezCluster != null) {
+ tezCluster.stop();
+ tezCluster = null;
+ }
+ if (dfsCluster != null) {
+ dfsCluster.shutdown(true);
+ dfsCluster = null;
+ }
+ }
+
+ @Before
+ public void setup() throws Exception {
+ Path remoteStagingDir = remoteFs.makeQualified(new Path(TEST_ROOT_DIR, String
+ .valueOf(new Random().nextInt(100000))));
+ TezClientUtils.ensureStagingDirExists(dfsConf, remoteStagingDir);
+
+ tezConf = new TezConfiguration(tezCluster.getConfig());
+ tezConf.setInt(TezConfiguration.DAG_RECOVERY_MAX_UNFLUSHED_EVENTS, 0);
+ tezConf.set(TezConfiguration.TEZ_AM_LOG_LEVEL, "INFO");
+ tezConf.set(TezConfiguration.TEZ_AM_STAGING_DIR, remoteStagingDir.toString());
+ tezConf.setInt(TezConfiguration.TEZ_AM_RESOURCE_MEMORY_MB, 500);
+ tezConf.set(TezConfiguration.TEZ_AM_LAUNCH_CMD_OPTS, " -Xmx256m");
+ tezConf.setBoolean(TezConfiguration.TEZ_AM_STAGING_SCRATCH_DATA_AUTO_DELETE, false);
+ tezConf.setBoolean(RecoveryService.TEZ_TEST_RECOVERY_DRAIN_EVENTS_WHEN_STOPPED, true);
+
+ tezSession = TezClient.create("TestAMRecoveryAggregationBroadcast", tezConf);
+ tezSession.start();
+ }
+
+ @After
+ public void teardown() throws InterruptedException {
+ if (tezSession != null) {
+ try {
+ LOG.info("Stopping Tez Session");
+ tezSession.stop();
+ } catch (Exception e) {
+ LOG.error("Failed to stop Tez session", e);
+ }
+ }
+ tezSession = null;
+ }
+
+ @Test(timeout = 120000)
+ public void testSucceed() throws Exception {
+ DAG dag = createDAG("Succeed");
+ TezCounters counters = runDAGAndVerify(dag, false);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ // No retry happens
+ assertEquals(Collections.emptyList(), readRecoveryLog(2));
+ }
+
+ @Test(timeout = 120000)
+ public void testTableScanTemporalFailure() throws Exception {
+ tezConf.setBoolean(TABLE_SCAN_SLEEP, true);
+ DAG dag = createDAG("TableScanTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ @Test(timeout = 120000)
+ public void testAggregationTemporalFailure() throws Exception {
+ tezConf.setBoolean(AGGREGATION_SLEEP, true);
+ DAG dag = createDAG("AggregationTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ @Test(timeout = 120000)
+ public void testMapJoinTemporalFailure() throws Exception {
+ tezConf.setBoolean(MAP_JOIN_SLEEP, true);
+ DAG dag = createDAG("MapJoinTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ /**
+ * v1 scans lines and emit (line -> 1), imitating a simple Map vertex.
+ * v2 aggregates the lines and emit (line -> # of duplicated values), imitating an aggregation.
+ * v3 joins the output of v2 with another input. v2 broadcasts its output.
+ * (input1)
+ * \
+ * v1
+ * \
+ * v2 (input2)
+ * \ /
+ * v3
+ */
+ private DAG createDAG(String dagName) throws Exception {
+ UserPayload payload =TezUtils.createUserPayloadFromConf(tezConf);
+ DataSourceDescriptor dataSource = MRInput
+ .createConfigBuilder(new Configuration(tezConf), TextInputFormat.class,
+ INPUT_FILE.toString())
+ .build();
+ // each line -> 1
+ Vertex tableScanVertex = Vertex
+ .create(TABLE_SCAN, ProcessorDescriptor.create(TableScanProcessor.class.getName())
+ .setUserPayload(payload))
+ .addDataSource(INPUT1, dataSource);
+
+ // key -> num keys
+ Vertex aggregationVertex = Vertex
+ .create(AGGREGATION, ProcessorDescriptor
+ .create(AggregationProcessor.class.getName()).setUserPayload(payload), 1);
+
+ DataSinkDescriptor dataSink = MROutput
+ .createConfigBuilder(new Configuration(tezConf), TextOutputFormat.class,
+ OUT_PATH.toString())
+ .build();
+ // Broadcast Hash Join
+ Vertex mapJoinVertex = Vertex
+ .create(MAP_JOIN, ProcessorDescriptor.create(MapJoinProcessor.class.getName())
+ .setUserPayload(payload))
+ .addDataSource(INPUT2, dataSource)
+ .addDataSink(OUTPUT, dataSink);
+
+ EdgeProperty orderedEdge = OrderedPartitionedKVEdgeConfig
+ .newBuilder(Text.class.getName(), IntWritable.class.getName(), HashPartitioner.class.getName())
+ .setFromConfiguration(tezConf)
+ .build()
+ .createDefaultEdgeProperty();
+ EdgeProperty broadcastEdge = UnorderedKVEdgeConfig
+ .newBuilder(Text.class.getName(), IntWritable.class.getName())
+ .setFromConfiguration(tezConf)
+ .build()
+ .createDefaultBroadcastEdgeProperty();
+
+ DAG dag = DAG.create("TestAMRecoveryAggregationBroadcast_" + dagName);
+ dag.addVertex(tableScanVertex)
+ .addVertex(aggregationVertex)
+ .addVertex(mapJoinVertex)
+ .addEdge(Edge.create(tableScanVertex, aggregationVertex, orderedEdge))
+ .addEdge(Edge.create(aggregationVertex, mapJoinVertex, broadcastEdge));
+ return dag;
+ }
+
+ TezCounters runDAGAndVerify(DAG dag, boolean killAM) throws Exception {
+ tezSession.waitTillReady();
+ DAGClient dagClient = tezSession.submitDAG(dag);
+
+ if (killAM) {
+ TimeUnit.SECONDS.sleep(10);
+ YarnClient yarnClient = YarnClient.createYarnClient();
+ yarnClient.init(tezConf);
+ yarnClient.start();
+ ApplicationAttemptId id = ApplicationAttemptId.newInstance(
+ tezSession.getAppMasterApplicationId(), 1);
+ yarnClient.failApplicationAttempt(id);
+ yarnClient.close();
+ }
+ DAGStatus dagStatus = dagClient
+ .waitForCompletionWithStatusUpdates(EnumSet.of(StatusGetOpts.GET_COUNTERS));
+ LOG.info("Diagnosis: " + dagStatus.getDiagnostics());
+ Assert.assertEquals(State.SUCCEEDED, dagStatus.getState());
+
+ FSDataInputStream in = remoteFs.open(new Path(OUT_PATH, "part-v002-o000-r-00000"));
+ ByteBuffer buf = ByteBuffer.allocate(100);
+ in.read(buf);
+ buf.flip();
+ Assert.assertEquals(EXPECTED_OUTPUT, StandardCharsets.UTF_8.decode(buf).toString());
+ return dagStatus.getDAGCounters();
+ }
+
+ private List<HistoryEvent> readRecoveryLog(int attemptNum) throws IOException {
+ ApplicationId appId = tezSession.getAppMasterApplicationId();
+ Path tezSystemStagingDir = TezCommonUtils.getTezSystemStagingPath(tezConf, appId.toString());
+ Path recoveryDataDir = TezCommonUtils.getRecoveryPath(tezSystemStagingDir, tezConf);
+ FileSystem fs = tezSystemStagingDir.getFileSystem(tezConf);
+ List<HistoryEvent> historyEvents = new ArrayList<>();
+ Path currentAttemptRecoveryDataDir = TezCommonUtils.getAttemptRecoveryPath(recoveryDataDir, attemptNum);
+ Path recoveryFilePath =
+ new Path(currentAttemptRecoveryDataDir, appId.toString().replace("application", "dag")
+ + "_1" + TezConstants.DAG_RECOVERY_RECOVER_FILE_SUFFIX);
+ if (fs.exists(recoveryFilePath)) {
+ LOG.info("Read recovery file:" + recoveryFilePath);
+ historyEvents.addAll(RecoveryParser.parseDAGRecoveryFile(fs.open(recoveryFilePath)));
+ }
+ printHistoryEvents(historyEvents, attemptNum);
+ return historyEvents;
+ }
+
+ 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("");
Review Comment:
I ported it from [TestAMRecovery](https://github.com/apache/tez/blob/1084699ddb1baf6e4a6df97d988ac696d33f41e8/tez-tests/src/test/java/org/apache/tez/test/TestAMRecovery.java#L506). I guess we wanted a new line to visibly separate huge output.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "okumin (via GitHub)" <gi...@apache.org>.
okumin commented on code in PR #361:
URL: https://github.com/apache/tez/pull/361#discussion_r1895635895
##########
tez-tests/src/test/java/org/apache/tez/test/TestAMRecoveryAggregationBroadcast.java:
##########
@@ -0,0 +1,513 @@
+/**
+ * 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.tez.test;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.apache.tez.client.TezClient;
+import org.apache.tez.client.TezClientUtils;
+import org.apache.tez.common.Preconditions;
+import org.apache.tez.common.TezCommonUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.DAGCounter;
+import org.apache.tez.common.counters.TezCounters;
+import org.apache.tez.dag.api.DAG;
+import org.apache.tez.dag.api.DataSinkDescriptor;
+import org.apache.tez.dag.api.DataSourceDescriptor;
+import org.apache.tez.dag.api.Edge;
+import org.apache.tez.dag.api.EdgeProperty;
+import org.apache.tez.dag.api.ProcessorDescriptor;
+import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.api.TezConstants;
+import org.apache.tez.dag.api.UserPayload;
+import org.apache.tez.dag.api.Vertex;
+import org.apache.tez.dag.api.client.DAGClient;
+import org.apache.tez.dag.api.client.DAGStatus;
+import org.apache.tez.dag.api.client.DAGStatus.State;
+import org.apache.tez.dag.api.client.StatusGetOpts;
+import org.apache.tez.dag.api.oldrecords.TaskAttemptState;
+import org.apache.tez.dag.app.RecoveryParser;
+import org.apache.tez.dag.history.HistoryEvent;
+import org.apache.tez.dag.history.HistoryEventType;
+import org.apache.tez.dag.history.events.TaskAttemptFinishedEvent;
+import org.apache.tez.dag.history.recovery.RecoveryService;
+import org.apache.tez.mapreduce.input.MRInput;
+import org.apache.tez.mapreduce.output.MROutput;
+import org.apache.tez.mapreduce.processor.SimpleMRProcessor;
+import org.apache.tez.runtime.api.ProcessorContext;
+import org.apache.tez.runtime.library.api.KeyValueReader;
+import org.apache.tez.runtime.library.api.KeyValueWriter;
+import org.apache.tez.runtime.library.api.KeyValuesReader;
+import org.apache.tez.runtime.library.conf.OrderedPartitionedKVEdgeConfig;
+import org.apache.tez.runtime.library.conf.UnorderedKVEdgeConfig;
+import org.apache.tez.runtime.library.partitioner.HashPartitioner;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TestAMRecoveryAggregationBroadcast {
+ private static final Logger LOG = LoggerFactory.getLogger(TestAMRecoveryAggregationBroadcast.class);
+ private static final String INPUT1 = "Input";
+ private static final String INPUT2 = "Input";
+ private static final String OUTPUT = "Output";
+ private static final String TABLE_SCAN = "TableScan";
+ private static final String AGGREGATION = "Aggregation";
+ private static final String MAP_JOIN = "MapJoin";
+ private static final String TEST_ROOT_DIR = "target" + Path.SEPARATOR
+ + TestAMRecoveryAggregationBroadcast.class.getName() + "-tmpDir";
+ private static final Path INPUT_FILE = new Path(TEST_ROOT_DIR, "input.csv");
+ private static final Path OUT_PATH = new Path(TEST_ROOT_DIR, "out-groups");
+ private static final String EXPECTED_OUTPUT = "1-5\n1-5\n1-5\n1-5\n1-5\n"
+ + "2-4\n2-4\n2-4\n2-4\n" + "3-3\n3-3\n3-3\n" + "4-2\n4-2\n" + "5-1\n";
+ private static final String TABLE_SCAN_SLEEP = "tez.test.table.scan.sleep";
+ private static final String AGGREGATION_SLEEP = "tez.test.aggregation.sleep";
+ private static final String MAP_JOIN_SLEEP = "tez.test.map.join.sleep";
+
+ private static Configuration dfsConf;
+ private static MiniDFSCluster dfsCluster;
+ private static MiniTezCluster tezCluster;
+ private static FileSystem remoteFs;
+
+ private TezConfiguration tezConf;
+ private TezClient tezSession;
+
+ @BeforeClass
+ public static void setupAll() {
+ try {
+ dfsConf = new Configuration();
+ dfsConf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, TEST_ROOT_DIR);
+ dfsCluster = new MiniDFSCluster.Builder(dfsConf).numDataNodes(3)
+ .format(true).racks(null).build();
+ remoteFs = dfsCluster.getFileSystem();
+ createSampleFile();
+ } catch (IOException io) {
+ throw new RuntimeException("problem starting mini dfs cluster", io);
+ }
+
+ if (tezCluster == null) {
+ tezCluster = new MiniTezCluster(TestAMRecoveryAggregationBroadcast.class.getName(), 1, 1, 1);
+ Configuration conf = new Configuration(dfsConf);
+ conf.set("fs.defaultFS", remoteFs.getUri().toString()); // use HDFS
+ conf.setInt("yarn.nodemanager.delete.debug-delay-sec", 20000);
+ conf.setLong(TezConfiguration.TEZ_AM_SLEEP_TIME_BEFORE_EXIT_MILLIS, 500);
+ tezCluster.init(conf);
+ tezCluster.start();
+ }
+ }
+
+ private static void createSampleFile() throws IOException {
+ FSDataOutputStream out = remoteFs.create(INPUT_FILE);
+ BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(out));
+ // 1, 1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 4, 4, 5
+ for (int i = 1; i <= 5; i++) {
+ for (int j = 0; j <= 5 - i; j++) {
+ writer.write(String.valueOf(i));
+ writer.newLine();
+ }
+ }
+ writer.close();
+ }
+
+ @AfterClass
+ public static void tearDownAll() {
+ if (tezCluster != null) {
+ tezCluster.stop();
+ tezCluster = null;
+ }
+ if (dfsCluster != null) {
+ dfsCluster.shutdown(true);
+ dfsCluster = null;
+ }
+ }
+
+ @Before
+ public void setup() throws Exception {
+ Path remoteStagingDir = remoteFs.makeQualified(new Path(TEST_ROOT_DIR, String
+ .valueOf(new Random().nextInt(100000))));
+ TezClientUtils.ensureStagingDirExists(dfsConf, remoteStagingDir);
+
+ tezConf = new TezConfiguration(tezCluster.getConfig());
+ tezConf.setInt(TezConfiguration.DAG_RECOVERY_MAX_UNFLUSHED_EVENTS, 0);
+ tezConf.set(TezConfiguration.TEZ_AM_LOG_LEVEL, "INFO");
+ tezConf.set(TezConfiguration.TEZ_AM_STAGING_DIR, remoteStagingDir.toString());
+ tezConf.setInt(TezConfiguration.TEZ_AM_RESOURCE_MEMORY_MB, 500);
+ tezConf.set(TezConfiguration.TEZ_AM_LAUNCH_CMD_OPTS, " -Xmx256m");
+ tezConf.setBoolean(TezConfiguration.TEZ_AM_STAGING_SCRATCH_DATA_AUTO_DELETE, false);
+ tezConf.setBoolean(RecoveryService.TEZ_TEST_RECOVERY_DRAIN_EVENTS_WHEN_STOPPED, true);
+
+ tezSession = TezClient.create("TestAMRecoveryAggregationBroadcast", tezConf);
+ tezSession.start();
+ }
+
+ @After
+ public void teardown() throws InterruptedException {
+ if (tezSession != null) {
+ try {
+ LOG.info("Stopping Tez Session");
+ tezSession.stop();
+ } catch (Exception e) {
+ LOG.error("Failed to stop Tez session", e);
+ }
+ }
+ tezSession = null;
+ }
+
+ @Test(timeout = 120000)
+ public void testSucceed() throws Exception {
+ DAG dag = createDAG("Succeed");
+ TezCounters counters = runDAGAndVerify(dag, false);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ // No retry happens
+ assertEquals(Collections.emptyList(), readRecoveryLog(2));
+ }
+
+ @Test(timeout = 120000)
+ public void testTableScanTemporalFailure() throws Exception {
+ tezConf.setBoolean(TABLE_SCAN_SLEEP, true);
+ DAG dag = createDAG("TableScanTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ @Test(timeout = 120000)
+ public void testAggregationTemporalFailure() throws Exception {
+ tezConf.setBoolean(AGGREGATION_SLEEP, true);
+ DAG dag = createDAG("AggregationTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ @Test(timeout = 120000)
+ public void testMapJoinTemporalFailure() throws Exception {
+ tezConf.setBoolean(MAP_JOIN_SLEEP, true);
+ DAG dag = createDAG("MapJoinTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ /**
+ * v1 scans lines and emit (line -> 1), imitating a simple Map vertex.
+ * v2 aggregates the lines and emit (line -> # of duplicated values), imitating an aggregation.
+ * v3 joins the output of v2 with another input. v2 broadcasts its output.
+ * (input1)
+ * \
+ * v1
+ * \
+ * v2 (input2)
+ * \ /
+ * v3
+ */
+ private DAG createDAG(String dagName) throws Exception {
+ UserPayload payload =TezUtils.createUserPayloadFromConf(tezConf);
+ DataSourceDescriptor dataSource = MRInput
+ .createConfigBuilder(new Configuration(tezConf), TextInputFormat.class,
+ INPUT_FILE.toString())
+ .build();
+ // each line -> 1
+ Vertex tableScanVertex = Vertex
+ .create(TABLE_SCAN, ProcessorDescriptor.create(TableScanProcessor.class.getName())
+ .setUserPayload(payload))
+ .addDataSource(INPUT1, dataSource);
+
+ // key -> num keys
+ Vertex aggregationVertex = Vertex
+ .create(AGGREGATION, ProcessorDescriptor
+ .create(AggregationProcessor.class.getName()).setUserPayload(payload), 1);
+
+ DataSinkDescriptor dataSink = MROutput
+ .createConfigBuilder(new Configuration(tezConf), TextOutputFormat.class,
+ OUT_PATH.toString())
+ .build();
+ // Broadcast Hash Join
+ Vertex mapJoinVertex = Vertex
+ .create(MAP_JOIN, ProcessorDescriptor.create(MapJoinProcessor.class.getName())
+ .setUserPayload(payload))
+ .addDataSource(INPUT2, dataSource)
+ .addDataSink(OUTPUT, dataSink);
+
+ EdgeProperty orderedEdge = OrderedPartitionedKVEdgeConfig
+ .newBuilder(Text.class.getName(), IntWritable.class.getName(), HashPartitioner.class.getName())
+ .setFromConfiguration(tezConf)
+ .build()
+ .createDefaultEdgeProperty();
+ EdgeProperty broadcastEdge = UnorderedKVEdgeConfig
+ .newBuilder(Text.class.getName(), IntWritable.class.getName())
+ .setFromConfiguration(tezConf)
+ .build()
+ .createDefaultBroadcastEdgeProperty();
+
+ DAG dag = DAG.create("TestAMRecoveryAggregationBroadcast_" + dagName);
+ dag.addVertex(tableScanVertex)
+ .addVertex(aggregationVertex)
+ .addVertex(mapJoinVertex)
+ .addEdge(Edge.create(tableScanVertex, aggregationVertex, orderedEdge))
+ .addEdge(Edge.create(aggregationVertex, mapJoinVertex, broadcastEdge));
+ return dag;
+ }
+
+ TezCounters runDAGAndVerify(DAG dag, boolean killAM) throws Exception {
+ tezSession.waitTillReady();
+ DAGClient dagClient = tezSession.submitDAG(dag);
+
+ if (killAM) {
+ TimeUnit.SECONDS.sleep(10);
+ YarnClient yarnClient = YarnClient.createYarnClient();
+ yarnClient.init(tezConf);
+ yarnClient.start();
+ ApplicationAttemptId id = ApplicationAttemptId.newInstance(
+ tezSession.getAppMasterApplicationId(), 1);
+ yarnClient.failApplicationAttempt(id);
+ yarnClient.close();
+ }
+ DAGStatus dagStatus = dagClient
+ .waitForCompletionWithStatusUpdates(EnumSet.of(StatusGetOpts.GET_COUNTERS));
Review Comment:
Done
https://github.com/apache/tez/pull/361/commits/7f0b3b8554ab26d481c10f665ba5ee7881135e68
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "abstractdog (via GitHub)" <gi...@apache.org>.
abstractdog commented on PR #361:
URL: https://github.com/apache/tez/pull/361#issuecomment-2559782144
> I'm checking why TestDAGRecovery failed
sometimes recovery tests are flaky, I simply restarted the last precommit job
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "tez-yetus (via GitHub)" <gi...@apache.org>.
tez-yetus commented on PR #361:
URL: https://github.com/apache/tez/pull/361#issuecomment-2162620410
:broken_heart: **-1 overall**
| Vote | Subsystem | Runtime | Comment |
|:----:|----------:|--------:|:--------|
| +0 :ok: | reexec | 27m 8s | Docker mode activated. |
||| _ Prechecks _ |
| +1 :green_heart: | dupname | 0m 0s | No case conflicting files found. |
| +1 :green_heart: | @author | 0m 0s | The patch does not contain any @author tags. |
| +1 :green_heart: | test4tests | 0m 0s | The patch appears to include 1 new or modified test files. |
||| _ master Compile Tests _ |
| +0 :ok: | mvndep | 6m 32s | Maven dependency ordering for branch |
| +1 :green_heart: | mvninstall | 12m 39s | master passed |
| +1 :green_heart: | compile | 1m 17s | master passed with JDK Ubuntu-11.0.23+9-post-Ubuntu-1ubuntu122.04.1 |
| +1 :green_heart: | compile | 1m 12s | master passed with JDK Private Build-1.8.0_412-8u412-ga-1~22.04.1-b08 |
| +1 :green_heart: | checkstyle | 1m 30s | master passed |
| +1 :green_heart: | javadoc | 1m 0s | master passed with JDK Ubuntu-11.0.23+9-post-Ubuntu-1ubuntu122.04.1 |
| +1 :green_heart: | javadoc | 0m 45s | master passed with JDK Private Build-1.8.0_412-8u412-ga-1~22.04.1-b08 |
| +0 :ok: | spotbugs | 0m 50s | Used deprecated FindBugs config; considering switching to SpotBugs. |
| +1 :green_heart: | findbugs | 2m 39s | master passed |
||| _ Patch Compile Tests _ |
| +0 :ok: | mvndep | 0m 10s | Maven dependency ordering for patch |
| +1 :green_heart: | mvninstall | 0m 49s | the patch passed |
| +1 :green_heart: | compile | 0m 53s | the patch passed with JDK Ubuntu-11.0.23+9-post-Ubuntu-1ubuntu122.04.1 |
| +1 :green_heart: | javac | 0m 53s | the patch passed |
| +1 :green_heart: | compile | 0m 45s | the patch passed with JDK Private Build-1.8.0_412-8u412-ga-1~22.04.1-b08 |
| +1 :green_heart: | javac | 0m 45s | the patch passed |
| -0 :warning: | checkstyle | 0m 11s | tez-tests: The patch generated 1 new + 0 unchanged - 0 fixed = 1 total (was 0) |
| +1 :green_heart: | whitespace | 0m 0s | The patch has no whitespace issues. |
| +1 :green_heart: | javadoc | 0m 20s | the patch passed with JDK Ubuntu-11.0.23+9-post-Ubuntu-1ubuntu122.04.1 |
| +1 :green_heart: | javadoc | 0m 19s | the patch passed with JDK Private Build-1.8.0_412-8u412-ga-1~22.04.1-b08 |
| +1 :green_heart: | findbugs | 1m 56s | the patch passed |
||| _ Other Tests _ |
| +1 :green_heart: | unit | 5m 8s | tez-dag in the patch passed. |
| +1 :green_heart: | unit | 42m 27s | tez-tests in the patch passed. |
| -1 :x: | asflicense | 0m 30s | The patch generated 1 ASF License warnings. |
| | | 109m 57s | |
| Subsystem | Report/Notes |
|----------:|:-------------|
| Docker | ClientAPI=1.45 ServerAPI=1.45 base: https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-361/1/artifact/out/Dockerfile |
| GITHUB PR | https://github.com/apache/tez/pull/361 |
| Optional Tests | dupname asflicense javac javadoc unit spotbugs findbugs checkstyle compile |
| uname | Linux 4a42ce80b08c 5.15.0-106-generic #116-Ubuntu SMP Wed Apr 17 09:17:56 UTC 2024 x86_64 x86_64 x86_64 GNU/Linux |
| Build tool | maven |
| Personality | personality/tez.sh |
| git revision | master / e08d0279c |
| Default Java | Private Build-1.8.0_412-8u412-ga-1~22.04.1-b08 |
| Multi-JDK versions | /usr/lib/jvm/java-11-openjdk-amd64:Ubuntu-11.0.23+9-post-Ubuntu-1ubuntu122.04.1 /usr/lib/jvm/java-8-openjdk-amd64:Private Build-1.8.0_412-8u412-ga-1~22.04.1-b08 |
| checkstyle | https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-361/1/artifact/out/diff-checkstyle-tez-tests.txt |
| Test Results | https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-361/1/testReport/ |
| asflicense | https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-361/1/artifact/out/patch-asflicense-problems.txt |
| Max. process+thread count | 1233 (vs. ulimit of 5500) |
| modules | C: tez-dag tez-tests U: . |
| Console output | https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-361/1/console |
| versions | git=2.34.1 maven=3.6.3 findbugs=3.0.1 |
| Powered by | Apache Yetus 0.12.0 https://yetus.apache.org |
This message was automatically generated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "tez-yetus (via GitHub)" <gi...@apache.org>.
tez-yetus commented on PR #361:
URL: https://github.com/apache/tez/pull/361#issuecomment-2559676466
:broken_heart: **-1 overall**
| Vote | Subsystem | Runtime | Comment |
|:----:|----------:|--------:|:--------|
| +0 :ok: | reexec | 28m 3s | Docker mode activated. |
||| _ Prechecks _ |
| +1 :green_heart: | dupname | 0m 0s | No case conflicting files found. |
| +1 :green_heart: | @author | 0m 0s | The patch does not contain any @author tags. |
| +1 :green_heart: | test4tests | 0m 0s | The patch appears to include 1 new or modified test files. |
||| _ master Compile Tests _ |
| +0 :ok: | mvndep | 2m 58s | Maven dependency ordering for branch |
| +1 :green_heart: | mvninstall | 10m 56s | master passed |
| +1 :green_heart: | compile | 1m 24s | master passed with JDK Ubuntu-11.0.25+9-post-Ubuntu-1ubuntu122.04 |
| +1 :green_heart: | compile | 1m 9s | master passed with JDK Private Build-1.8.0_432-8u432-ga~us1-0ubuntu2~22.04-ga |
| +1 :green_heart: | checkstyle | 1m 16s | master passed |
| +1 :green_heart: | javadoc | 0m 51s | master passed with JDK Ubuntu-11.0.25+9-post-Ubuntu-1ubuntu122.04 |
| +1 :green_heart: | javadoc | 0m 41s | master passed with JDK Private Build-1.8.0_432-8u432-ga~us1-0ubuntu2~22.04-ga |
| +0 :ok: | spotbugs | 1m 3s | Used deprecated FindBugs config; considering switching to SpotBugs. |
| +1 :green_heart: | findbugs | 2m 57s | master passed |
||| _ Patch Compile Tests _ |
| +0 :ok: | mvndep | 0m 10s | Maven dependency ordering for patch |
| +1 :green_heart: | mvninstall | 0m 51s | the patch passed |
| +1 :green_heart: | compile | 0m 55s | the patch passed with JDK Ubuntu-11.0.25+9-post-Ubuntu-1ubuntu122.04 |
| +1 :green_heart: | javac | 0m 55s | the patch passed |
| +1 :green_heart: | compile | 0m 48s | the patch passed with JDK Private Build-1.8.0_432-8u432-ga~us1-0ubuntu2~22.04-ga |
| +1 :green_heart: | javac | 0m 48s | the patch passed |
| -0 :warning: | checkstyle | 0m 11s | tez-tests: The patch generated 1 new + 0 unchanged - 0 fixed = 1 total (was 0) |
| +1 :green_heart: | whitespace | 0m 0s | The patch has no whitespace issues. |
| +1 :green_heart: | javadoc | 0m 19s | the patch passed with JDK Ubuntu-11.0.25+9-post-Ubuntu-1ubuntu122.04 |
| +1 :green_heart: | javadoc | 0m 19s | the patch passed with JDK Private Build-1.8.0_432-8u432-ga~us1-0ubuntu2~22.04-ga |
| +1 :green_heart: | findbugs | 2m 0s | the patch passed |
||| _ Other Tests _ |
| -1 :x: | unit | 5m 2s | tez-dag in the patch failed. |
| +1 :green_heart: | unit | 43m 19s | tez-tests in the patch passed. |
| +1 :green_heart: | asflicense | 0m 21s | The patch does not generate ASF License warnings. |
| | | 106m 23s | |
| Reason | Tests |
|-------:|:------|
| Failed junit tests | tez.dag.app.dag.impl.TestDAGRecovery |
| Subsystem | Report/Notes |
|----------:|:-------------|
| Docker | ClientAPI=1.47 ServerAPI=1.47 base: https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-361/3/artifact/out/Dockerfile |
| GITHUB PR | https://github.com/apache/tez/pull/361 |
| JIRA Issue | TEZ-4569 |
| Optional Tests | dupname asflicense javac javadoc unit spotbugs findbugs checkstyle compile |
| uname | Linux 1c2c562e42b5 5.15.0-124-generic #134-Ubuntu SMP Fri Sep 27 20:20:17 UTC 2024 x86_64 x86_64 x86_64 GNU/Linux |
| Build tool | maven |
| Personality | personality/tez.sh |
| git revision | master / 9efa6f14d |
| Default Java | Private Build-1.8.0_432-8u432-ga~us1-0ubuntu2~22.04-ga |
| Multi-JDK versions | /usr/lib/jvm/java-11-openjdk-amd64:Ubuntu-11.0.25+9-post-Ubuntu-1ubuntu122.04 /usr/lib/jvm/java-8-openjdk-amd64:Private Build-1.8.0_432-8u432-ga~us1-0ubuntu2~22.04-ga |
| checkstyle | https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-361/3/artifact/out/diff-checkstyle-tez-tests.txt |
| unit | https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-361/3/artifact/out/patch-unit-tez-dag.txt |
| Test Results | https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-361/3/testReport/ |
| Max. process+thread count | 1134 (vs. ulimit of 5500) |
| modules | C: tez-dag tez-tests U: . |
| Console output | https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-361/3/console |
| versions | git=2.34.1 maven=3.6.3 findbugs=3.0.1 |
| Powered by | Apache Yetus 0.12.0 https://yetus.apache.org |
This message was automatically generated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "abstractdog (via GitHub)" <gi...@apache.org>.
abstractdog commented on code in PR #361:
URL: https://github.com/apache/tez/pull/361#discussion_r1894629769
##########
tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java:
##########
@@ -2817,6 +2817,18 @@ private VertexState setupVertex() {
return VertexState.INITED;
}
+ private boolean canSkipInitializingParents() {
+ // Both cases use RootInputVertexManager. RootInputVertexManager can start tasks even though
+ // any parents are not fully initialized.
+ if (vertexPlan.hasVertexManagerPlugin()) {
+ final VertexManagerPluginDescriptor pluginDesc = DagTypeConverters
+ .convertVertexManagerPluginDescriptorFromDAGPlan(vertexPlan.getVertexManagerPlugin());
+ return pluginDesc.getClassName().equals(RootInputVertexManager.class.getName());
+ } else {
+ return inputsWithInitializers != null;
Review Comment:
in the else branch: it's not a RootInputVertexManager, but has inputsWithInitializers, what kind of plugin is used here if any?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "okumin (via GitHub)" <gi...@apache.org>.
okumin commented on PR #361:
URL: https://github.com/apache/tez/pull/361#issuecomment-2559744712
I'm checking why TestDAGRecovery failed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "abstractdog (via GitHub)" <gi...@apache.org>.
abstractdog commented on code in PR #361:
URL: https://github.com/apache/tez/pull/361#discussion_r1895387136
##########
tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java:
##########
@@ -2843,7 +2855,7 @@ private void assignVertexManager() throws TezException {
if (recoveryData != null && recoveryData.shouldSkipInit()
Review Comment:
yeah, that would makes sense
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "abstractdog (via GitHub)" <gi...@apache.org>.
abstractdog commented on code in PR #361:
URL: https://github.com/apache/tez/pull/361#discussion_r1894902860
##########
tez-tests/src/test/java/org/apache/tez/test/TestAMRecoveryAggregationBroadcast.java:
##########
@@ -0,0 +1,513 @@
+/**
+ * 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.tez.test;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.apache.tez.client.TezClient;
+import org.apache.tez.client.TezClientUtils;
+import org.apache.tez.common.Preconditions;
+import org.apache.tez.common.TezCommonUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.DAGCounter;
+import org.apache.tez.common.counters.TezCounters;
+import org.apache.tez.dag.api.DAG;
+import org.apache.tez.dag.api.DataSinkDescriptor;
+import org.apache.tez.dag.api.DataSourceDescriptor;
+import org.apache.tez.dag.api.Edge;
+import org.apache.tez.dag.api.EdgeProperty;
+import org.apache.tez.dag.api.ProcessorDescriptor;
+import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.api.TezConstants;
+import org.apache.tez.dag.api.UserPayload;
+import org.apache.tez.dag.api.Vertex;
+import org.apache.tez.dag.api.client.DAGClient;
+import org.apache.tez.dag.api.client.DAGStatus;
+import org.apache.tez.dag.api.client.DAGStatus.State;
+import org.apache.tez.dag.api.client.StatusGetOpts;
+import org.apache.tez.dag.api.oldrecords.TaskAttemptState;
+import org.apache.tez.dag.app.RecoveryParser;
+import org.apache.tez.dag.history.HistoryEvent;
+import org.apache.tez.dag.history.HistoryEventType;
+import org.apache.tez.dag.history.events.TaskAttemptFinishedEvent;
+import org.apache.tez.dag.history.recovery.RecoveryService;
+import org.apache.tez.mapreduce.input.MRInput;
+import org.apache.tez.mapreduce.output.MROutput;
+import org.apache.tez.mapreduce.processor.SimpleMRProcessor;
+import org.apache.tez.runtime.api.ProcessorContext;
+import org.apache.tez.runtime.library.api.KeyValueReader;
+import org.apache.tez.runtime.library.api.KeyValueWriter;
+import org.apache.tez.runtime.library.api.KeyValuesReader;
+import org.apache.tez.runtime.library.conf.OrderedPartitionedKVEdgeConfig;
+import org.apache.tez.runtime.library.conf.UnorderedKVEdgeConfig;
+import org.apache.tez.runtime.library.partitioner.HashPartitioner;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TestAMRecoveryAggregationBroadcast {
+ private static final Logger LOG = LoggerFactory.getLogger(TestAMRecoveryAggregationBroadcast.class);
+ private static final String INPUT1 = "Input";
+ private static final String INPUT2 = "Input";
+ private static final String OUTPUT = "Output";
+ private static final String TABLE_SCAN = "TableScan";
+ private static final String AGGREGATION = "Aggregation";
+ private static final String MAP_JOIN = "MapJoin";
+ private static final String TEST_ROOT_DIR = "target" + Path.SEPARATOR
+ + TestAMRecoveryAggregationBroadcast.class.getName() + "-tmpDir";
+ private static final Path INPUT_FILE = new Path(TEST_ROOT_DIR, "input.csv");
+ private static final Path OUT_PATH = new Path(TEST_ROOT_DIR, "out-groups");
+ private static final String EXPECTED_OUTPUT = "1-5\n1-5\n1-5\n1-5\n1-5\n"
+ + "2-4\n2-4\n2-4\n2-4\n" + "3-3\n3-3\n3-3\n" + "4-2\n4-2\n" + "5-1\n";
+ private static final String TABLE_SCAN_SLEEP = "tez.test.table.scan.sleep";
+ private static final String AGGREGATION_SLEEP = "tez.test.aggregation.sleep";
+ private static final String MAP_JOIN_SLEEP = "tez.test.map.join.sleep";
+
+ private static Configuration dfsConf;
+ private static MiniDFSCluster dfsCluster;
+ private static MiniTezCluster tezCluster;
+ private static FileSystem remoteFs;
+
+ private TezConfiguration tezConf;
+ private TezClient tezSession;
+
+ @BeforeClass
+ public static void setupAll() {
+ try {
+ dfsConf = new Configuration();
+ dfsConf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, TEST_ROOT_DIR);
+ dfsCluster = new MiniDFSCluster.Builder(dfsConf).numDataNodes(3)
+ .format(true).racks(null).build();
+ remoteFs = dfsCluster.getFileSystem();
+ createSampleFile();
+ } catch (IOException io) {
+ throw new RuntimeException("problem starting mini dfs cluster", io);
+ }
+
+ if (tezCluster == null) {
+ tezCluster = new MiniTezCluster(TestAMRecoveryAggregationBroadcast.class.getName(), 1, 1, 1);
+ Configuration conf = new Configuration(dfsConf);
+ conf.set("fs.defaultFS", remoteFs.getUri().toString()); // use HDFS
+ conf.setInt("yarn.nodemanager.delete.debug-delay-sec", 20000);
+ conf.setLong(TezConfiguration.TEZ_AM_SLEEP_TIME_BEFORE_EXIT_MILLIS, 500);
+ tezCluster.init(conf);
+ tezCluster.start();
+ }
+ }
+
+ private static void createSampleFile() throws IOException {
+ FSDataOutputStream out = remoteFs.create(INPUT_FILE);
+ BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(out));
+ // 1, 1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 4, 4, 5
+ for (int i = 1; i <= 5; i++) {
+ for (int j = 0; j <= 5 - i; j++) {
+ writer.write(String.valueOf(i));
+ writer.newLine();
+ }
+ }
+ writer.close();
+ }
+
+ @AfterClass
+ public static void tearDownAll() {
+ if (tezCluster != null) {
+ tezCluster.stop();
+ tezCluster = null;
+ }
+ if (dfsCluster != null) {
+ dfsCluster.shutdown(true);
+ dfsCluster = null;
+ }
+ }
+
+ @Before
+ public void setup() throws Exception {
+ Path remoteStagingDir = remoteFs.makeQualified(new Path(TEST_ROOT_DIR, String
+ .valueOf(new Random().nextInt(100000))));
+ TezClientUtils.ensureStagingDirExists(dfsConf, remoteStagingDir);
+
+ tezConf = new TezConfiguration(tezCluster.getConfig());
+ tezConf.setInt(TezConfiguration.DAG_RECOVERY_MAX_UNFLUSHED_EVENTS, 0);
+ tezConf.set(TezConfiguration.TEZ_AM_LOG_LEVEL, "INFO");
+ tezConf.set(TezConfiguration.TEZ_AM_STAGING_DIR, remoteStagingDir.toString());
+ tezConf.setInt(TezConfiguration.TEZ_AM_RESOURCE_MEMORY_MB, 500);
+ tezConf.set(TezConfiguration.TEZ_AM_LAUNCH_CMD_OPTS, " -Xmx256m");
+ tezConf.setBoolean(TezConfiguration.TEZ_AM_STAGING_SCRATCH_DATA_AUTO_DELETE, false);
+ tezConf.setBoolean(RecoveryService.TEZ_TEST_RECOVERY_DRAIN_EVENTS_WHEN_STOPPED, true);
+
+ tezSession = TezClient.create("TestAMRecoveryAggregationBroadcast", tezConf);
+ tezSession.start();
+ }
+
+ @After
+ public void teardown() throws InterruptedException {
+ if (tezSession != null) {
+ try {
+ LOG.info("Stopping Tez Session");
+ tezSession.stop();
+ } catch (Exception e) {
+ LOG.error("Failed to stop Tez session", e);
+ }
+ }
+ tezSession = null;
+ }
+
+ @Test(timeout = 120000)
+ public void testSucceed() throws Exception {
+ DAG dag = createDAG("Succeed");
+ TezCounters counters = runDAGAndVerify(dag, false);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ // No retry happens
+ assertEquals(Collections.emptyList(), readRecoveryLog(2));
+ }
+
+ @Test(timeout = 120000)
+ public void testTableScanTemporalFailure() throws Exception {
+ tezConf.setBoolean(TABLE_SCAN_SLEEP, true);
+ DAG dag = createDAG("TableScanTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ @Test(timeout = 120000)
+ public void testAggregationTemporalFailure() throws Exception {
+ tezConf.setBoolean(AGGREGATION_SLEEP, true);
+ DAG dag = createDAG("AggregationTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ @Test(timeout = 120000)
+ public void testMapJoinTemporalFailure() throws Exception {
+ tezConf.setBoolean(MAP_JOIN_SLEEP, true);
+ DAG dag = createDAG("MapJoinTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ /**
+ * v1 scans lines and emit (line -> 1), imitating a simple Map vertex.
+ * v2 aggregates the lines and emit (line -> # of duplicated values), imitating an aggregation.
+ * v3 joins the output of v2 with another input. v2 broadcasts its output.
+ * (input1)
+ * \
+ * v1
+ * \
+ * v2 (input2)
+ * \ /
+ * v3
+ */
+ private DAG createDAG(String dagName) throws Exception {
+ UserPayload payload =TezUtils.createUserPayloadFromConf(tezConf);
+ DataSourceDescriptor dataSource = MRInput
+ .createConfigBuilder(new Configuration(tezConf), TextInputFormat.class,
+ INPUT_FILE.toString())
+ .build();
+ // each line -> 1
+ Vertex tableScanVertex = Vertex
+ .create(TABLE_SCAN, ProcessorDescriptor.create(TableScanProcessor.class.getName())
+ .setUserPayload(payload))
+ .addDataSource(INPUT1, dataSource);
+
+ // key -> num keys
+ Vertex aggregationVertex = Vertex
+ .create(AGGREGATION, ProcessorDescriptor
+ .create(AggregationProcessor.class.getName()).setUserPayload(payload), 1);
+
+ DataSinkDescriptor dataSink = MROutput
+ .createConfigBuilder(new Configuration(tezConf), TextOutputFormat.class,
+ OUT_PATH.toString())
+ .build();
+ // Broadcast Hash Join
+ Vertex mapJoinVertex = Vertex
+ .create(MAP_JOIN, ProcessorDescriptor.create(MapJoinProcessor.class.getName())
+ .setUserPayload(payload))
+ .addDataSource(INPUT2, dataSource)
+ .addDataSink(OUTPUT, dataSink);
+
+ EdgeProperty orderedEdge = OrderedPartitionedKVEdgeConfig
+ .newBuilder(Text.class.getName(), IntWritable.class.getName(), HashPartitioner.class.getName())
+ .setFromConfiguration(tezConf)
+ .build()
+ .createDefaultEdgeProperty();
+ EdgeProperty broadcastEdge = UnorderedKVEdgeConfig
+ .newBuilder(Text.class.getName(), IntWritable.class.getName())
+ .setFromConfiguration(tezConf)
+ .build()
+ .createDefaultBroadcastEdgeProperty();
+
+ DAG dag = DAG.create("TestAMRecoveryAggregationBroadcast_" + dagName);
+ dag.addVertex(tableScanVertex)
+ .addVertex(aggregationVertex)
+ .addVertex(mapJoinVertex)
+ .addEdge(Edge.create(tableScanVertex, aggregationVertex, orderedEdge))
+ .addEdge(Edge.create(aggregationVertex, mapJoinVertex, broadcastEdge));
+ return dag;
+ }
+
+ TezCounters runDAGAndVerify(DAG dag, boolean killAM) throws Exception {
+ tezSession.waitTillReady();
+ DAGClient dagClient = tezSession.submitDAG(dag);
+
+ if (killAM) {
+ TimeUnit.SECONDS.sleep(10);
+ YarnClient yarnClient = YarnClient.createYarnClient();
+ yarnClient.init(tezConf);
+ yarnClient.start();
+ ApplicationAttemptId id = ApplicationAttemptId.newInstance(
+ tezSession.getAppMasterApplicationId(), 1);
+ yarnClient.failApplicationAttempt(id);
+ yarnClient.close();
+ }
+ DAGStatus dagStatus = dagClient
+ .waitForCompletionWithStatusUpdates(EnumSet.of(StatusGetOpts.GET_COUNTERS));
+ LOG.info("Diagnosis: " + dagStatus.getDiagnostics());
+ Assert.assertEquals(State.SUCCEEDED, dagStatus.getState());
+
+ FSDataInputStream in = remoteFs.open(new Path(OUT_PATH, "part-v002-o000-r-00000"));
+ ByteBuffer buf = ByteBuffer.allocate(100);
+ in.read(buf);
+ buf.flip();
+ Assert.assertEquals(EXPECTED_OUTPUT, StandardCharsets.UTF_8.decode(buf).toString());
+ return dagStatus.getDAGCounters();
+ }
+
+ private List<HistoryEvent> readRecoveryLog(int attemptNum) throws IOException {
+ ApplicationId appId = tezSession.getAppMasterApplicationId();
+ Path tezSystemStagingDir = TezCommonUtils.getTezSystemStagingPath(tezConf, appId.toString());
+ Path recoveryDataDir = TezCommonUtils.getRecoveryPath(tezSystemStagingDir, tezConf);
+ FileSystem fs = tezSystemStagingDir.getFileSystem(tezConf);
+ List<HistoryEvent> historyEvents = new ArrayList<>();
+ Path currentAttemptRecoveryDataDir = TezCommonUtils.getAttemptRecoveryPath(recoveryDataDir, attemptNum);
+ Path recoveryFilePath =
+ new Path(currentAttemptRecoveryDataDir, appId.toString().replace("application", "dag")
+ + "_1" + TezConstants.DAG_RECOVERY_RECOVER_FILE_SUFFIX);
+ if (fs.exists(recoveryFilePath)) {
+ LOG.info("Read recovery file:" + recoveryFilePath);
+ historyEvents.addAll(RecoveryParser.parseDAGRecoveryFile(fs.open(recoveryFilePath)));
+ }
+ printHistoryEvents(historyEvents, attemptNum);
+ return historyEvents;
+ }
+
+ 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("");
Review Comment:
hm, this looks strange, is this really needed? printed only once after all the events are printed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "abstractdog (via GitHub)" <gi...@apache.org>.
abstractdog merged PR #361:
URL: https://github.com/apache/tez/pull/361
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "okumin (via GitHub)" <gi...@apache.org>.
okumin commented on code in PR #361:
URL: https://github.com/apache/tez/pull/361#discussion_r1895635642
##########
tez-tests/src/test/java/org/apache/tez/test/TestAMRecoveryAggregationBroadcast.java:
##########
@@ -0,0 +1,513 @@
+/**
+ * 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.tez.test;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.apache.tez.client.TezClient;
+import org.apache.tez.client.TezClientUtils;
+import org.apache.tez.common.Preconditions;
+import org.apache.tez.common.TezCommonUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.DAGCounter;
+import org.apache.tez.common.counters.TezCounters;
+import org.apache.tez.dag.api.DAG;
+import org.apache.tez.dag.api.DataSinkDescriptor;
+import org.apache.tez.dag.api.DataSourceDescriptor;
+import org.apache.tez.dag.api.Edge;
+import org.apache.tez.dag.api.EdgeProperty;
+import org.apache.tez.dag.api.ProcessorDescriptor;
+import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.api.TezConstants;
+import org.apache.tez.dag.api.UserPayload;
+import org.apache.tez.dag.api.Vertex;
+import org.apache.tez.dag.api.client.DAGClient;
+import org.apache.tez.dag.api.client.DAGStatus;
+import org.apache.tez.dag.api.client.DAGStatus.State;
+import org.apache.tez.dag.api.client.StatusGetOpts;
+import org.apache.tez.dag.api.oldrecords.TaskAttemptState;
+import org.apache.tez.dag.app.RecoveryParser;
+import org.apache.tez.dag.history.HistoryEvent;
+import org.apache.tez.dag.history.HistoryEventType;
+import org.apache.tez.dag.history.events.TaskAttemptFinishedEvent;
+import org.apache.tez.dag.history.recovery.RecoveryService;
+import org.apache.tez.mapreduce.input.MRInput;
+import org.apache.tez.mapreduce.output.MROutput;
+import org.apache.tez.mapreduce.processor.SimpleMRProcessor;
+import org.apache.tez.runtime.api.ProcessorContext;
+import org.apache.tez.runtime.library.api.KeyValueReader;
+import org.apache.tez.runtime.library.api.KeyValueWriter;
+import org.apache.tez.runtime.library.api.KeyValuesReader;
+import org.apache.tez.runtime.library.conf.OrderedPartitionedKVEdgeConfig;
+import org.apache.tez.runtime.library.conf.UnorderedKVEdgeConfig;
+import org.apache.tez.runtime.library.partitioner.HashPartitioner;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TestAMRecoveryAggregationBroadcast {
+ private static final Logger LOG = LoggerFactory.getLogger(TestAMRecoveryAggregationBroadcast.class);
+ private static final String INPUT1 = "Input";
+ private static final String INPUT2 = "Input";
+ private static final String OUTPUT = "Output";
+ private static final String TABLE_SCAN = "TableScan";
+ private static final String AGGREGATION = "Aggregation";
+ private static final String MAP_JOIN = "MapJoin";
+ private static final String TEST_ROOT_DIR = "target" + Path.SEPARATOR
+ + TestAMRecoveryAggregationBroadcast.class.getName() + "-tmpDir";
+ private static final Path INPUT_FILE = new Path(TEST_ROOT_DIR, "input.csv");
+ private static final Path OUT_PATH = new Path(TEST_ROOT_DIR, "out-groups");
+ private static final String EXPECTED_OUTPUT = "1-5\n1-5\n1-5\n1-5\n1-5\n"
+ + "2-4\n2-4\n2-4\n2-4\n" + "3-3\n3-3\n3-3\n" + "4-2\n4-2\n" + "5-1\n";
+ private static final String TABLE_SCAN_SLEEP = "tez.test.table.scan.sleep";
+ private static final String AGGREGATION_SLEEP = "tez.test.aggregation.sleep";
+ private static final String MAP_JOIN_SLEEP = "tez.test.map.join.sleep";
+
+ private static Configuration dfsConf;
+ private static MiniDFSCluster dfsCluster;
+ private static MiniTezCluster tezCluster;
+ private static FileSystem remoteFs;
+
+ private TezConfiguration tezConf;
+ private TezClient tezSession;
+
+ @BeforeClass
+ public static void setupAll() {
+ try {
+ dfsConf = new Configuration();
+ dfsConf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, TEST_ROOT_DIR);
+ dfsCluster = new MiniDFSCluster.Builder(dfsConf).numDataNodes(3)
+ .format(true).racks(null).build();
+ remoteFs = dfsCluster.getFileSystem();
+ createSampleFile();
+ } catch (IOException io) {
+ throw new RuntimeException("problem starting mini dfs cluster", io);
+ }
+
+ if (tezCluster == null) {
+ tezCluster = new MiniTezCluster(TestAMRecoveryAggregationBroadcast.class.getName(), 1, 1, 1);
+ Configuration conf = new Configuration(dfsConf);
+ conf.set("fs.defaultFS", remoteFs.getUri().toString()); // use HDFS
+ conf.setInt("yarn.nodemanager.delete.debug-delay-sec", 20000);
+ conf.setLong(TezConfiguration.TEZ_AM_SLEEP_TIME_BEFORE_EXIT_MILLIS, 500);
+ tezCluster.init(conf);
+ tezCluster.start();
+ }
+ }
+
+ private static void createSampleFile() throws IOException {
+ FSDataOutputStream out = remoteFs.create(INPUT_FILE);
+ BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(out));
+ // 1, 1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 4, 4, 5
+ for (int i = 1; i <= 5; i++) {
+ for (int j = 0; j <= 5 - i; j++) {
+ writer.write(String.valueOf(i));
+ writer.newLine();
+ }
+ }
+ writer.close();
+ }
+
+ @AfterClass
+ public static void tearDownAll() {
+ if (tezCluster != null) {
+ tezCluster.stop();
+ tezCluster = null;
+ }
+ if (dfsCluster != null) {
+ dfsCluster.shutdown(true);
+ dfsCluster = null;
+ }
+ }
+
+ @Before
+ public void setup() throws Exception {
+ Path remoteStagingDir = remoteFs.makeQualified(new Path(TEST_ROOT_DIR, String
+ .valueOf(new Random().nextInt(100000))));
+ TezClientUtils.ensureStagingDirExists(dfsConf, remoteStagingDir);
+
+ tezConf = new TezConfiguration(tezCluster.getConfig());
+ tezConf.setInt(TezConfiguration.DAG_RECOVERY_MAX_UNFLUSHED_EVENTS, 0);
+ tezConf.set(TezConfiguration.TEZ_AM_LOG_LEVEL, "INFO");
+ tezConf.set(TezConfiguration.TEZ_AM_STAGING_DIR, remoteStagingDir.toString());
+ tezConf.setInt(TezConfiguration.TEZ_AM_RESOURCE_MEMORY_MB, 500);
+ tezConf.set(TezConfiguration.TEZ_AM_LAUNCH_CMD_OPTS, " -Xmx256m");
+ tezConf.setBoolean(TezConfiguration.TEZ_AM_STAGING_SCRATCH_DATA_AUTO_DELETE, false);
+ tezConf.setBoolean(RecoveryService.TEZ_TEST_RECOVERY_DRAIN_EVENTS_WHEN_STOPPED, true);
+
+ tezSession = TezClient.create("TestAMRecoveryAggregationBroadcast", tezConf);
+ tezSession.start();
+ }
+
+ @After
+ public void teardown() throws InterruptedException {
+ if (tezSession != null) {
+ try {
+ LOG.info("Stopping Tez Session");
+ tezSession.stop();
+ } catch (Exception e) {
+ LOG.error("Failed to stop Tez session", e);
+ }
+ }
+ tezSession = null;
+ }
+
+ @Test(timeout = 120000)
+ public void testSucceed() throws Exception {
+ DAG dag = createDAG("Succeed");
+ TezCounters counters = runDAGAndVerify(dag, false);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ // No retry happens
+ assertEquals(Collections.emptyList(), readRecoveryLog(2));
+ }
+
+ @Test(timeout = 120000)
+ public void testTableScanTemporalFailure() throws Exception {
+ tezConf.setBoolean(TABLE_SCAN_SLEEP, true);
+ DAG dag = createDAG("TableScanTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ @Test(timeout = 120000)
+ public void testAggregationTemporalFailure() throws Exception {
+ tezConf.setBoolean(AGGREGATION_SLEEP, true);
+ DAG dag = createDAG("AggregationTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ @Test(timeout = 120000)
+ public void testMapJoinTemporalFailure() throws Exception {
+ tezConf.setBoolean(MAP_JOIN_SLEEP, true);
+ DAG dag = createDAG("MapJoinTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ /**
+ * v1 scans lines and emit (line -> 1), imitating a simple Map vertex.
+ * v2 aggregates the lines and emit (line -> # of duplicated values), imitating an aggregation.
+ * v3 joins the output of v2 with another input. v2 broadcasts its output.
+ * (input1)
+ * \
+ * v1
+ * \
+ * v2 (input2)
+ * \ /
+ * v3
+ */
+ private DAG createDAG(String dagName) throws Exception {
+ UserPayload payload =TezUtils.createUserPayloadFromConf(tezConf);
Review Comment:
Done
https://github.com/apache/tez/pull/361/commits/7f0b3b8554ab26d481c10f665ba5ee7881135e68
##########
tez-tests/src/test/java/org/apache/tez/test/TestAMRecoveryAggregationBroadcast.java:
##########
@@ -0,0 +1,513 @@
+/**
+ * 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.tez.test;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.apache.tez.client.TezClient;
+import org.apache.tez.client.TezClientUtils;
+import org.apache.tez.common.Preconditions;
+import org.apache.tez.common.TezCommonUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.DAGCounter;
+import org.apache.tez.common.counters.TezCounters;
+import org.apache.tez.dag.api.DAG;
+import org.apache.tez.dag.api.DataSinkDescriptor;
+import org.apache.tez.dag.api.DataSourceDescriptor;
+import org.apache.tez.dag.api.Edge;
+import org.apache.tez.dag.api.EdgeProperty;
+import org.apache.tez.dag.api.ProcessorDescriptor;
+import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.api.TezConstants;
+import org.apache.tez.dag.api.UserPayload;
+import org.apache.tez.dag.api.Vertex;
+import org.apache.tez.dag.api.client.DAGClient;
+import org.apache.tez.dag.api.client.DAGStatus;
+import org.apache.tez.dag.api.client.DAGStatus.State;
+import org.apache.tez.dag.api.client.StatusGetOpts;
+import org.apache.tez.dag.api.oldrecords.TaskAttemptState;
+import org.apache.tez.dag.app.RecoveryParser;
+import org.apache.tez.dag.history.HistoryEvent;
+import org.apache.tez.dag.history.HistoryEventType;
+import org.apache.tez.dag.history.events.TaskAttemptFinishedEvent;
+import org.apache.tez.dag.history.recovery.RecoveryService;
+import org.apache.tez.mapreduce.input.MRInput;
+import org.apache.tez.mapreduce.output.MROutput;
+import org.apache.tez.mapreduce.processor.SimpleMRProcessor;
+import org.apache.tez.runtime.api.ProcessorContext;
+import org.apache.tez.runtime.library.api.KeyValueReader;
+import org.apache.tez.runtime.library.api.KeyValueWriter;
+import org.apache.tez.runtime.library.api.KeyValuesReader;
+import org.apache.tez.runtime.library.conf.OrderedPartitionedKVEdgeConfig;
+import org.apache.tez.runtime.library.conf.UnorderedKVEdgeConfig;
+import org.apache.tez.runtime.library.partitioner.HashPartitioner;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TestAMRecoveryAggregationBroadcast {
+ private static final Logger LOG = LoggerFactory.getLogger(TestAMRecoveryAggregationBroadcast.class);
+ private static final String INPUT1 = "Input";
+ private static final String INPUT2 = "Input";
+ private static final String OUTPUT = "Output";
+ private static final String TABLE_SCAN = "TableScan";
+ private static final String AGGREGATION = "Aggregation";
+ private static final String MAP_JOIN = "MapJoin";
+ private static final String TEST_ROOT_DIR = "target" + Path.SEPARATOR
+ + TestAMRecoveryAggregationBroadcast.class.getName() + "-tmpDir";
+ private static final Path INPUT_FILE = new Path(TEST_ROOT_DIR, "input.csv");
+ private static final Path OUT_PATH = new Path(TEST_ROOT_DIR, "out-groups");
+ private static final String EXPECTED_OUTPUT = "1-5\n1-5\n1-5\n1-5\n1-5\n"
+ + "2-4\n2-4\n2-4\n2-4\n" + "3-3\n3-3\n3-3\n" + "4-2\n4-2\n" + "5-1\n";
+ private static final String TABLE_SCAN_SLEEP = "tez.test.table.scan.sleep";
+ private static final String AGGREGATION_SLEEP = "tez.test.aggregation.sleep";
+ private static final String MAP_JOIN_SLEEP = "tez.test.map.join.sleep";
+
+ private static Configuration dfsConf;
+ private static MiniDFSCluster dfsCluster;
+ private static MiniTezCluster tezCluster;
+ private static FileSystem remoteFs;
+
+ private TezConfiguration tezConf;
+ private TezClient tezSession;
+
+ @BeforeClass
+ public static void setupAll() {
+ try {
+ dfsConf = new Configuration();
+ dfsConf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, TEST_ROOT_DIR);
+ dfsCluster = new MiniDFSCluster.Builder(dfsConf).numDataNodes(3)
+ .format(true).racks(null).build();
+ remoteFs = dfsCluster.getFileSystem();
+ createSampleFile();
+ } catch (IOException io) {
+ throw new RuntimeException("problem starting mini dfs cluster", io);
+ }
+
+ if (tezCluster == null) {
+ tezCluster = new MiniTezCluster(TestAMRecoveryAggregationBroadcast.class.getName(), 1, 1, 1);
+ Configuration conf = new Configuration(dfsConf);
+ conf.set("fs.defaultFS", remoteFs.getUri().toString()); // use HDFS
+ conf.setInt("yarn.nodemanager.delete.debug-delay-sec", 20000);
+ conf.setLong(TezConfiguration.TEZ_AM_SLEEP_TIME_BEFORE_EXIT_MILLIS, 500);
+ tezCluster.init(conf);
+ tezCluster.start();
+ }
+ }
+
+ private static void createSampleFile() throws IOException {
+ FSDataOutputStream out = remoteFs.create(INPUT_FILE);
+ BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(out));
+ // 1, 1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 4, 4, 5
+ for (int i = 1; i <= 5; i++) {
+ for (int j = 0; j <= 5 - i; j++) {
+ writer.write(String.valueOf(i));
+ writer.newLine();
+ }
+ }
+ writer.close();
+ }
+
+ @AfterClass
+ public static void tearDownAll() {
+ if (tezCluster != null) {
+ tezCluster.stop();
+ tezCluster = null;
+ }
+ if (dfsCluster != null) {
+ dfsCluster.shutdown(true);
+ dfsCluster = null;
+ }
+ }
+
+ @Before
+ public void setup() throws Exception {
+ Path remoteStagingDir = remoteFs.makeQualified(new Path(TEST_ROOT_DIR, String
+ .valueOf(new Random().nextInt(100000))));
+ TezClientUtils.ensureStagingDirExists(dfsConf, remoteStagingDir);
+
+ tezConf = new TezConfiguration(tezCluster.getConfig());
+ tezConf.setInt(TezConfiguration.DAG_RECOVERY_MAX_UNFLUSHED_EVENTS, 0);
+ tezConf.set(TezConfiguration.TEZ_AM_LOG_LEVEL, "INFO");
+ tezConf.set(TezConfiguration.TEZ_AM_STAGING_DIR, remoteStagingDir.toString());
+ tezConf.setInt(TezConfiguration.TEZ_AM_RESOURCE_MEMORY_MB, 500);
+ tezConf.set(TezConfiguration.TEZ_AM_LAUNCH_CMD_OPTS, " -Xmx256m");
+ tezConf.setBoolean(TezConfiguration.TEZ_AM_STAGING_SCRATCH_DATA_AUTO_DELETE, false);
+ tezConf.setBoolean(RecoveryService.TEZ_TEST_RECOVERY_DRAIN_EVENTS_WHEN_STOPPED, true);
+
+ tezSession = TezClient.create("TestAMRecoveryAggregationBroadcast", tezConf);
+ tezSession.start();
+ }
+
+ @After
+ public void teardown() throws InterruptedException {
+ if (tezSession != null) {
+ try {
+ LOG.info("Stopping Tez Session");
+ tezSession.stop();
+ } catch (Exception e) {
+ LOG.error("Failed to stop Tez session", e);
+ }
+ }
+ tezSession = null;
+ }
+
+ @Test(timeout = 120000)
+ public void testSucceed() throws Exception {
+ DAG dag = createDAG("Succeed");
+ TezCounters counters = runDAGAndVerify(dag, false);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ // No retry happens
+ assertEquals(Collections.emptyList(), readRecoveryLog(2));
+ }
+
+ @Test(timeout = 120000)
+ public void testTableScanTemporalFailure() throws Exception {
+ tezConf.setBoolean(TABLE_SCAN_SLEEP, true);
+ DAG dag = createDAG("TableScanTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ @Test(timeout = 120000)
+ public void testAggregationTemporalFailure() throws Exception {
+ tezConf.setBoolean(AGGREGATION_SLEEP, true);
+ DAG dag = createDAG("AggregationTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ @Test(timeout = 120000)
+ public void testMapJoinTemporalFailure() throws Exception {
+ tezConf.setBoolean(MAP_JOIN_SLEEP, true);
+ DAG dag = createDAG("MapJoinTemporalFailure");
+ TezCounters counters = runDAGAndVerify(dag, true);
+ assertEquals(3, counters.findCounter(DAGCounter.NUM_SUCCEEDED_TASKS).getValue());
+
+ List<HistoryEvent> historyEvents1 = readRecoveryLog(1);
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 0, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents1, 1, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents1, 2, 0).size());
+
+ List<HistoryEvent> historyEvents2 = readRecoveryLog(2);
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 0, 0).size());
+ assertEquals(0, findTaskAttemptFinishedEvent(historyEvents2, 1, 0).size());
+ assertEquals(1, findTaskAttemptFinishedEvent(historyEvents2, 2, 0).size());
+
+ assertEquals(Collections.emptyList(), readRecoveryLog(3));
+ }
+
+ /**
+ * v1 scans lines and emit (line -> 1), imitating a simple Map vertex.
+ * v2 aggregates the lines and emit (line -> # of duplicated values), imitating an aggregation.
+ * v3 joins the output of v2 with another input. v2 broadcasts its output.
+ * (input1)
+ * \
+ * v1
+ * \
+ * v2 (input2)
+ * \ /
+ * v3
+ */
+ private DAG createDAG(String dagName) throws Exception {
+ UserPayload payload =TezUtils.createUserPayloadFromConf(tezConf);
+ DataSourceDescriptor dataSource = MRInput
+ .createConfigBuilder(new Configuration(tezConf), TextInputFormat.class,
+ INPUT_FILE.toString())
+ .build();
+ // each line -> 1
+ Vertex tableScanVertex = Vertex
+ .create(TABLE_SCAN, ProcessorDescriptor.create(TableScanProcessor.class.getName())
+ .setUserPayload(payload))
+ .addDataSource(INPUT1, dataSource);
+
+ // key -> num keys
+ Vertex aggregationVertex = Vertex
+ .create(AGGREGATION, ProcessorDescriptor
+ .create(AggregationProcessor.class.getName()).setUserPayload(payload), 1);
+
+ DataSinkDescriptor dataSink = MROutput
+ .createConfigBuilder(new Configuration(tezConf), TextOutputFormat.class,
+ OUT_PATH.toString())
+ .build();
+ // Broadcast Hash Join
+ Vertex mapJoinVertex = Vertex
+ .create(MAP_JOIN, ProcessorDescriptor.create(MapJoinProcessor.class.getName())
+ .setUserPayload(payload))
+ .addDataSource(INPUT2, dataSource)
+ .addDataSink(OUTPUT, dataSink);
+
+ EdgeProperty orderedEdge = OrderedPartitionedKVEdgeConfig
+ .newBuilder(Text.class.getName(), IntWritable.class.getName(), HashPartitioner.class.getName())
+ .setFromConfiguration(tezConf)
+ .build()
+ .createDefaultEdgeProperty();
+ EdgeProperty broadcastEdge = UnorderedKVEdgeConfig
+ .newBuilder(Text.class.getName(), IntWritable.class.getName())
+ .setFromConfiguration(tezConf)
+ .build()
+ .createDefaultBroadcastEdgeProperty();
+
+ DAG dag = DAG.create("TestAMRecoveryAggregationBroadcast_" + dagName);
+ dag.addVertex(tableScanVertex)
+ .addVertex(aggregationVertex)
+ .addVertex(mapJoinVertex)
+ .addEdge(Edge.create(tableScanVertex, aggregationVertex, orderedEdge))
+ .addEdge(Edge.create(aggregationVertex, mapJoinVertex, broadcastEdge));
+ return dag;
+ }
+
+ TezCounters runDAGAndVerify(DAG dag, boolean killAM) throws Exception {
+ tezSession.waitTillReady();
+ DAGClient dagClient = tezSession.submitDAG(dag);
+
+ if (killAM) {
+ TimeUnit.SECONDS.sleep(10);
+ YarnClient yarnClient = YarnClient.createYarnClient();
+ yarnClient.init(tezConf);
+ yarnClient.start();
+ ApplicationAttemptId id = ApplicationAttemptId.newInstance(
+ tezSession.getAppMasterApplicationId(), 1);
Review Comment:
Done
https://github.com/apache/tez/pull/361/commits/7f0b3b8554ab26d481c10f665ba5ee7881135e68
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "okumin (via GitHub)" <gi...@apache.org>.
okumin commented on code in PR #361:
URL: https://github.com/apache/tez/pull/361#discussion_r1895638369
##########
tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java:
##########
@@ -2817,6 +2817,18 @@ private VertexState setupVertex() {
return VertexState.INITED;
}
+ private boolean canSkipInitializingParents() {
+ // Both cases use RootInputVertexManager. RootInputVertexManager can start tasks even though
+ // any parents are not fully initialized.
+ if (vertexPlan.hasVertexManagerPlugin()) {
+ final VertexManagerPluginDescriptor pluginDesc = DagTypeConverters
+ .convertVertexManagerPluginDescriptorFromDAGPlan(vertexPlan.getVertexManagerPlugin());
+ return pluginDesc.getClassName().equals(RootInputVertexManager.class.getName());
+ } else {
+ return inputsWithInitializers != null;
Review Comment:
Done.
https://github.com/apache/tez/pull/361/commits/c8c18fcb49de1d0cadd368ed8db73750f9d154f7
I kept this branch intentionally because there might be a case where a vertex manager plugin is explicitly configured but it is not a RootInputVertexManager.
```
if (vertexPlan.hasVertexManagerPlugin()) {
final VertexManagerPluginDescriptor pluginDesc = DagTypeConverters
.convertVertexManagerPluginDescriptorFromDAGPlan(vertexPlan.getVertexManagerPlugin());
return pluginDesc.getClassName().equals(RootInputVertexManager.class.getName());
} else {
// This case implicitly uses RootInputVertexManager. See VertexImpl#assignVertexManager
return hasInputInitializers();
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "okumin (via GitHub)" <gi...@apache.org>.
okumin commented on PR #361:
URL: https://github.com/apache/tez/pull/361#issuecomment-2559503795
I rebased this branch and also added two cosmetic changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "abstractdog (via GitHub)" <gi...@apache.org>.
abstractdog commented on PR #361:
URL: https://github.com/apache/tez/pull/361#issuecomment-2559794345
> @abstractdog Sorry, my hand-made refactoring included a mistake. I copy-pasted the method names used in the original condition. I appreciate it if you could double-check it. [bdece70](https://github.com/apache/tez/commit/bdece70288d20463b131d852511128103f093cdf)
missed it, I saw it's fixed, glad to see that unit tests revealed the problem
+1 still holds if tests will pass
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-4569: SCATTER_GATHER + BROADCAST hangs on DAG Recovery [tez]
Posted by "tez-yetus (via GitHub)" <gi...@apache.org>.
tez-yetus commented on PR #361:
URL: https://github.com/apache/tez/pull/361#issuecomment-2559919940
:confetti_ball: **+1 overall**
| Vote | Subsystem | Runtime | Comment |
|:----:|----------:|--------:|:--------|
| +0 :ok: | reexec | 0m 15s | Docker mode activated. |
||| _ Prechecks _ |
| +1 :green_heart: | dupname | 0m 0s | No case conflicting files found. |
| +1 :green_heart: | @author | 0m 0s | The patch does not contain any @author tags. |
| +1 :green_heart: | test4tests | 0m 0s | The patch appears to include 1 new or modified test files. |
||| _ master Compile Tests _ |
| +0 :ok: | mvndep | 2m 48s | Maven dependency ordering for branch |
| +1 :green_heart: | mvninstall | 13m 59s | master passed |
| +1 :green_heart: | compile | 1m 19s | master passed with JDK Ubuntu-11.0.25+9-post-Ubuntu-1ubuntu122.04 |
| +1 :green_heart: | compile | 1m 11s | master passed with JDK Private Build-1.8.0_432-8u432-ga~us1-0ubuntu2~22.04-ga |
| +1 :green_heart: | checkstyle | 1m 31s | master passed |
| +1 :green_heart: | javadoc | 0m 57s | master passed with JDK Ubuntu-11.0.25+9-post-Ubuntu-1ubuntu122.04 |
| +1 :green_heart: | javadoc | 0m 42s | master passed with JDK Private Build-1.8.0_432-8u432-ga~us1-0ubuntu2~22.04-ga |
| +0 :ok: | spotbugs | 0m 50s | Used deprecated FindBugs config; considering switching to SpotBugs. |
| +1 :green_heart: | findbugs | 2m 37s | master passed |
||| _ Patch Compile Tests _ |
| +0 :ok: | mvndep | 0m 10s | Maven dependency ordering for patch |
| +1 :green_heart: | mvninstall | 0m 48s | the patch passed |
| +1 :green_heart: | compile | 0m 51s | the patch passed with JDK Ubuntu-11.0.25+9-post-Ubuntu-1ubuntu122.04 |
| +1 :green_heart: | javac | 0m 51s | the patch passed |
| +1 :green_heart: | compile | 0m 44s | the patch passed with JDK Private Build-1.8.0_432-8u432-ga~us1-0ubuntu2~22.04-ga |
| +1 :green_heart: | javac | 0m 44s | the patch passed |
| -0 :warning: | checkstyle | 0m 12s | tez-tests: The patch generated 1 new + 0 unchanged - 0 fixed = 1 total (was 0) |
| +1 :green_heart: | whitespace | 0m 0s | The patch has no whitespace issues. |
| +1 :green_heart: | javadoc | 0m 19s | the patch passed with JDK Ubuntu-11.0.25+9-post-Ubuntu-1ubuntu122.04 |
| +1 :green_heart: | javadoc | 0m 19s | the patch passed with JDK Private Build-1.8.0_432-8u432-ga~us1-0ubuntu2~22.04-ga |
| +1 :green_heart: | findbugs | 2m 0s | the patch passed |
||| _ Other Tests _ |
| +1 :green_heart: | unit | 5m 15s | tez-dag in the patch passed. |
| +1 :green_heart: | unit | 46m 32s | tez-tests in the patch passed. |
| +1 :green_heart: | asflicense | 0m 22s | The patch does not generate ASF License warnings. |
| | | 84m 40s | |
| Subsystem | Report/Notes |
|----------:|:-------------|
| Docker | ClientAPI=1.47 ServerAPI=1.47 base: https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-361/5/artifact/out/Dockerfile |
| GITHUB PR | https://github.com/apache/tez/pull/361 |
| JIRA Issue | TEZ-4569 |
| Optional Tests | dupname asflicense javac javadoc unit spotbugs findbugs checkstyle compile |
| uname | Linux 7e9468acddb5 5.15.0-124-generic #134-Ubuntu SMP Fri Sep 27 20:20:17 UTC 2024 x86_64 x86_64 x86_64 GNU/Linux |
| Build tool | maven |
| Personality | personality/tez.sh |
| git revision | master / 9efa6f14d |
| Default Java | Private Build-1.8.0_432-8u432-ga~us1-0ubuntu2~22.04-ga |
| Multi-JDK versions | /usr/lib/jvm/java-11-openjdk-amd64:Ubuntu-11.0.25+9-post-Ubuntu-1ubuntu122.04 /usr/lib/jvm/java-8-openjdk-amd64:Private Build-1.8.0_432-8u432-ga~us1-0ubuntu2~22.04-ga |
| checkstyle | https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-361/5/artifact/out/diff-checkstyle-tez-tests.txt |
| Test Results | https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-361/5/testReport/ |
| Max. process+thread count | 1187 (vs. ulimit of 5500) |
| modules | C: tez-dag tez-tests U: . |
| Console output | https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-361/5/console |
| versions | git=2.34.1 maven=3.6.3 findbugs=3.0.1 |
| Powered by | Apache Yetus 0.12.0 https://yetus.apache.org |
This message was automatically generated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org