abstractdog commented on code in PR #520:
URL: https://github.com/apache/tez/pull/520#discussion_r3630878580
##########
tez-tests/src/test/java/org/apache/tez/test/TestAMRecoveryAggregationBroadcast.java:
##########
@@ -242,7 +248,9 @@ public void testTableScanTemporalFailure() throws Exception
{
public void testAggregationTemporalFailure() throws Exception {
tezConf.setBoolean(AGGREGATION_SLEEP, true);
DAG dag = createDAG("AggregationTemporalFailure");
- TezCounters counters = runDAGAndVerify(dag, true);
+ // Wait for TableScan to be SUCCEEDED before killing the
+ TezCounters counters = runDAGAndVerify(dag, true,
+ Collections.singletonList(TABLE_SCAN));
Review Comment:
line break is not needed here
##########
tez-tests/src/test/java/org/apache/tez/test/TestAMRecoveryAggregationBroadcast.java:
##########
@@ -263,7 +271,9 @@ public void testAggregationTemporalFailure() throws
Exception {
public void testMapJoinTemporalFailure() throws Exception {
tezConf.setBoolean(MAP_JOIN_SLEEP, true);
DAG dag = createDAG("MapJoinTemporalFailure");
- TezCounters counters = runDAGAndVerify(dag, true);
+ // Wait for both upstream vertices to be SUCCEEDED before killing the AM
+ TezCounters counters = runDAGAndVerify(dag, true,
+ Arrays.asList(TABLE_SCAN, AGGREGATION));
Review Comment:
line break is not needed here
##########
tez-tests/src/test/java/org/apache/tez/test/TestAMRecoveryAggregationBroadcast.java:
##########
@@ -356,14 +374,40 @@ TezCounters runDAGAndVerify(DAG dag, boolean killAM)
throws Exception {
LOG.info("Diagnosis: " + dagStatus.getDiagnostics());
assertEquals(State.SUCCEEDED, dagStatus.getState());
- FSDataInputStream in = remoteFs.open(new Path(OUT_PATH,
"part-v002-o000-r-00000"));
+ FSDataInputStream in = remoteFs.open(new Path(outPath,
"part-v002-o000-r-00000"));
ByteBuffer buf = ByteBuffer.allocate(100);
in.read(buf);
buf.flip();
assertEquals(EXPECTED_OUTPUT,
StandardCharsets.UTF_8.decode(buf).toString());
return dagStatus.getDAGCounters();
}
+ private void waitForVertexSucceeded(DAGClient dagClient, String vertexName,
+ long timeoutMs) throws Exception {
+ long deadline = System.currentTimeMillis() + timeoutMs;
+ while (System.currentTimeMillis() < deadline) {
Review Comment:
try to use a monothonic clock instead of `System.currentTimeMillis()`
##########
tez-tests/src/test/java/org/apache/tez/test/TestAMRecoveryAggregationBroadcast.java:
##########
@@ -356,14 +374,40 @@ TezCounters runDAGAndVerify(DAG dag, boolean killAM)
throws Exception {
LOG.info("Diagnosis: " + dagStatus.getDiagnostics());
assertEquals(State.SUCCEEDED, dagStatus.getState());
- FSDataInputStream in = remoteFs.open(new Path(OUT_PATH,
"part-v002-o000-r-00000"));
+ FSDataInputStream in = remoteFs.open(new Path(outPath,
"part-v002-o000-r-00000"));
ByteBuffer buf = ByteBuffer.allocate(100);
in.read(buf);
buf.flip();
assertEquals(EXPECTED_OUTPUT,
StandardCharsets.UTF_8.decode(buf).toString());
return dagStatus.getDAGCounters();
}
+ private void waitForVertexSucceeded(DAGClient dagClient, String vertexName,
+ long timeoutMs) throws Exception {
+ long deadline = System.currentTimeMillis() + timeoutMs;
+ while (System.currentTimeMillis() < deadline) {
+ // Before the vertex is initialized on the AM, getVertexStatus may
+ // return null - treat that the same as NEW / INITIALIZING and keep
+ // polling.
+ VertexStatus status = dagClient.getVertexStatus(vertexName, null);
+ if (status != null) {
+ VertexStatus.State state = status.getState();
+ if (state == VertexStatus.State.SUCCEEDED) {
+ return;
+ }
+ if (state == VertexStatus.State.FAILED
+ || state == VertexStatus.State.KILLED
+ || state == VertexStatus.State.ERROR) {
+ throw new AssertionError("Vertex " + vertexName
+ + " reached terminal non-success state: " + state);
+ }
+ }
Review Comment:
what about a switch case here, something like
```
switch (lastState) {
case SUCCEEDED -> { return; }
case FAILED, KILLED, ERROR -> throw new AssertionError("Vertex " +
vertexName
+ " reached terminal non-success state: " + lastState);
...
}
```
##########
tez-tests/src/test/java/org/apache/tez/test/TestAMRecoveryAggregationBroadcast.java:
##########
@@ -339,12 +349,20 @@ private DAG createDAG(String dagName) throws Exception {
return dag;
}
- TezCounters runDAGAndVerify(DAG dag, boolean killAM) throws Exception {
+ TezCounters runDAGAndVerify(DAG dag, boolean killAM,
+ List<String> vertexNamesToWaitFor) throws Exception {
tezSession.waitTillReady();
DAGClient dagClient = tezSession.submitDAG(dag);
if (killAM) {
- TimeUnit.SECONDS.sleep(10);
+ // Deterministic wait: block until every named upstream vertex reaches
+ // SUCCEEDED. Replaces a fixed Thread.sleep(10s) which was too short on
+ // slow CI machines and caused the recovery-log assertions below to
+ // fail intermittently.
+ for (String vertexName : vertexNamesToWaitFor) {
+ waitForVertexSucceeded(dagClient, vertexName,
+ TimeUnit.SECONDS.toMillis(60));
Review Comment:
line break is not needed here
##########
tez-api/src/main/java/org/apache/tez/client/TezClientUtils.java:
##########
@@ -961,7 +961,26 @@ static DAGClientAMProtocolBlockingPB
getAMProxy(FrameworkClient frameworkClient,
throw new TezException(e);
}
- return getAMProxy(conf, appReport.getHost(), appReport.getRpcPort(),
+ // YARN-808 gap: when the AM container is first allocated YARN briefly
+ // reports state=RUNNING before the AM has registered with the RM or bound
+ // its RPC listener. During that window the ApplicationReport contains
+ // sentinel values that must not be passed to
+ // NetUtils.createSocketAddrForHost() (which would throw
+ // IllegalArgumentException: port out of range) or to RPC.getProxy():
+ // host == null / "N/A" : RM has not received registerApplicationMaster()
+ // rpcPort == 0 : protobuf wire default
+ // rpcPort == -1 : container up but RPC listener not yet bound
+ // Returning null lets callers (waitForProxy, sendAMHeartbeat) back off
+ // and retry rather than crash.
Review Comment:
this long comment can be put to a new method that contains the refactored
condition below
except the "return null lets callers" part, which belongs to this method
##########
tez-api/src/main/java/org/apache/tez/dag/api/client/rpc/DAGClientRPCImpl.java:
##########
@@ -280,10 +280,14 @@ boolean createAMProxyIfNeeded() throws IOException,
TezException,
}
// YARN-808. Cannot ascertain if AM is ready until we connect to it.
- // workaround check the default string set by YARN
- if(appReport.getHost() == null || appReport.getHost().equals("N/A") ||
- appReport.getRpcPort() == 0){
- // attempt not running
+ // Workaround: check the default strings/sentinels set by YARN.
+ // port == 0 : protobuf wire default, AM has not called
+ // registerApplicationMaster() yet.
+ // port == -1 : AM container is allocated (state=RUNNING) but the
+ // RPC listener has not been bound yet.
Review Comment:
this comment can be put to the new method that contains the refactored
condition below
##########
tez-api/src/test/java/org/apache/tez/dag/api/client/rpc/TestDAGClient.java:
##########
@@ -725,6 +726,55 @@ public void testGetDagStatusWithCachedStatusExpiration()
throws Exception {
}
}
+ /**
+ * Covers the YARN-808 guard in DAGClientRPCImpl#createAMProxyIfNeeded
+ */
+ @Test
+ @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
+ public void testCreateAMProxyIfNeededReturnsFalseOnBadEndpoint() throws
Exception {
+ TezConfiguration tezConf = new TezConfiguration();
+
+ // Case: rpcPort == 0 (protobuf default sentinel).
+ assertFalse(mockReportClient(tezConf, "somehost",
0).createAMProxyIfNeeded(),
+ "rpcPort == 0 should return false");
+
+ // Case: rpcPort == -1 (YARN-808 gap — AM allocated but RPC not bound).
+ assertFalse(mockReportClient(tezConf, "somehost",
-1).createAMProxyIfNeeded(),
+ "rpcPort == -1 should return false");
+
+ // Case: host == null.
+ assertFalse(mockReportClient(tezConf, null, 8080).createAMProxyIfNeeded(),
+ "host == null should return false");
+
+ // Case: host == "N/A".
+ assertFalse(mockReportClient(tezConf, "N/A", 8080).createAMProxyIfNeeded(),
+ "host == N/A should return false");
+ }
+
+ private DAGClientRPCImplWithFakeReport mockReportClient(TezConfiguration
conf,
+ String host, int rpcPort) throws IOException {
+ ApplicationReport report = mock(ApplicationReport.class);
+
when(report.getYarnApplicationState()).thenReturn(YarnApplicationState.RUNNING);
+ when(report.getHost()).thenReturn(host);
+ when(report.getRpcPort()).thenReturn(rpcPort);
+ return new DAGClientRPCImplWithFakeReport(mockAppId, dagIdStr, conf,
report);
+ }
Review Comment:
it's strange that mocking and real classes are mixed: if there is a real
`DAGClientRPCImplWithFakeReport`, it could leverage a real `ApplicationReport`,
couldn't it?
##########
tez-api/src/main/java/org/apache/tez/client/TezClientUtils.java:
##########
@@ -961,7 +961,26 @@ static DAGClientAMProtocolBlockingPB
getAMProxy(FrameworkClient frameworkClient,
throw new TezException(e);
}
- return getAMProxy(conf, appReport.getHost(), appReport.getRpcPort(),
+ // YARN-808 gap: when the AM container is first allocated YARN briefly
+ // reports state=RUNNING before the AM has registered with the RM or bound
+ // its RPC listener. During that window the ApplicationReport contains
+ // sentinel values that must not be passed to
+ // NetUtils.createSocketAddrForHost() (which would throw
+ // IllegalArgumentException: port out of range) or to RPC.getProxy():
+ // host == null / "N/A" : RM has not received registerApplicationMaster()
+ // rpcPort == 0 : protobuf wire default
+ // rpcPort == -1 : container up but RPC listener not yet bound
+ // Returning null lets callers (waitForProxy, sendAMHeartbeat) back off
+ // and retry rather than crash.
+ String amHost = appReport.getHost();
+ int amRpcPort = appReport.getRpcPort();
+ if (amHost == null || amHost.equals("N/A") || amRpcPort <= 0) {
Review Comment:
I think this condition can be refactored to a separate method with
`appReport` parameter and be reused across different codepaths
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
