KYLIN-3597 Close resources after they are used.

apache · Oct 9, 2018 · ee93848 · ee93848
1 parent 46c9f55
commit ee93848
Show file tree

Hide file tree

Showing 2 changed files with 53 additions and 50 deletions.
diff --git a/core-common/src/main/java/org/apache/kylin/common/persistence/JDBCResourceDAO.java b/core-common/src/main/java/org/apache/kylin/common/persistence/JDBCResourceDAO.java
@@ -442,9 +442,11 @@ public void execute(Connection connection) throws SQLException {
             }
 
             private boolean checkTableExists(final String tableName, final Connection connection) throws SQLException {
-                final PreparedStatement ps = connection.prepareStatement(getCheckTableExistsSql(tableName));
-                final ResultSet rs = ps.executeQuery();
+                PreparedStatement ps = null;
+                ResultSet rs = null;
                 try {
+                    ps = connection.prepareStatement(getCheckTableExistsSql(tableName));
+                    rs = ps.executeQuery();
                     while (rs.next()) {
                         if (tableName.equals(rs.getString(1))) {
                             return true;

diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/SparkCubeHFile.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/SparkCubeHFile.java
@@ -180,58 +180,59 @@ protected void execute(OptionsHelper optionsHelper) throws Exception {
 
             //HBase conf
             logger.info("Loading HBase configuration from:{}", hbaseConfFile);
-            FSDataInputStream confInput = fs.open(new Path(hbaseConfFile));
-
-            Configuration hbaseJobConf = new Configuration();
-            hbaseJobConf.addResource(confInput);
-            hbaseJobConf.set("spark.hadoop.dfs.replication", "3"); // HFile, replication=3
-            Job job = Job.getInstance(hbaseJobConf, cubeSegment.getStorageLocationIdentifier());
-
-            FileOutputFormat.setOutputPath(job, new Path(outputPath));
-
-            JavaPairRDD<Text, Text> inputRDDs = SparkUtil.parseInputPath(inputPath, fs, sc, Text.class, Text.class);
-            final JavaPairRDD<RowKeyWritable, KeyValue> hfilerdd;
-            if (quickPath) {
-                hfilerdd = inputRDDs.mapToPair(new PairFunction<Tuple2<Text, Text>, RowKeyWritable, KeyValue>() {
-                    @Override
-                    public Tuple2<RowKeyWritable, KeyValue> call(Tuple2<Text, Text> textTextTuple2) throws Exception {
-                        KeyValue outputValue = keyValueCreators.get(0).create(textTextTuple2._1,
-                                textTextTuple2._2.getBytes(), 0, textTextTuple2._2.getLength());
-                        return new Tuple2<>(new RowKeyWritable(outputValue.createKeyOnly(false).getKey()), outputValue);
-                    }
-                });
-            } else {
-                hfilerdd = inputRDDs.flatMapToPair(new PairFlatMapFunction<Tuple2<Text, Text>, RowKeyWritable, KeyValue>() {
-                    @Override
-                    public Iterator<Tuple2<RowKeyWritable, KeyValue>> call(Tuple2<Text, Text> textTextTuple2)
-                            throws Exception {
-
-                        List<Tuple2<RowKeyWritable, KeyValue>> result = Lists.newArrayListWithExpectedSize(cfNum);
-                        Object[] inputMeasures = new Object[cubeDesc.getMeasures().size()];
-                        inputCodec.decode(ByteBuffer.wrap(textTextTuple2._2.getBytes(), 0, textTextTuple2._2.getLength()),
-                                inputMeasures);
-
-                        for (int i = 0; i < cfNum; i++) {
-                            KeyValue outputValue = keyValueCreators.get(i).create(textTextTuple2._1, inputMeasures);
-                            result.add(new Tuple2<>(new RowKeyWritable(outputValue.createKeyOnly(false).getKey()),
-                                    outputValue));
-                        }
 
-                        return result.iterator();
-                    }
-                });
-            }
+            try (FSDataInputStream confInput = fs.open(new Path(hbaseConfFile))) {
+                Configuration hbaseJobConf = new Configuration();
+                hbaseJobConf.addResource(confInput);
+                hbaseJobConf.set("spark.hadoop.dfs.replication", "3"); // HFile, replication=3
+                Job job = Job.getInstance(hbaseJobConf, cubeSegment.getStorageLocationIdentifier());
+
+                FileOutputFormat.setOutputPath(job, new Path(outputPath));
 
-            hfilerdd.repartitionAndSortWithinPartitions(new HFilePartitioner(keys),
-                    RowKeyWritable.RowKeyComparator.INSTANCE)
-                    .mapToPair(new PairFunction<Tuple2<RowKeyWritable, KeyValue>, ImmutableBytesWritable, KeyValue>() {
+                JavaPairRDD<Text, Text> inputRDDs = SparkUtil.parseInputPath(inputPath, fs, sc, Text.class, Text.class);
+                final JavaPairRDD<RowKeyWritable, KeyValue> hfilerdd;
+                if (quickPath) {
+                    hfilerdd = inputRDDs.mapToPair(new PairFunction<Tuple2<Text, Text>, RowKeyWritable, KeyValue>() {
                         @Override
-                        public Tuple2<ImmutableBytesWritable, KeyValue> call(
-                                Tuple2<RowKeyWritable, KeyValue> rowKeyWritableKeyValueTuple2) throws Exception {
-                            return new Tuple2<>(new ImmutableBytesWritable(rowKeyWritableKeyValueTuple2._2.getKey()),
-                                    rowKeyWritableKeyValueTuple2._2);
+                        public Tuple2<RowKeyWritable, KeyValue> call(Tuple2<Text, Text> textTextTuple2) throws Exception {
+                            KeyValue outputValue = keyValueCreators.get(0).create(textTextTuple2._1,
+                                    textTextTuple2._2.getBytes(), 0, textTextTuple2._2.getLength());
+                            return new Tuple2<>(new RowKeyWritable(outputValue.createKeyOnly(false).getKey()), outputValue);
                         }
-                    }).saveAsNewAPIHadoopDataset(job.getConfiguration());
+                    });
+                } else {
+                    hfilerdd = inputRDDs.flatMapToPair(new PairFlatMapFunction<Tuple2<Text, Text>, RowKeyWritable, KeyValue>() {
+                        @Override
+                        public Iterator<Tuple2<RowKeyWritable, KeyValue>> call(Tuple2<Text, Text> textTextTuple2)
+                                throws Exception {
+
+                            List<Tuple2<RowKeyWritable, KeyValue>> result = Lists.newArrayListWithExpectedSize(cfNum);
+                            Object[] inputMeasures = new Object[cubeDesc.getMeasures().size()];
+                            inputCodec.decode(ByteBuffer.wrap(textTextTuple2._2.getBytes(), 0, textTextTuple2._2.getLength()),
+                                    inputMeasures);
+
+                            for (int i = 0; i < cfNum; i++) {
+                                KeyValue outputValue = keyValueCreators.get(i).create(textTextTuple2._1, inputMeasures);
+                                result.add(new Tuple2<>(new RowKeyWritable(outputValue.createKeyOnly(false).getKey()),
+                                        outputValue));
+                            }
+
+                            return result.iterator();
+                        }
+                    });
+                }
+
+                hfilerdd.repartitionAndSortWithinPartitions(new HFilePartitioner(keys),
+                        RowKeyWritable.RowKeyComparator.INSTANCE)
+                        .mapToPair(new PairFunction<Tuple2<RowKeyWritable, KeyValue>, ImmutableBytesWritable, KeyValue>() {
+                            @Override
+                            public Tuple2<ImmutableBytesWritable, KeyValue> call(
+                                    Tuple2<RowKeyWritable, KeyValue> rowKeyWritableKeyValueTuple2) throws Exception {
+                                return new Tuple2<>(new ImmutableBytesWritable(rowKeyWritableKeyValueTuple2._2.getKey()),
+                                        rowKeyWritableKeyValueTuple2._2);
+                            }
+                        }).saveAsNewAPIHadoopDataset(job.getConfiguration());
+            }
 
             logger.info("HDFS: Number of bytes written={}", jobListener.metrics.getBytesWritten());