You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by be...@apache.org on 2023/08/27 08:33:10 UTC
[incubator-streampark] branch upload created (now ea968bb28)
This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a change to branch upload
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
at ea968bb28 [Improve] FileUtils minor improvement
This branch includes the following new commits:
new ea968bb28 [Improve] FileUtils minor improvement
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[incubator-streampark] 01/01: [Improve] FileUtils minor improvement
Posted by be...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a commit to branch upload
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
commit ea968bb28873b33cca7714dc6746cd2fbb880fca
Author: benjobs <be...@apache.org>
AuthorDate: Sun Aug 27 16:32:52 2023 +0800
[Improve] FileUtils minor improvement
---
.../apache/streampark/common/util/FileUtils.scala | 138 ++++++++++++++++++---
.../streampark/common/util/HadoopUtils.scala | 5 +-
.../org/apache/streampark/common/util/Logger.scala | 2 +-
.../streampark/console/base/util/FileUtils.java | 107 ----------------
.../core/service/impl/AppBuildPipeServiceImpl.java | 2 +-
.../core/service/impl/ApplicationServiceImpl.java | 14 +--
.../core/service/impl/ProjectServiceImpl.java | 2 +-
.../console/base/util/FileUtilsTest.java | 2 +
.../flink/core/FlinkStreamingInitializer.scala | 2 +-
9 files changed, 134 insertions(+), 140 deletions(-)
diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/util/FileUtils.scala b/streampark-common/src/main/scala/org/apache/streampark/common/util/FileUtils.scala
index 719715e22..40dd1d120 100644
--- a/streampark-common/src/main/scala/org/apache/streampark/common/util/FileUtils.scala
+++ b/streampark-common/src/main/scala/org/apache/streampark/common/util/FileUtils.scala
@@ -18,13 +18,17 @@ package org.apache.streampark.common.util
import java.io._
import java.net.URL
+import java.nio.ByteBuffer
+import java.nio.channels.Channels
+import java.nio.charset.StandardCharsets
+import java.nio.file.{Files, Paths}
import java.util
-import java.util.Scanner
+import java.util.stream.Collectors
import scala.collection.convert.ImplicitConversions._
import scala.collection.mutable
-object FileUtils {
+object FileUtils extends org.apache.commons.io.FileUtils {
private[this] def bytesToHexString(src: Array[Byte]): String = {
val stringBuilder = new mutable.StringBuilder
@@ -73,12 +77,6 @@ object FileUtils {
s"[StreamPark] Failed to create directory within $TEMP_DIR_ATTEMPTS attempts (tried $baseName 0 to $baseName ${TEMP_DIR_ATTEMPTS - 1})")
}
- def exists(path: String): Unit = {
- require(
- path != null && path.nonEmpty && new File(path).exists(),
- s"[StreamPark] FileUtils.exists: file $path is not exist!")
- }
-
def getPathFromEnv(env: String): String = {
val path = Option(System.getenv(env)).getOrElse(System.getProperty(env))
require(
@@ -115,6 +113,14 @@ object FileUtils {
}
}
+ def exists(file: Serializable): Boolean = {
+ file match {
+ case null => false
+ case f: java.io.File => f.exists()
+ case p => new java.io.File(p.toString).exists()
+ }
+ }
+
def equals(file1: File, file2: File): Boolean = {
(file1, file2) match {
case (a, b) if a == null || b == null => false
@@ -143,20 +149,112 @@ object FileUtils {
}
@throws[IOException]
- def readString(file: File): String = {
- require(file != null && file.isFile)
- val reader = new FileReader(file)
- val scanner = new Scanner(reader)
- val buffer = new mutable.StringBuilder()
- if (scanner.hasNextLine) {
- buffer.append(scanner.nextLine())
+ def readInputStream(in: InputStream, array: Array[Byte]): Unit = {
+ var toRead = array.length
+ var ret = 0
+ var off = 0
+ while (toRead > 0) {
+ ret = in.read(array, off, toRead)
+ if (ret < 0) throw new IOException("Bad inputStream, premature EOF")
+ toRead -= ret
+ off += ret
+ }
+ in.close()
+ }
+
+ @throws[IOException]
+ def readFile(file: File): String = {
+ if (file.length >= Int.MaxValue) {
+ throw new IOException("Too large file, unexpected!")
+ } else {
+ val len = file.length
+ val array = new Array[Byte](len.toInt)
+ val is = Files.newInputStream(file.toPath)
+ readInputStream(is, array)
+ is.close()
+ new String(array, StandardCharsets.UTF_8)
+ }
+ }
+
+ @throws[IOException]
+ def writeFile(content: String, file: File): Unit = {
+ val outputStream = Files.newOutputStream(file.toPath)
+ val channel = Channels.newChannel(outputStream)
+ val buffer = ByteBuffer.wrap(content.getBytes(StandardCharsets.UTF_8))
+ channel.write(buffer)
+ channel.close()
+ outputStream.flush()
+ outputStream.close()
+ }
+
+ @throws[IOException]
+ def readEndOfFile(file: File, maxSize: Long): Array[Byte] = {
+ var readSize = maxSize
+ var fileContent: Array[Byte] = null
+ try {
+ val raFile = new RandomAccessFile(file, "r")
+ try {
+ if (raFile.length > maxSize) raFile.seek(raFile.length - maxSize)
+ else if (raFile.length < maxSize) readSize = raFile.length.toInt
+ fileContent = new Array[Byte](readSize.toInt)
+ raFile.read(fileContent)
+ } finally if (raFile != null) raFile.close()
}
- while (scanner.hasNextLine) {
- buffer.append("\r\n")
- buffer.append(scanner.nextLine())
+ fileContent
+ }
+
+ /**
+ * Read the content of a file from a specified offset.
+ *
+ * @param file
+ * The file to read from
+ * @param startOffset
+ * The offset from where to start reading the file
+ * @param maxSize
+ * The maximum size of the file to read
+ * @return
+ * The content of the file as a byte array
+ * @throws IOException
+ * if an I/O error occurs while reading the file
+ * @throws IllegalArgumentException
+ * if the startOffset is greater than the file length
+ */
+ @throws[IOException]
+ def readFileFromOffset(file: File, startOffset: Long, maxSize: Long): Array[Byte] = {
+ if (file.length < startOffset) {
+ throw new IllegalArgumentException(
+ s"The startOffset $startOffset is great than the file length ${file.length}")
+ }
+ Utils.using(new RandomAccessFile(file, "r")) {
+ raFile =>
+ val readSize = Math.min(maxSize, file.length - startOffset)
+ raFile.seek(startOffset)
+ val fileContent = new Array[Byte](readSize.toInt)
+ raFile.read(fileContent)
+ fileContent
+ }
+ }
+
+ /**
+ * Roll View Log.
+ *
+ * @param path
+ * The file path.
+ * @param offset
+ * The offset.
+ * @param limit
+ * The limit.
+ * @return
+ * The content of the file.
+ */
+ def tailOf(path: String, offset: Int, limit: Int): String = try {
+ val file = new File(path)
+ if (file.exists && file.isFile) {
+ Utils.using(Files.lines(Paths.get(path))) {
+ stream => stream.skip(offset).limit(limit).collect(Collectors.joining("\r\n"))
+ }
}
- Utils.close(scanner, reader)
- buffer.toString()
+ null
}
}
diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/util/HadoopUtils.scala b/streampark-common/src/main/scala/org/apache/streampark/common/util/HadoopUtils.scala
index b9823d8f0..715afa00d 100644
--- a/streampark-common/src/main/scala/org/apache/streampark/common/util/HadoopUtils.scala
+++ b/streampark-common/src/main/scala/org/apache/streampark/common/util/HadoopUtils.scala
@@ -137,7 +137,10 @@ object HadoopUtils extends Logger {
def getConfigurationFromHadoopConfDir(confDir: String = hadoopConfDir): Configuration = {
if (!configurationCache.containsKey(confDir)) {
- FileUtils.exists(confDir)
+ if (!FileUtils.exists(confDir)) {
+ throw new ExceptionInInitializerError(
+ s"[StreamPark] hadoop conf file " + confDir + " is not exist!")
+ }
val hadoopConfDir = new File(confDir)
val confName = List("hdfs-default.xml", "core-site.xml", "hdfs-site.xml", "yarn-site.xml")
val files =
diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/util/Logger.scala b/streampark-common/src/main/scala/org/apache/streampark/common/util/Logger.scala
index 583fc91f2..0b97498b1 100644
--- a/streampark-common/src/main/scala/org/apache/streampark/common/util/Logger.scala
+++ b/streampark-common/src/main/scala/org/apache/streampark/common/util/Logger.scala
@@ -140,7 +140,7 @@ private[this] object LoggerFactory extends LoggerFactoryBinder {
val configurator = new JoranConfigurator()
configurator.setContext(loggerContext)
val text = FileUtils
- .readString(new File(path))
+ .readFile(new File(path))
.replaceAll("org.slf4j", s"$shadedPackage.org.slf4j")
.replaceAll("ch.qos.logback", s"$shadedPackage.ch.qos.logback")
.replaceAll("org.apache.log4j", s"$shadedPackage.org.apache.log4j")
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/util/FileUtils.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/util/FileUtils.java
deleted file mode 100644
index 4aaa624f2..000000000
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/util/FileUtils.java
+++ /dev/null
@@ -1,107 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.streampark.console.base.util;
-
-import org.apache.streampark.console.base.exception.ApiDetailException;
-
-import java.io.File;
-import java.io.IOException;
-import java.io.RandomAccessFile;
-import java.nio.file.Files;
-import java.nio.file.Paths;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-
-/** The file utils. */
-public class FileUtils {
-
- private FileUtils() {}
-
- /**
- * Reads the last portion of a file as a byte array.
- *
- * @param file the file to read
- * @param maxSize the maximum number of bytes to read from the end of the file
- * @return the byte array containing the content read from the file
- * @throws IOException if an I/O error occurs
- */
- public static byte[] readEndOfFile(File file, long maxSize) throws IOException {
- long readSize = maxSize;
- byte[] fileContent;
- try (RandomAccessFile raFile = new RandomAccessFile(file, "r")) {
- if (raFile.length() > maxSize) {
- raFile.seek(raFile.length() - maxSize);
- } else if (raFile.length() < maxSize) {
- readSize = (int) raFile.length();
- }
- fileContent = new byte[(int) readSize];
- raFile.read(fileContent);
- }
- return fileContent;
- }
-
- /**
- * Read the content of a file from a specified offset.
- *
- * @param file The file to read from
- * @param startOffset The offset from where to start reading the file
- * @param maxSize The maximum size of the file to read
- * @return The content of the file as a byte array
- * @throws IOException if an I/O error occurs while reading the file
- * @throws IllegalArgumentException if the startOffset is greater than the file length
- */
- public static byte[] readFileFromOffset(File file, long startOffset, long maxSize)
- throws IOException {
- if (file.length() < startOffset) {
- throw new IllegalArgumentException(
- String.format(
- "The startOffset %s is great than the file length %s", startOffset, file.length()));
- }
- byte[] fileContent;
- try (RandomAccessFile raFile = new RandomAccessFile(file, "r")) {
- long readSize = Math.min(maxSize, file.length() - startOffset);
- raFile.seek(startOffset);
- fileContent = new byte[(int) readSize];
- raFile.read(fileContent);
- }
- return fileContent;
- }
-
- /**
- * Roll View Log.
- *
- * @param path The file path.
- * @param offset The offset.
- * @param limit The limit.
- * @return The content of the file.
- * @throws ApiDetailException if there's an error rolling the view log.
- */
- public static String rollViewLog(String path, int offset, int limit) {
- try {
- File file = new File(path);
- if (file.exists() && file.isFile()) {
- try (Stream<String> stream = Files.lines(Paths.get(path))) {
- return stream.skip(offset).limit(limit).collect(Collectors.joining("\r\n"));
- }
- }
- return null;
- } catch (Exception e) {
- throw new ApiDetailException("roll view log error: " + e);
- }
- }
-}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java
index f9017a724..a20f5a08c 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java
@@ -424,7 +424,7 @@ public class AppBuildPipeServiceImpl
FlinkEnv flinkEnv = flinkEnvService.getByIdOrDefault(app.getVersionId());
String flinkUserJar = retrieveFlinkUserJar(flinkEnv, app);
- if (!new File(flinkUserJar).exists()) {
+ if (!FileUtils.exists(flinkUserJar)) {
Resource resource = resourceService.findByResourceName(app.getTeamId(), app.getJar());
if (resource != null && StringUtils.isNotBlank(resource.getFilePath())) {
flinkUserJar = resource.getFilePath();
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
index a94879640..6806825b5 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
@@ -29,6 +29,7 @@ import org.apache.streampark.common.fs.HdfsOperator;
import org.apache.streampark.common.fs.LfsOperator;
import org.apache.streampark.common.util.CompletableFutureUtils;
import org.apache.streampark.common.util.DeflaterUtils;
+import org.apache.streampark.common.util.FileUtils;
import org.apache.streampark.common.util.HadoopUtils;
import org.apache.streampark.common.util.PropertiesUtils;
import org.apache.streampark.common.util.ThreadUtils;
@@ -97,7 +98,6 @@ import org.apache.streampark.flink.kubernetes.model.TrackId;
import org.apache.streampark.flink.packer.pipeline.BuildResult;
import org.apache.streampark.flink.packer.pipeline.ShadedBuildResponse;
-import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.CoreOptions;
@@ -129,7 +129,6 @@ import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.net.URI;
-import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Base64;
import java.util.Collection;
@@ -619,9 +618,8 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
if (!future.isDone()) {
future.cancel(true);
}
- if (path != null) {
- return org.apache.streampark.console.base.util.FileUtils.rollViewLog(
- path, offset, limit);
+ if (FileUtils.exists(path)) {
+ return FileUtils.tailOf(path, offset, limit);
}
return null;
})
@@ -738,7 +736,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
jarPath = resource.getFilePath();
}
}
- appParam.setJarCheckSum(FileUtils.checksumCRC32(new File(jarPath)));
+ appParam.setJarCheckSum(org.apache.commons.io.FileUtils.checksumCRC32(new File(jarPath)));
}
if (save(appParam)) {
@@ -866,7 +864,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
File jarFile = new File(WebUtils.getAppTempDir(), appParam.getJar());
if (jarFile.exists()) {
try {
- long checkSum = FileUtils.checksumCRC32(jarFile);
+ long checkSum = org.apache.commons.io.FileUtils.checksumCRC32(jarFile);
if (!Objects.equals(checkSum, application.getJarCheckSum())) {
application.setBuild(true);
}
@@ -1144,7 +1142,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
@Override
public String readConf(Application appParam) throws IOException {
File file = new File(appParam.getConfig());
- String conf = FileUtils.readFileToString(file, StandardCharsets.UTF_8);
+ String conf = FileUtils.readFile(file);
return Base64.getEncoder().encodeToString(conf.getBytes());
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ProjectServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ProjectServiceImpl.java
index e568d976b..d48228706 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ProjectServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ProjectServiceImpl.java
@@ -21,6 +21,7 @@ import org.apache.streampark.common.conf.CommonConfig;
import org.apache.streampark.common.conf.InternalConfigHolder;
import org.apache.streampark.common.conf.Workspace;
import org.apache.streampark.common.util.CompletableFutureUtils;
+import org.apache.streampark.common.util.FileUtils;
import org.apache.streampark.common.util.ThreadUtils;
import org.apache.streampark.common.util.Utils;
import org.apache.streampark.console.base.domain.ResponseCode;
@@ -28,7 +29,6 @@ import org.apache.streampark.console.base.domain.RestRequest;
import org.apache.streampark.console.base.domain.RestResponse;
import org.apache.streampark.console.base.exception.ApiAlertException;
import org.apache.streampark.console.base.mybatis.pager.MybatisPager;
-import org.apache.streampark.console.base.util.FileUtils;
import org.apache.streampark.console.base.util.GZipUtils;
import org.apache.streampark.console.core.entity.Application;
import org.apache.streampark.console.core.entity.Project;
diff --git a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/base/util/FileUtilsTest.java b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/base/util/FileUtilsTest.java
index 57db8581c..d31083327 100644
--- a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/base/util/FileUtilsTest.java
+++ b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/base/util/FileUtilsTest.java
@@ -17,6 +17,8 @@
package org.apache.streampark.console.base.util;
+import org.apache.streampark.common.util.FileUtils;
+
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkStreamingInitializer.scala b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkStreamingInitializer.scala
index b3ee24c33..bf2e2c68b 100644
--- a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkStreamingInitializer.scala
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkStreamingInitializer.scala
@@ -130,7 +130,7 @@ private[flink] class FlinkStreamingInitializer(args: Array[String], apiType: Api
require(
configFile.exists(),
s"[StreamPark] Usage: application config file: $configFile is not found!!!")
- val text = FileUtils.readString(configFile)
+ val text = FileUtils.readFile(configFile)
readConfig(text)
}
map.filter(_._2.nonEmpty)