Skip to content

Commit 739adf4

Browse files
binbinChengbinbinCheng
authored
[linkis-orchestrator-ecm-plugin] Modification of scala file floating red (#3166)
* [linkis-orchestrator-ecm-plugin] Modification of scala file floating red
1 parent c01d9a7 commit 739adf4

File tree

3 files changed

+28
-38
lines changed

3 files changed

+28
-38
lines changed

‎linkis-orchestrator/plugin/linkis-orchestrator-ecm-plugin/src/main/scala/org/apache/linkis/orchestrator/ecm/ComputationEngineConnManager.scala

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ import java.util
4848
import java.util.concurrent.TimeUnit
4949
import java.util.concurrent.atomic.AtomicInteger
5050

51-
import scala.collection.JavaConversions._
51+
import scala.collection.JavaConverters._
5252
import scala.concurrent.duration.Duration
5353

5454
/**
@@ -64,7 +64,7 @@ class ComputationEngineConnManager extends AbstractEngineConnManager with Loggin
6464
override def applyMark(markReq: MarkReq): Mark = {
6565
if (null == markReq) return null
6666
val mark = MARK_CACHE_LOCKER.synchronized {
67-
val markCache = getMarkCache().keys
67+
val markCache = getMarkCache().asScala.keys
6868
val maybeMark = markCache.find(_.getMarkReq.equals(markReq))
6969
maybeMark.orNull
7070
}
@@ -120,7 +120,7 @@ class ComputationEngineConnManager extends AbstractEngineConnManager with Loggin
120120
new ComputationEngineConnExecutor(engineNode)
121121
}
122122
if (null != engineNode.getLabels) {
123-
engineConnExecutor.setLabels(engineNode.getLabels.toList.toArray)
123+
engineConnExecutor.setLabels(engineNode.getLabels.asScala.toList.toArray)
124124
}
125125
return engineConnExecutor
126126
}

‎linkis-orchestrator/plugin/linkis-orchestrator-ecm-plugin/src/main/scala/org/apache/linkis/orchestrator/ecm/EngineConnManager.scala

Lines changed: 11 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ import org.apache.linkis.orchestrator.ecm.service.EngineConnExecutor
2828

2929
import java.util
3030

31-
import scala.collection.JavaConversions._
31+
import scala.collection.JavaConverters._
3232

3333
/**
3434
*/
@@ -146,12 +146,13 @@ abstract class AbstractEngineConnManager extends EngineConnManager with Logging
146146
val instances = getInstances(mark)
147147
if (null != instances) {
148148
val executors = Utils.tryAndWarn {
149-
instances.map(getEngineConnExecutorCache().get(_)).filter(null != _).sortBy { executor =>
150-
if (null == executor.getRunningTaskCount) {
151-
0
152-
} else {
153-
executor.getRunningTaskCount
154-
}
149+
instances.asScala.map(getEngineConnExecutorCache().get(_)).filter(null != _).sortBy {
150+
executor =>
151+
if (null == executor.getRunningTaskCount) {
152+
0
153+
} else {
154+
executor.getRunningTaskCount
155+
}
155156
}
156157
}
157158

@@ -206,9 +207,9 @@ abstract class AbstractEngineConnManager extends EngineConnManager with Logging
206207

207208
protected def getMarksByInstance(serviceInstance: ServiceInstance): Array[Mark] =
208209
MARK_CACHE_LOCKER.synchronized {
209-
getMarkCache()
210+
getMarkCache().asScala
210211
.filter { keyValue =>
211-
keyValue._2.exists(serviceInstance.equals(_))
212+
keyValue._2.asScala.exists(serviceInstance.equals(_))
212213
}
213214
.keys
214215
.toArray
@@ -240,7 +241,7 @@ abstract class AbstractEngineConnManager extends EngineConnManager with Logging
240241
override def releaseMark(mark: Mark): Unit = {
241242
if (null != mark && getMarkCache().containsKey(mark)) {
242243
logger.debug(s"Start to release mark ${mark.getMarkId()}")
243-
val executors = getMarkCache().get(mark).map(getEngineConnExecutorCache().get(_))
244+
val executors = getMarkCache().get(mark).asScala.map(getEngineConnExecutorCache().get(_))
244245
Utils.tryAndError(executors.foreach { executor =>
245246
getEngineConnExecutorCache().remove(executor.getServiceInstance)
246247
executor.close()

‎linkis-orchestrator/plugin/linkis-orchestrator-ecm-plugin/src/main/scala/org/apache/linkis/orchestrator/ecm/LoadBalanceLabelEngineConnManager.scala

Lines changed: 14 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ import org.apache.linkis.server.BDPJettyServerHelper
3131
import java.util
3232
import java.util.Random
3333

34-
import scala.collection.JavaConversions._
34+
import scala.collection.JavaConverters._
3535
import scala.collection.mutable.ArrayBuffer
3636

3737
class LoadBalanceLabelEngineConnManager extends ComputationEngineConnManager with Logging {
@@ -65,31 +65,26 @@ class LoadBalanceLabelEngineConnManager extends ComputationEngineConnManager wit
6565
}
6666

6767
/**
68-
* 申请获取一个Mark
69-
* 1. 如果没有对应的Mark就生成新的 2. 一个MarkRequest对应多个Mark，一个Mark对应一个Engine 3.
70-
* 如果存在bindEngineLabel则需要在jobStart的时候随机选择一个，并缓存给后续jobGroup使用，在jobEnd的时候删除缓存 4. 将Mark进行返回
68+
* Apply for a mark
69+
* 1. If there is no corresponding mark, a new one will be generated. 2. A markrequest
70+
* corresponds to multiple marks, and a mark corresponds to an engine. 3 If there is a
71+
* bindenginelabel, you need to randomly select one at jobstart and cache it for subsequent
72+
* jobgroups. Delete the cache at jobend. 4. Return mark
7173
*
7274
* @param markReq
7375
* @return
7476
*/
7577
override def applyMark(markReq: MarkReq): Mark = {
7678
if (null == markReq) return null
77-
7879
val markNum: Int = getMarkNumByMarReq(markReq)
79-
/*val markReqCache = MARK_REQ_CACHE_LOCKER.synchronized {
80-
getMarkReqAndMarkCache().keys
81-
}
82-
var mayBeMarkReq = markReqCache.find(_.equals(markReq)).orNull*/
8380
var count = 0
8481
if (getMarkReqAndMarkCache().containsKey(markReq)) {
8582
count = getMarkReqAndMarkCache().get(markReq).size()
8683
}
87-
// toto check if count >= markNum, will not add more mark
8884
while (count < markNum) {
8985
createMark(markReq)
9086
count = getMarkReqAndMarkCache().get(markReq).size()
9187
}
92-
// markReq is in cache, and mark list is ready
9388
val markList = getMarkReqAndMarkCache().get(markReq)
9489
var chooseMark: Mark = null
9590
if (markReq.getLabels.containsKey(LabelKeyConstant.BIND_ENGINE_KEY)) {
@@ -98,13 +93,10 @@ class LoadBalanceLabelEngineConnManager extends ComputationEngineConnManager wit
9893
markReq.getLabels.get(LabelKeyConstant.BIND_ENGINE_KEY)
9994
)
10095
if (bindEngineLabel.getIsJobGroupHead) {
101-
// getRandom mark
102-
chooseMark = markList.get(new util.Random().nextInt(markList.length))
103-
// chooseMark.asInstanceOf[LoadBalanceMark].setTaskMarkReq(markReq)
96+
chooseMark = markList.get(new util.Random().nextInt(markList.asScala.length))
10497
getIdToMarkCache().put(bindEngineLabel.getJobGroupId, chooseMark)
10598
} else if (getIdToMarkCache().containsKey(bindEngineLabel.getJobGroupId)) {
10699
chooseMark = getIdToMarkCache().get(bindEngineLabel.getJobGroupId)
107-
// chooseMark.asInstanceOf[LoadBalanceMark].setTaskMarkReq(markReq)
108100
val insList = getMarkCache().get(chooseMark)
109101
if (null == insList || insList.size() != 1) {
110102
val msg =
@@ -131,16 +123,11 @@ class LoadBalanceLabelEngineConnManager extends ComputationEngineConnManager wit
131123
}
132124
}
133125
} else {
134-
// treat as isHead and isEnd
135126
chooseMark = markList.get(new Random().nextInt(count))
136127
}
137128
chooseMark
138129
}
139130

140-
/**
141-
* 1. 创建一个新的Mark 2. 生成新的Mark会存在请求引擎的过程，如果请求到了则存入Map中：Mark为Key，EngineConnExecutor为Value 3.
142-
* 生成的Mark数量等于LoadBalance的并发量
143-
*/
144131
override def createMark(markReq: MarkReq): Mark = {
145132
val mark = new LoadBalanceMark(nextMarkId(), markReq)
146133
addMark(mark, new util.ArrayList[ServiceInstance]())
@@ -152,7 +139,7 @@ class LoadBalanceLabelEngineConnManager extends ComputationEngineConnManager wit
152139
if (null != req) MARK_REQ_CACHE_LOCKER.synchronized {
153140
val markList = getMarkReqAndMarkCache().get(req)
154141
if (null != markList) {
155-
val mayBeMark = markList.find(_.getMarkId().equals(mark.getMarkId())).orNull
142+
val mayBeMark = markList.asScala.find(_.getMarkId().equals(mark.getMarkId())).orNull
156143
if (null == mayBeMark) {
157144
markList.add(mark)
158145
} else {
@@ -231,20 +218,22 @@ class LoadBalanceLabelEngineConnManager extends ComputationEngineConnManager wit
231218
if (null == marks || marks.isEmpty) {
232219
getMarkReqAndMarkCache().remove(mark.getMarkReq)
233220
} else {
234-
val newMarks = marks.filter(!_.getMarkId().equals(mark.getMarkId()))
221+
val newMarks = marks.asScala.filter(!_.getMarkId().equals(mark.getMarkId()))
235222
if (null == newMarks || newMarks.isEmpty) {
236223
getMarkReqAndMarkCache().remove(mark.getMarkReq)
237224
} else {
238-
getMarkReqAndMarkCache().put(mark.getMarkReq, newMarks)
225+
getMarkReqAndMarkCache().put(mark.getMarkReq, newMarks.asJava)
239226
}
240-
// getMarkReqAndMarkCache().put(mark.getMarkReq))
241227
}
242228
}
243229
}
244230

245231
protected def getAllInstances(): Array[String] = MARK_CACHE_LOCKER.synchronized {
246232
val instances = new ArrayBuffer[String]
247-
getMarkCache().values().foreach(_.foreach(s => instances.add(s.getInstance)))
233+
getMarkCache()
234+
.values()
235+
.asScala
236+
.foreach(_.asScala.foreach(s => instances.asJava.add(s.getInstance)))
248237
instances.toArray
249238
}
250239

0 commit comments

Comments
 (0)