Save FileSystem instead of constantly re-creating, ensure CsvReader closed

Author: billoley

balancer/src/main/java/timely/balancer/MetricResolver.java

@@ -91,17 +91,21 @@ public class MetricResolver {
     final protected BalancerProperties balancerProperties;
     final protected CacheProperties cacheProperties;
     final protected HealthChecker healthChecker;
+    final protected Path assignmentFile;
+    final protected FileSystem fs;
 
     private enum BalanceType {
         HIGH_LOW, HIGH_AVG, AVG_LOW
     }
 
     public MetricResolver(CuratorFramework curatorFramework, BalancerProperties balancerProperties, CacheProperties cacheProperties,
-                    HealthChecker healthChecker) {
+                    HealthChecker healthChecker) throws Exception {
         this.curatorFramework = curatorFramework;
         this.balancerProperties = balancerProperties;
         this.cacheProperties = cacheProperties;
         this.healthChecker = healthChecker;
+        this.assignmentFile = new Path(balancerProperties.getAssignmentFile());
+        this.fs = getFileSystem(this.assignmentFile);
     }
 
     public void start() {
@@ -916,20 +920,16 @@ protected FileSystem getFileSystem(Path path) throws IOException {
 
     protected void createAssignmentFile() {
         try {
-            log.info("Creating assignment file: " + balancerProperties.getAssignmentFile());
-            Path assignmentFile = new Path(balancerProperties.getAssignmentFile());
-            FileSystem fs = getFileSystem(assignmentFile);
-            fs.create(assignmentFile);
+            log.info("Creating assignment file: " + this.assignmentFile);
+            this.fs.create(this.assignmentFile);
         } catch (Exception e) {
             log.error(e.getMessage(), e);
         }
     }
 
     protected long getLastWrittenToHdfsTimestamp() {
         try {
-            Path assignmentFile = new Path(balancerProperties.getAssignmentFile());
-            FileSystem fs = getFileSystem(assignmentFile);
-            FileStatus fileStatus = fs.getFileStatus(assignmentFile);
+            FileStatus fileStatus = this.fs.getFileStatus(this.assignmentFile);
             return fileStatus.getModificationTime();
         } catch (FileNotFoundException e) {
             createAssignmentFile();
@@ -960,16 +960,15 @@ private void readAssignmentsFromHdfs(boolean checkIfNecessary) {
         // proceed with reading from HDFS
 
         Map<String,TimelyBalancedHost> assignedMetricToHostMap = new TreeMap<>();
+        CsvReader reader = null;
         try {
             boolean acquired = false;
             while (!acquired) {
                 acquired = assignmentsIPRWLock.readLock().acquire(60, TimeUnit.SECONDS);
             }
             balancerLock.writeLock().lock();
-            Path assignmentFile = new Path(balancerProperties.getAssignmentFile());
-            FileSystem fs = getFileSystem(assignmentFile);
-            FSDataInputStream iStream = fs.open(assignmentFile);
-            CsvReader reader = new CsvReader(iStream, ',', Charset.forName("UTF-8"));
+            FSDataInputStream iStream = this.fs.open(this.assignmentFile);
+            reader = new CsvReader(iStream, ',', Charset.forName("UTF-8"));
             reader.setUseTextQualifier(false);
 
             // skip the headers
@@ -1031,6 +1030,9 @@ private void readAssignmentsFromHdfs(boolean checkIfNecessary) {
         } catch (Exception e) {
             log.error(e.getMessage(), e);
         } finally {
+            if (reader != null) {
+                reader.close();
+            }
             balancerLock.writeLock().unlock();
             try {
                 assignmentsIPRWLock.readLock().release();
@@ -1052,12 +1054,10 @@ private void writeAssignmentsToHdfs() {
             nonCachedMetricsLocalLock.readLock().lock();
             try {
                 if (!metricToHostMap.isEmpty()) {
-                    Path assignmentFile = new Path(balancerProperties.getAssignmentFile());
-                    FileSystem fs = getFileSystem(assignmentFile);
-                    if (!fs.exists(assignmentFile.getParent())) {
-                        fs.mkdirs(assignmentFile.getParent());
+                    if (!this.fs.exists(this.assignmentFile.getParent())) {
+                        this.fs.mkdirs(this.assignmentFile.getParent());
                     }
-                    FSDataOutputStream oStream = fs.create(assignmentFile, true);
+                    FSDataOutputStream oStream = this.fs.create(this.assignmentFile, true);
                     writer = new CsvWriter(oStream, ',', Charset.forName("UTF-8"));
                     writer.setUseTextQualifier(false);
                     writer.write("metric");
@@ -1080,7 +1080,7 @@ private void writeAssignmentsToHdfs() {
                     if (!assignmentsLastUpdatedInHdfs.get().succeeded()) {
                         assignmentsLastUpdatedInHdfs.forceSet(now);
                     }
-                    log.debug("Wrote {} assignments to hdfs lastHdfsUpdate = lastLocalUpdate ({})", metricToHostMap.size(),
+                    log.info("Wrote {} assignments to hdfs lastHdfsUpdate = lastLocalUpdate ({})", metricToHostMap.size(),
                                     new Date(assignmentsLastUpdatedLocal.get()));
                     // remove metric from metricMap (ArrivalRate) if not being cached
                     metricMap.entrySet().removeIf(e -> !metricToHostMap.containsKey(e.getKey()));

balancer/src/main/java/timely/balancer/configuration/BalancerConfiguration.java

@@ -50,7 +50,7 @@ public HealthChecker healthChecker(BalancerProperties balancerProperties, Balanc
     @Bean
     @ConditionalOnMissingBean
     public MetricResolver metricResolver(CuratorFramework curatorFramework, BalancerProperties balancerProperties, CacheProperties cacheProperties,
-                    HealthChecker healthChecker) {
+                    HealthChecker healthChecker) throws Exception {
         MetricResolver metricResolver = new MetricResolver(curatorFramework, balancerProperties, cacheProperties, healthChecker);
         metricResolver.start();
         return metricResolver;

balancer/src/test/java/timely/balancer/configuration/TestBalancerConfiguration.java

@@ -48,7 +48,7 @@ public TestHealthChecker healthChecker(BalancerProperties balancerProperties, Ba
 
     @Bean
     public TestMetricResolver metricResolver(CuratorFramework curatorFramework, BalancerProperties balancerProperties, CacheProperties cacheProperties,
-                    HealthChecker healthChecker) {
+                    HealthChecker healthChecker) throws Exception {
         TestMetricResolver metricResolver = new TestMetricResolver(curatorFramework, balancerProperties, cacheProperties, healthChecker);
         metricResolver.start();
         return metricResolver;

balancer/src/test/java/timely/balancer/test/TestMetricResolver.java

@@ -20,7 +20,7 @@ public class TestMetricResolver extends MetricResolver {
     private static final Logger log = LoggerFactory.getLogger(TestMetricResolver.class);
 
     public TestMetricResolver(CuratorFramework curatorFramework, BalancerProperties balancerProperties, CacheProperties cacheProperties,
-                    HealthChecker healthChecker) {
+                    HealthChecker healthChecker) throws Exception {
         super(curatorFramework, balancerProperties, cacheProperties, healthChecker);
     }