Adds an option to fetch recently finished apps from RM (linkedin#212)

app-conf/GeneralConf.xml

@@ -37,4 +37,12 @@
     <value>true</value>
     <description>If this property is "false", search will only make exact matches</description>
   </property>
+  <!--
+  Initial window in MS to indicate how much older apps to fetch from RM.
+  -->
+  <!--
+  <property>
+    <name>drelephant.analysis.fetch.initial.windowMillis</name>
+    <value>3600000</value>
+  </property> -->
 </configuration>

app/com/linkedin/drelephant/analysis/AnalyticJobGeneratorHadoop2.java

@@ -46,6 +46,8 @@ public class AnalyticJobGeneratorHadoop2 implements AnalyticJobGenerator {
   private static final String IS_RM_HA_ENABLED = "yarn.resourcemanager.ha.enabled";
   private static final String RESOURCE_MANAGER_IDS = "yarn.resourcemanager.ha.rm-ids";
   private static final String RM_NODE_STATE_URL = "http://%s/ws/v1/cluster/info";
+  private static final String FETCH_INITIAL_WINDOW_MS = "drelephant.analysis.fetch.initial.windowMillis";
+
   private static Configuration configuration;
 
   // We provide one minute job fetch delay due to the job sending lag from AM/NM to JobHistoryServer HDFS
@@ -58,6 +60,7 @@ public class AnalyticJobGeneratorHadoop2 implements AnalyticJobGenerator {
 
   private String _resourceManagerAddress;
   private long _lastTime = 0;
+  private long _fetchStartTime = 0;
   private long _currentTime = 0;
   private long _tokenUpdatedTime = 0;
   private AuthenticatedURL.Token _token;
@@ -109,6 +112,12 @@ public void updateResourceManagerAddresses() {
   public void configure(Configuration configuration)
       throws IOException {
     this.configuration = configuration;
+    String initialFetchWindowString = configuration.get(FETCH_INITIAL_WINDOW_MS);
+    if (initialFetchWindowString != null) {
+      long initialFetchWindow = Long.getLong(initialFetchWindowString);
+      _lastTime = System.currentTimeMillis() - FETCH_DELAY - initialFetchWindow;
+      _fetchStartTime = _lastTime;
+    }
     updateResourceManagerAddresses();
   }
 
@@ -212,7 +221,7 @@ private List<AnalyticJob> readApps(URL url) throws IOException, AuthenticationEx
 
       // When called first time after launch, hit the DB and avoid duplicated analytic jobs that have been analyzed
       // before.
-      if (_lastTime > 0 || (_lastTime == 0 && AppResult.find.byId(appId) == null)) {
+      if (_lastTime > _fetchStartTime || (_lastTime == _fetchStartTime && AppResult.find.byId(appId) == null)) {
         String user = app.get("user").getValueAsText();
         String name = app.get("name").getValueAsText();
         String queueName = app.get("queue").getValueAsText();