This is an automated email from the ASF dual-hosted git repository.
casion pushed a commit to branch dev-1.1.3
in repository https://gitbox.apache.org/repos/asf/incubator-linkis.git
The following commit(s) were added to refs/heads/dev-1.1.3 by this push:
new 737c57859 add test for LoopArray and format somecode (#2272)
737c57859 is described below
commit 737c578592ec2947e6be231bfaa4fe1ae4c3f576
Author: Jack Xu <[email protected]>
AuthorDate: Tue Jun 14 11:17:33 2022 +0800
add test for LoopArray and format somecode (#2272)
* refactor: format some code and add test for LoopArray
---
.../assembly-combined/conf/application-linkis.yml | 2 -
.../main/java/org/apache/linkis/common/io/Fs.java | 22 +++----
.../apache/linkis/common/utils/ByteTimeUtils.java | 7 +--
.../linkis/common/collection/LoopArray.scala | 60 ++++++++++---------
.../linkis/common/conf/BDPConfiguration.scala | 29 ++++-----
.../apache/linkis/common/conf/Configuration.scala | 6 +-
.../linkis/common/conf/DWCArgumentsParser.scala | 22 +++----
.../linkis/common/collection/LoopArrayTest.scala | 64 ++++++++++++++++++++
.../linkis/scheduler/queue/LoopArrayQueue.scala | 69 +++++++++++-----------
.../linkis/storage/fs/impl/LocalFileSystem.java | 41 +++++++++++--
10 files changed, 210 insertions(+), 112 deletions(-)
diff --git
a/assembly-combined-package/assembly-combined/conf/application-linkis.yml
b/assembly-combined-package/assembly-combined/conf/application-linkis.yml
index b9104929a..f886ad974 100644
--- a/assembly-combined-package/assembly-combined/conf/application-linkis.yml
+++ b/assembly-combined-package/assembly-combined/conf/application-linkis.yml
@@ -17,8 +17,6 @@ eureka:
client:
serviceUrl:
defaultZone: http://127.0.0.1:20303/eureka/
-
-
management:
endpoints:
web:
diff --git
a/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/io/Fs.java
b/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/io/Fs.java
index f1c1bc48c..d85709338 100644
---
a/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/io/Fs.java
+++
b/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/io/Fs.java
@@ -26,33 +26,33 @@ import java.util.Map;
public interface Fs extends Closeable {
- public abstract void init(Map<String, String> properties) throws
IOException;
+ void init(Map<String, String> properties) throws IOException;
- public abstract String fsName();
+ String fsName();
String rootUserName();
FsPath get(String dest) throws IOException;
- public abstract InputStream read(FsPath dest) throws IOException;
+ InputStream read(FsPath dest) throws IOException;
- public abstract OutputStream write(FsPath dest, boolean overwrite) throws
IOException;
+ OutputStream write(FsPath dest, boolean overwrite) throws IOException;
boolean create(String dest) throws IOException;
List<FsPath> list(final FsPath path) throws IOException;
- public abstract boolean canRead(FsPath dest) throws IOException;
+ boolean canRead(FsPath dest) throws IOException;
- public abstract boolean canWrite(FsPath dest) throws IOException;
+ boolean canWrite(FsPath dest) throws IOException;
- public abstract boolean exists(FsPath dest) throws IOException;
+ boolean exists(FsPath dest) throws IOException;
- public abstract boolean delete(FsPath dest) throws IOException;
+ boolean delete(FsPath dest) throws IOException;
- public abstract boolean renameTo(FsPath oldDest, FsPath newDest) throws
IOException;
+ boolean renameTo(FsPath oldDest, FsPath newDest) throws IOException;
- public abstract boolean mkdir(FsPath dest) throws IOException;
+ boolean mkdir(FsPath dest) throws IOException;
- public abstract boolean mkdirs(FsPath dest) throws IOException;
+ boolean mkdirs(FsPath dest) throws IOException;
}
diff --git
a/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/utils/ByteTimeUtils.java
b/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/utils/ByteTimeUtils.java
index 79a09e78e..30f004652 100644
---
a/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/utils/ByteTimeUtils.java
+++
b/linkis-commons/linkis-common/src/main/java/org/apache/linkis/common/utils/ByteTimeUtils.java
@@ -38,10 +38,9 @@ public class ByteTimeUtils {
return hash != Integer.MIN_VALUE ? Math.abs(hash) : 0;
}
- /*
- * Delete a file or directory and its contents recursively.
- * Don't follow directories if they are symlinks.
- * Throws an exception if deletion is unsuccessful.
+ /**
+ * Delete a file or directory and its contents recursively. Don't follow
directories if they are
+ * symlinks. Throws an exception if deletion is unsuccessful.
*/
public static void deleteRecursively(File file) throws IOException {
if (file == null) {
diff --git
a/linkis-commons/linkis-common/src/main/scala/org/apache/linkis/common/collection/LoopArray.scala
b/linkis-commons/linkis-common/src/main/scala/org/apache/linkis/common/collection/LoopArray.scala
index e42e6787d..5becb7359 100644
---
a/linkis-commons/linkis-common/src/main/scala/org/apache/linkis/common/collection/LoopArray.scala
+++
b/linkis-commons/linkis-common/src/main/scala/org/apache/linkis/common/collection/LoopArray.scala
@@ -43,24 +43,28 @@ class LoopArray[T](maxCapacity: Int) {
t
}
+ @throws(classOf[IllegalArgumentException])
def get(index: Int): T = eventQueue synchronized {
- val _max = max
- if (index < realSize) throw new IllegalArgumentException("The index " +
index + " has already been deleted, now index must be better than " + realSize)
- else if(index > _max) throw new IllegalArgumentException("The index " +
index + " must be less than " + _max)
- val _index = (flag + (index - realSize)) % maxCapacity
- eventQueue(_index).asInstanceOf[T]
+ val curMax = max
+ if (index < realSize) {
+ throw new IllegalArgumentException("The index " + index + " has already
been deleted, now index must be better than " + realSize)
+ } else if (index > curMax) {
+ throw new IllegalArgumentException("The index " + index + " must be less
than " + curMax)
+ }
+
+ eventQueue(index % maxCapacity).asInstanceOf[T]
}
- def clear() = eventQueue synchronized {
+ def clear(): Unit = eventQueue synchronized {
flag = 0
tail = 0
realSize = 0
(0 until maxCapacity).foreach(eventQueue(_) = null)
}
- def min = realSize
+ def min: Int = realSize
- def max = {
+ def max: Int = {
var _size = filledSize
if(_size == 0) {
_size = 1
@@ -68,17 +72,17 @@ class LoopArray[T](maxCapacity: Int) {
realSize + _size - 1
}
- private def filledSize = if(tail >= flag) tail - flag else tail +
maxCapacity - flag
+ private def filledSize: Int = if (tail >= flag) tail - flag else tail +
maxCapacity - flag
- def size = filledSize
+ def size: Int = filledSize
- def isFull = filledSize == maxCapacity - 1
+ def isFull: Boolean = filledSize == maxCapacity - 1
- def nonEmpty = size > 0
+ def nonEmpty: Boolean = size > 0
- def toList = toIndexedSeq.toList
+ def toList: List[T] = toIndexedSeq.toList
- def toIndexedSeq: IndexedSeq[T] = if(filledSize == 0) IndexedSeq.empty[T]
else eventQueue synchronized {(min to max).map(get)}
+ def toIndexedSeq: IndexedSeq[T] = if (filledSize == 0) IndexedSeq.empty[T]
else eventQueue synchronized {(min to max).map(get)}
}
@@ -98,7 +102,7 @@ class BlockingLoopArray[T](maxCapacity: Int = 32) extends
LoopArray[T](maxCapaci
*/
def put(event: T): Boolean = {
writeLock synchronized {
- while(isFull) writeLock.wait(1000)
+ while (isFull) writeLock.wait(1000)
super.add(event)
}
readLock synchronized { readLock.notify() }
@@ -110,9 +114,9 @@ class BlockingLoopArray[T](maxCapacity: Int = 32) extends
LoopArray[T](maxCapaci
* @param event
* @return
*/
- def offer(event: T): Boolean = if(isFull) false else {
+ def offer(event: T): Boolean = if (isFull) false else {
writeLock synchronized {
- if(isFull) return false
+ if (isFull) return false
else super.add(event)
}
readLock synchronized { readLock.notify() }
@@ -125,10 +129,10 @@ class BlockingLoopArray[T](maxCapacity: Int = 32) extends
LoopArray[T](maxCapaci
*/
def take(): T = {
val t = readLock synchronized {
- while(waitingSize == 0 || takeIndex > max) {
+ while (waitingSize == 0 || takeIndex > max) {
readLock.wait(1000)
}
- if(takeIndex < min) takeIndex = min
+ if (takeIndex < min) takeIndex = min
val t = get(takeIndex)
takeIndex += 1
t
@@ -143,8 +147,8 @@ class BlockingLoopArray[T](maxCapacity: Int = 32) extends
LoopArray[T](maxCapaci
* @return
*/
def peek(): Option[T] = readLock synchronized {
- if(waitingSize == 0 || takeIndex > max) None
- else if(takeIndex < min) Some(get(min))
+ if (waitingSize == 0 || takeIndex > max) None
+ else if (takeIndex < min) Some(get(min))
else Option(get(takeIndex))
}
@@ -154,12 +158,12 @@ class BlockingLoopArray[T](maxCapacity: Int = 32) extends
LoopArray[T](maxCapaci
* @return
*/
def poll(): Option[T] = {
- if(waitingSize == 0) return None
+ if (waitingSize == 0) return None
val event = readLock synchronized {
val _min = min
val _max = max
if(takeIndex < _min) takeIndex = _min
- else if(takeIndex > _max) return None
+ else if (takeIndex > _max) return None
val t = get(takeIndex)
takeIndex += 1
Option(t)
@@ -168,9 +172,9 @@ class BlockingLoopArray[T](maxCapacity: Int = 32) extends
LoopArray[T](maxCapaci
event
}
- override def isFull = super.isFull && takeIndex == realSize
+ override def isFull: Boolean = super.isFull && takeIndex == realSize
- def waitingSize: Int = if(takeIndex <= realSize) super.size else {
+ def waitingSize: Int = if (takeIndex <= realSize) super.size else {
val length = super.size - takeIndex + realSize
if(length < 0) 0 else length
}
@@ -180,13 +184,13 @@ class BlockingLoopArray[T](maxCapacity: Int = 32) extends
LoopArray[T](maxCapaci
super.clear()
}
- override def toIndexedSeq: IndexedSeq[T] = if(waitingSize == 0)
IndexedSeq.empty[T] else readLock synchronized {(takeIndex to max).map(get)}
+ override def toIndexedSeq: IndexedSeq[T] = if (waitingSize == 0)
IndexedSeq.empty[T] else readLock synchronized {(takeIndex to max).map(get)}
}
object LoopArray {
- def apply[T](maxCapacity: Int) = new LoopArray[T](maxCapacity)
+ def apply[T](maxCapacity: Int): LoopArray[T] = new LoopArray[T](maxCapacity)
- def apply[T]() = new LoopArray[T]()
+ def apply[T](): LoopArray[T] = new LoopArray[T]()
}
diff --git
a/linkis-commons/linkis-common/src/main/scala/org/apache/linkis/common/conf/BDPConfiguration.scala
b/linkis-commons/linkis-common/src/main/scala/org/apache/linkis/common/conf/BDPConfiguration.scala
index 917e065bc..160543288 100644
---
a/linkis-commons/linkis-common/src/main/scala/org/apache/linkis/common/conf/BDPConfiguration.scala
+++
b/linkis-commons/linkis-common/src/main/scala/org/apache/linkis/common/conf/BDPConfiguration.scala
@@ -21,9 +21,9 @@ import java.io.{File, FileInputStream, IOException,
InputStream}
import java.util.Properties
import org.apache.linkis.common.utils.{Logging, Utils}
import org.apache.commons.io.IOUtils
-import org.apache.commons.lang.StringUtils
+import org.apache.commons.lang3.StringUtils
-import scala.collection.JavaConversions._
+import scala.collection.JavaConverters.mapAsJavaMapConverter
private[conf] object BDPConfiguration extends Logging {
@@ -99,29 +99,30 @@ private[conf] object BDPConfiguration extends Logging {
}
def getOption(key: String): Option[String] = {
- if(extractConfig.containsKey(key))
+ if (extractConfig.containsKey(key)) {
return Some(extractConfig.getProperty(key))
+ }
val value = config.getProperty(key)
if(StringUtils.isNotEmpty(value)) {
return Some(value)
}
- val propsValue = sysProps.get(key).orElse(sys.props.get(key))
- if(propsValue.isDefined){
+ val propsValue = sysProps.get(key).orElse(sys.props.get(key))
+ if (propsValue.isDefined) {
return propsValue
}
env.get(key)
}
- def properties = {
+ def properties: Properties = {
val props = new Properties
- props.putAll(sysProps)
+ props.putAll(sysProps.asJava)
props.putAll(config)
props.putAll(extractConfig)
- props.putAll(env)
+ props.putAll(env.asJava)
props
}
- def getOption[T](commonVars: CommonVars[T]): Option[T] = if(commonVars.value
!= null) Option(commonVars.value)
+ def getOption[T](commonVars: CommonVars[T]): Option[T] = if
(commonVars.value != null) Option(commonVars.value)
else {
val value = BDPConfiguration.getOption(commonVars.key)
if (value.isEmpty) Option(commonVars.defaultValue)
@@ -129,7 +130,7 @@ private[conf] object BDPConfiguration extends Logging {
}
private[common] def formatValue[T](defaultValue: T, value: Option[String]):
Option[T] = {
- if(value.isEmpty || value.exists(StringUtils.isEmpty)) return
Option(defaultValue)
+ if (value.isEmpty || value.exists(StringUtils.isEmpty)) return
Option(defaultValue)
val formattedValue = defaultValue match {
case _: String => value
case _: Byte => value.map(_.toByte)
@@ -147,11 +148,11 @@ private[conf] object BDPConfiguration extends Logging {
formattedValue.asInstanceOf[Option[T]]
}
- def set(key: String, value: String) = extractConfig.setProperty(key, value)
+ def set(key: String, value: String): AnyRef = extractConfig.setProperty(key,
value)
- def setIfNotExists(key: String, value: String) =
if(!config.containsKey(key)) set(key, value)
+ def setIfNotExists(key: String, value: String): Any = if
(!config.containsKey(key)) set(key, value)
- def getBoolean(key: String, default: Boolean):Boolean =
getOption(key).map(_.toBoolean).getOrElse(default)
+ def getBoolean(key: String, default: Boolean): Boolean =
getOption(key).map(_.toBoolean).getOrElse(default)
def getBoolean(commonVars: CommonVars[Boolean]): Option[Boolean] =
getOption(commonVars)
def get(key: String, default: String): String =
getOption(key).getOrElse(default)
@@ -159,7 +160,7 @@ private[conf] object BDPConfiguration extends Logging {
def get(key: String): String = getOption(key).getOrElse(throw new
NoSuchElementException(key))
- def getInt(key: String, default: Int):Int =
getOption(key).map(_.toInt).getOrElse(default)
+ def getInt(key: String, default: Int): Int =
getOption(key).map(_.toInt).getOrElse(default)
def getInt(commonVars: CommonVars[Int]): Option[Int] = getOption(commonVars)
def contains(key: String): Boolean = getOption(key).isDefined
diff --git
a/linkis-commons/linkis-common/src/main/scala/org/apache/linkis/common/conf/Configuration.scala
b/linkis-commons/linkis-common/src/main/scala/org/apache/linkis/common/conf/Configuration.scala
index 2ba403b56..c5a0cc984 100644
---
a/linkis-commons/linkis-common/src/main/scala/org/apache/linkis/common/conf/Configuration.scala
+++
b/linkis-commons/linkis-common/src/main/scala/org/apache/linkis/common/conf/Configuration.scala
@@ -38,16 +38,16 @@ object Configuration extends Logging {
val GATEWAY_URL: CommonVars[String] =
CommonVars[String]("wds.linkis.gateway.url", "http://127.0.0.1:9001/";)
- val LINKIS_WEB_VERSION:CommonVars[String] =
CommonVars[String]("wds.linkis.web.version", "v1")
+ val LINKIS_WEB_VERSION: CommonVars[String] =
CommonVars[String]("wds.linkis.web.version", "v1")
- val REFLECT_SCAN_PACKAGE =
CommonVars.apply("wds.linkis.reflect.scan.package",
"org.apache.linkis,com.webank.wedatasphere").getValue.split(",")
+ val REFLECT_SCAN_PACKAGE: Array[String] =
CommonVars.apply("wds.linkis.reflect.scan.package",
"org.apache.linkis,com.webank.wedatasphere").getValue.split(",")
val CLOUD_CONSOLE_CONFIGURATION_SPRING_APPLICATION_NAME =
CommonVars("wds.linkis.console.configuration.application.name",
"linkis-ps-configuration")
val CLOUD_CONSOLE_VARIABLE_SPRING_APPLICATION_NAME =
CommonVars("wds.linkis.console.variable.application.name",
"linkis-ps-publicservice")
//read from env
- val EUREKA_PREFER_IP = CommonVars("EUREKA_PREFER_IP", false).getValue
+ val EUREKA_PREFER_IP: Boolean = CommonVars("EUREKA_PREFER_IP",
false).getValue
val GOVERNANCE_STATION_ADMIN =
CommonVars("wds.linkis.governance.station.admin", "hadoop")
diff --git
a/linkis-commons/linkis-common/src/main/scala/org/apache/linkis/common/conf/DWCArgumentsParser.scala
b/linkis-commons/linkis-common/src/main/scala/org/apache/linkis/common/conf/DWCArgumentsParser.scala
index 2719aa97f..65db1fe28 100644
---
a/linkis-commons/linkis-common/src/main/scala/org/apache/linkis/common/conf/DWCArgumentsParser.scala
+++
b/linkis-commons/linkis-common/src/main/scala/org/apache/linkis/common/conf/DWCArgumentsParser.scala
@@ -17,11 +17,11 @@
package org.apache.linkis.common.conf
-import org.apache.commons.lang.StringUtils
+import org.apache.commons.lang3.StringUtils
+import scala.collection.JavaConverters.mapAsJavaMapConverter
import scala.collection.mutable.ArrayBuffer
-import scala.collection.{JavaConversions, mutable}
-
+import scala.collection.mutable
object DWCArgumentsParser {
protected val DWC_CONF = "--engineconn-conf"
@@ -29,7 +29,7 @@ object DWCArgumentsParser {
private var dwcOptionMap = Map.empty[String, String]
private[linkis] def setDWCOptionMap(dwcOptionMap: Map[String, String]) =
this.dwcOptionMap = dwcOptionMap
- def getDWCOptionMap = dwcOptionMap
+ def getDWCOptionMap: Map[String, String] = dwcOptionMap
def parse(args: Array[String]): DWCArgumentsParser = {
val keyValueRegex = "([^=]+)=(.+)".r
@@ -85,10 +85,10 @@ class DWCArgumentsParser {
import DWCArgumentsParser._
private val dwcOptionMap = new mutable.HashMap[String, String]()
private val springOptionMap = new mutable.HashMap[String, String]()
- def getSpringConfMap = springOptionMap.toMap
- def getSpringConfs = JavaConversions.mapAsJavaMap(springOptionMap)
- def getDWCConfMap = dwcOptionMap.toMap
- def setConf(optionType: String, key: String, value: String) = {
+ def getSpringConfMap: Map[String, String] = springOptionMap.toMap
+ def getSpringConfs: java.util.Map[String, String] = springOptionMap.asJava
+ def getDWCConfMap: Map[String, String] = dwcOptionMap.toMap
+ def setConf(optionType: String, key: String, value: String):
DWCArgumentsParser = {
optionType match {
case DWC_CONF =>
dwcOptionMap += key -> value
@@ -98,12 +98,12 @@ class DWCArgumentsParser {
this
}
def setSpringConf(optionMap: Map[String, String]): DWCArgumentsParser = {
- if(optionMap != null) this.springOptionMap ++= optionMap
+ if (optionMap != null) this.springOptionMap ++= optionMap
this
}
def setDWCConf(optionMap: Map[String, String]): DWCArgumentsParser = {
- if(optionMap != null) this.dwcOptionMap ++= optionMap
+ if (optionMap != null) this.dwcOptionMap ++= optionMap
this
}
- def validate() = {}
+ def validate(): Unit = {}
}
\ No newline at end of file
diff --git
a/linkis-commons/linkis-common/src/test/scala/org/apache/linkis/common/collection/LoopArrayTest.scala
b/linkis-commons/linkis-common/src/test/scala/org/apache/linkis/common/collection/LoopArrayTest.scala
new file mode 100644
index 000000000..c603fb4ec
--- /dev/null
+++
b/linkis-commons/linkis-common/src/test/scala/org/apache/linkis/common/collection/LoopArrayTest.scala
@@ -0,0 +1,64 @@
+/*
+ * 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.linkis.common.collection
+
+import org.assertj.core.api.Assertions.{assertThat, assertThatThrownBy}
+import org.assertj.core.api.ThrowableAssert.ThrowingCallable
+import org.junit.jupiter.api.{BeforeEach, Test}
+
+class LoopArrayTest {
+
+ private var loopArray: LoopArray[Int] = _
+ private val capability = 4
+ private val total = 10
+ private val realSize = 8
+
+ @BeforeEach private[collection] def setUp(): Unit = {
+ loopArray = LoopArray(capability)
+ for (i <- 0 to total) {
+ loopArray.add(i)
+ }
+ }
+
+ @Test private[collection] def add(): Unit = {
+ val removed = loopArray.add(realSize)
+ assertThat(removed).isEqualTo(realSize-1)
+ assertThat(loopArray.max).isEqualTo(total + 1)
+ assertThat(loopArray.min).isEqualTo(realSize + 1)
+ }
+
+ @Test private[collection] def clear(): Unit = {
+ loopArray.clear()
+ assertThat(loopArray.max).isEqualTo(0)
+ assertThat(loopArray.min).isEqualTo(0)
+ }
+
+ @Test private[collection] def get(): Unit = {
+ assertThatThrownBy(new ThrowingCallable {
+ override def call(): Unit = {
+ loopArray.get(realSize-1)
+ }
+ }).isInstanceOf(classOf[IllegalArgumentException])
+ assertThat(loopArray.get(total)).isEqualTo(total)
+ assertThatThrownBy(new ThrowingCallable {
+ override def call(): Unit = {
+ loopArray.get(total + 1)
+ }
+ }).isInstanceOf(classOf[IllegalArgumentException])
+ }
+}
diff --git
a/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/LoopArrayQueue.scala
b/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/LoopArrayQueue.scala
index e4a83d6c7..dd34fa14f 100644
---
a/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/LoopArrayQueue.scala
+++
b/linkis-commons/linkis-scheduler/src/main/scala/org/apache/linkis/scheduler/queue/LoopArrayQueue.scala
@@ -33,7 +33,7 @@ class LoopArrayQueue(var group: Group) extends ConsumeQueue
with Logging {
protected[this] var realSize = 0
- private def filledSize = if(tail >= flag) tail - flag else tail +
maxCapacity - flag
+ private def filledSize: Int = if (tail >= flag) tail - flag else tail +
maxCapacity - flag
override def remove(event: SchedulerEvent): Unit = {
get(event).foreach(x => x.cancel())
@@ -47,7 +47,7 @@ class LoopArrayQueue(var group: Group) extends ConsumeQueue
with Logging {
override def size: Int = filledSize
- override def isEmpty: Boolean = size == 0
+ override def isEmpty: Boolean = size == 0
override def isFull: Boolean = filledSize == maxCapacity - 1 && takeIndex ==
realSize
@@ -60,19 +60,19 @@ class LoopArrayQueue(var group: Group) extends ConsumeQueue
with Logging {
override def get(event: SchedulerEvent): Option[SchedulerEvent] = {
eventQueue synchronized {
- val eventSeq =toIndexedSeq.filter(x => x.getId.equals(event.getId)).seq
+ val eventSeq = toIndexedSeq.filter(x => x.getId.equals(event.getId)).seq
if(eventSeq.size >0) Some(eventSeq(0)) else None
}
}
override def get(index: Int): Option[SchedulerEvent] = {
- var event:SchedulerEvent =null
+ var event: SchedulerEvent = null
eventQueue synchronized {
val _max = max
if (index < realSize) throw new IllegalArgumentException("The index " +
index + " has already been deleted, now index must be better than " + realSize)
- else if(index > _max) throw new IllegalArgumentException("The index " +
index + " must be less than " + _max)
+ else if (index > _max) throw new IllegalArgumentException("The index " +
index + " must be less than " + _max)
val _index = (flag + (index - realSize)) % maxCapacity
- event =eventQueue(_index).asInstanceOf[SchedulerEvent]
+ event = eventQueue(_index).asInstanceOf[SchedulerEvent]
}
Option(event)
}
@@ -80,12 +80,13 @@ class LoopArrayQueue(var group: Group) extends ConsumeQueue
with Logging {
override def getGroup: Group = group
override def setGroup(group: Group): Unit = {
- this.group=group
+ this.group = group
}
- def toIndexedSeq: IndexedSeq[SchedulerEvent] = if(filledSize == 0)
IndexedSeq.empty[SchedulerEvent] else eventQueue synchronized {(min to
max).map(x =>get(x).get).filter(x => x != None)}
+ def toIndexedSeq: IndexedSeq[SchedulerEvent] = if (filledSize == 0)
IndexedSeq.empty[SchedulerEvent]
+ else eventQueue synchronized {(min to max).map(x => get(x).get).filter(x =>
x != None) }
- def add(event:SchedulerEvent):Int={
+ def add(event: SchedulerEvent): Int = {
eventQueue synchronized {
val index = (tail + 1) % maxCapacity
if (index == flag) {
@@ -98,14 +99,14 @@ class LoopArrayQueue(var group: Group) extends ConsumeQueue
with Logging {
max
}
- def waitingSize: Int = if(takeIndex <= realSize) size else {
+ def waitingSize: Int = if (takeIndex <= realSize) size else {
val length = size - takeIndex + realSize
if(length < 0) 0 else length
}
- def min = realSize
+ def min: Int = realSize
- def max = {
+ def max: Int = {
var _size = filledSize
if(_size == 0) {
_size = 1
@@ -113,8 +114,6 @@ class LoopArrayQueue(var group: Group) extends ConsumeQueue
with Logging {
realSize + _size - 1
}
-
-
/**
* Add one, if the queue is full, it will block until the queue is
available（添加一个，如果队列满了，将会一直阻塞，直到队列可用）
*
@@ -123,7 +122,7 @@ class LoopArrayQueue(var group: Group) extends ConsumeQueue
with Logging {
override def put(event: SchedulerEvent): Int = {
var index = -1
writeLock synchronized {
- while(isFull) writeLock.wait(1000)
+ while (isFull) writeLock.wait(1000)
index = add(event)
}
readLock synchronized { readLock.notify() }
@@ -136,17 +135,17 @@ class LoopArrayQueue(var group: Group) extends
ConsumeQueue with Logging {
*
* @return
*/
-override def offer(event: SchedulerEvent): Option[Int] = {
- var index = -1
- writeLock synchronized {
- if(isFull) return None
- else{
- index = add(event)
+ override def offer(event: SchedulerEvent): Option[Int] = {
+ var index = -1
+ writeLock synchronized {
+ if (isFull) return None
+ else {
+ index = add(event)
+ }
}
+ readLock synchronized { readLock.notify() }
+ Some(index)
}
- readLock synchronized { readLock.notify() }
- Some(index)
-}
/**
* Get the latest SchedulerEvent of a group, if it does not exist, it will
block [<br>（获取某个group最新的SchedulerEvent，如果不存在，就一直阻塞<br>）
@@ -156,10 +155,10 @@ override def offer(event: SchedulerEvent): Option[Int] = {
*/
override def take(): SchedulerEvent = {
val t = readLock synchronized {
- while(waitingSize == 0 || takeIndex > max) {
+ while (waitingSize == 0 || takeIndex > max) {
readLock.wait(1000)
}
- if(takeIndex < min) takeIndex = min
+ if (takeIndex < min) takeIndex = min
val t = get(takeIndex)
takeIndex += 1
t
@@ -176,9 +175,9 @@ override def offer(event: SchedulerEvent): Option[Int] = {
*/
override def take(mills: Long): Option[SchedulerEvent] = {
val t = readLock synchronized {
- if(waitingSize == 0 || takeIndex > max) readLock.wait(mills)
- if(waitingSize == 0 || takeIndex > max) return None
- if(takeIndex < min) takeIndex = min
+ if (waitingSize == 0 || takeIndex > max) readLock.wait(mills)
+ if (waitingSize == 0 || takeIndex > max) return None
+ if (takeIndex < min) takeIndex = min
val t = get(takeIndex)
takeIndex += 1
t
@@ -198,12 +197,12 @@ override def offer(event: SchedulerEvent): Option[Int] = {
val _min = min
val _max = max
if(takeIndex < _min) takeIndex = _min
- else if(takeIndex > _max) {
+ else if (takeIndex > _max) {
logger.info(s"none, notice...max: ${_max}, takeIndex: $takeIndex,
realSize: $realSize.")
return None
}
val t = get(takeIndex)
- if(t == null) {
+ if (t == null) {
logger.info("null, notice...")
}
takeIndex += 1
@@ -220,8 +219,8 @@ override def offer(event: SchedulerEvent): Option[Int] = {
* @return
*/
override def peek(): Option[SchedulerEvent] = readLock synchronized {
- if(waitingSize == 0 || takeIndex > max) None
- else if(takeIndex < min) get(min)
+ if (waitingSize == 0 || takeIndex > max) None
+ else if (takeIndex < min) get(min)
else get(takeIndex)
}
@@ -232,8 +231,8 @@ override def offer(event: SchedulerEvent): Option[Int] = {
* @return
*/
override def peek(op: (SchedulerEvent) => Boolean): Option[SchedulerEvent] =
{
- if(waitingSize == 0 || takeIndex > max) None
- else if(takeIndex < min) {
+ if (waitingSize == 0 || takeIndex > max) None
+ else if (takeIndex < min) {
val event = get(min)
if(op(event.get)) event else None
}
diff --git
a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/fs/impl/LocalFileSystem.java
b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/fs/impl/LocalFileSystem.java
index 8db01b325..a4fc1312e 100644
---
a/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/fs/impl/LocalFileSystem.java
+++
b/linkis-commons/linkis-storage/src/main/java/org/apache/linkis/storage/fs/impl/LocalFileSystem.java
@@ -32,10 +32,30 @@ import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.*;
-import java.nio.file.*;
-import java.nio.file.attribute.*;
-import java.util.*;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.LinkOption;
+import java.nio.file.NoSuchFileException;
+import java.nio.file.Paths;
+import java.nio.file.attribute.GroupPrincipal;
+import java.nio.file.attribute.PosixFileAttributeView;
+import java.nio.file.attribute.PosixFileAttributes;
+import java.nio.file.attribute.PosixFilePermission;
+import java.nio.file.attribute.PosixFilePermissions;
+import java.nio.file.attribute.UserPrincipal;
+import java.nio.file.attribute.UserPrincipalLookupService;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.Stack;
public class LocalFileSystem extends FileSystem {
@@ -260,6 +280,7 @@ public class LocalFileSystem extends FileSystem {
* <p>TODO Caching /etc/passwd information to the local as
* object(将/etc/passwd的信息缓存到本地作为object进行判断)
*/
+ @Override
public void init(Map<String, String> properties) throws IOException {
if (MapUtils.isNotEmpty(properties)) {
@@ -284,14 +305,17 @@ public class LocalFileSystem extends FileSystem {
}
}
+ @Override
public String fsName() {
return "file";
}
+ @Override
public String rootUserName() {
return StorageConfiguration.LOCAL_ROOT_USER().getValue();
}
+ @Override
public FsPath get(String dest) throws IOException {
FsPath fsPath = null;
if (FsPath.WINDOWS) {
@@ -319,6 +343,7 @@ public class LocalFileSystem extends FileSystem {
return fsPath;
}
+ @Override
public InputStream read(FsPath dest) throws IOException {
if (canRead(dest)) {
return new FileInputStream(dest.getPath());
@@ -326,6 +351,7 @@ public class LocalFileSystem extends FileSystem {
throw new IOException("you have no permission to read path " +
dest.getPath());
}
+ @Override
public OutputStream write(FsPath dest, boolean overwrite) throws
IOException {
String path = dest.getPath();
if (new File(path).isDirectory()) {
@@ -339,6 +365,7 @@ public class LocalFileSystem extends FileSystem {
throw new IOException("you have no permission to write file " + path);
}
+ @Override
public boolean create(String dest) throws IOException {
File file = new File(dest);
@@ -362,6 +389,7 @@ public class LocalFileSystem extends FileSystem {
return true;
}
+ @Override
public List<FsPath> list(FsPath path) throws IOException {
File file = new File(path.getPath());
File[] files = file.listFiles();
@@ -376,6 +404,7 @@ public class LocalFileSystem extends FileSystem {
}
}
+ @Override
public boolean canRead(FsPath dest) throws IOException {
return can(
dest,
@@ -384,6 +413,7 @@ public class LocalFileSystem extends FileSystem {
PosixFilePermission.OTHERS_READ);
}
+ @Override
public boolean canWrite(FsPath dest) throws IOException {
return can(
dest,
@@ -392,10 +422,12 @@ public class LocalFileSystem extends FileSystem {
PosixFilePermission.OTHERS_WRITE);
}
+ @Override
public boolean exists(FsPath dest) throws IOException {
return new File(dest.getPath()).exists();
}
+ @Override
public boolean delete(FsPath dest) throws IOException {
String path = dest.getPath();
if (isOwner(path)) {
@@ -404,6 +436,7 @@ public class LocalFileSystem extends FileSystem {
throw new IOException("only owner can delete file " + path);
}
+ @Override
public boolean renameTo(FsPath oldDest, FsPath newDest) throws IOException
{
String path = oldDest.getPath();
if (isOwner(path)) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
