[GitHub] [tez] abstractdog commented on a change in pull request #161: TEZ-4349. DAGClient gets stuck with invalid cached DAGStatus

[GitBox <gi...@apache.org>]

abstractdog commented on a change in pull request #161:
URL: https://github.com/apache/tez/pull/161#discussion_r748856603



##########
File path: tez-api/src/test/java/org/apache/tez/dag/api/client/rpc/TestDAGClient.java
##########
@@ -578,4 +604,110 @@ public void testTimelineClientCleanup() throws Exception {
     }
     Assert.assertFalse("Reloader is still alive", reloaderStillAlive);
   }
+
+  @Test(timeout = 50000)
+  public void testGetDagStatusWithCachedStatusExpiration() throws Exception {
+    long startTime;
+    long endTime;
+    long diff;
+
+    TezConfiguration tezConf = new TezConfiguration();
+    tezConf.setLong(TezConfiguration.TEZ_DAG_STATUS_POLLINTERVAL_MS, 800l);
+    tezConf.setLong(TezConfiguration.TEZ_CLIENT_DAG_STATUS_CACHE_TIMEOUT_MINUTES, 100000l);
+
+    DAGClientImplForTest dagClient =
+        new DAGClientImplForTest(mockAppId, dagIdStr, tezConf, null);
+    DAGClientRPCImplForTest dagClientRpc =
+        new DAGClientRPCImplForTest(mockAppId, dagIdStr, tezConf, null);
+    dagClient.setRealClient(dagClientRpc);
+
+    DAGStatus dagStatus;
+    DAGStatus rmDagStatus;
+
+    // Fetch from AM. RUNNING
+    rmDagStatus =
+        new DAGStatus(constructDagStatusProto(DAGStatusStateProto.DAG_RUNNING), DagStatusSource.RM);
+    dagClient.setRmDagStatus(rmDagStatus);
+    dagClientRpc.setAMProxy(createMockProxy(DAGStatusStateProto.DAG_RUNNING, -1));
+
+    startTime = System.currentTimeMillis();
+    dagStatus = dagClient.getDAGStatus(EnumSet.noneOf(StatusGetOpts.class), 2000l);
+    endTime = System.currentTimeMillis();
+    diff = endTime - startTime;
+    assertTrue(diff > 1500l && diff < 2500l);
+    // Directly from AM
+    assertEquals(0, dagClient.numGetStatusViaRmInvocations);
+    // Directly from AM - one refresh. One with timeout.
+    assertEquals(2, dagClientRpc.numGetStatusViaAmInvocations);
+    assertEquals(DAGStatus.State.RUNNING, dagStatus.getState());
+
+    // Fetch from AM. Success.
+    dagClient.resetCounters();
+    dagClientRpc.resetCountesr();
+    rmDagStatus =
+        new DAGStatus(constructDagStatusProto(DAGStatusStateProto.DAG_RUNNING), DagStatusSource.RM);
+    dagClient.setRmDagStatus(rmDagStatus);
+    dagClientRpc.setAMProxy(createMockProxy(DAGStatusStateProto.DAG_SUCCEEDED, 1000l));
+
+    startTime = System.currentTimeMillis();
+    dagStatus = dagClient.getDAGStatus(EnumSet.noneOf(StatusGetOpts.class), 2000l);
+    endTime = System.currentTimeMillis();
+    diff = endTime - startTime;
+    assertTrue("diff is " + diff, diff > 500l && diff < 1500l);
+    // Directly from AM
+    assertEquals(0, dagClient.numGetStatusViaRmInvocations);
+    // Directly from AM - previous request cached, so single invocation only.
+    assertEquals(1, dagClientRpc.numGetStatusViaAmInvocations);
+    assertEquals(DAGStatus.State.SUCCEEDED, dagStatus.getState());
+
+    // verify that the cachedDAGStatus is correct
+    DAGStatus cachedDagStatus = dagClient.getCachedDAGStatus();
+    Assert.assertNotNull(cachedDagStatus);
+    Assert.assertTrue(dagStatus == cachedDagStatus);
+
+    // When AM proxy throws an exception, the cachedDAGStatus should be returned
+    dagClient.resetCounters();
+    dagClientRpc.resetCountesr();
+    dagClientRpc.injectAMFault(new IOException("injected Fault"));
+    dagStatus = dagClient.getDAGStatus(EnumSet.noneOf(StatusGetOpts.class));
+    // get the Status from the cache
+    assertEquals(0, dagClient.numGetStatusViaRmInvocations);
+    // Directly from AM - previous request cached, so single invocation only.
+    assertEquals(1, dagClientRpc.numGetStatusViaAmInvocations);
+    assertEquals(DAGStatus.State.SUCCEEDED, dagStatus.getState());
+    Assert.assertTrue(dagStatus == cachedDagStatus);
+
+    // test that RM is invoked when the cacheExpires and the AM fails.
+    dagClientRpc.setAMProxy(createMockProxy(DAGStatusStateProto.DAG_SUCCEEDED, 1000l));
+    dagClientRpc.injectAMFault(new IOException("injected AM Fault"));
+    dagClient.resetCounters();
+    dagClientRpc.resetCountesr();

Review comment:
       could you please fix this typo?
   (I know it's not introduced with this patch, but it's worth fixing as you already working on this area)




-- 
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