Reduce mixed language module linkismange #4463 #4509
Conversation
CharlieYan24
changed the title
|
Test result of Spark sql engine job is following, job launch log : Job result: |
|
Test result of Shell engine job is following, job launch log : Job result: |
|
Sorry for failed the spotless test, Already fixed it. |
CharlieYan24
force-pushed
the
dev-1.4.0-reduce-mixlanguage-module-linkismange
branch
from
abc6e56 to
4a1a413
Compare
|
Modified code and passed maven build with parameters: -Pspark-2.4 -Phadoop-2.7. |
| int fromLine = getAs(parameters, "fromLine", 1); | ||
| boolean enableTail = getAs(parameters, "enableTail", false); | ||
| if (lastRows > EngineConnLogOperator.MAX_LOG_FETCH_SIZE.getValue()) { | ||
| throw new WarnException( |
There was a problem hiding this comment.
Should use ECMErrorException
There was a problem hiding this comment.
Use ECMErrorException instead of WarnException.Thanks for your good advice.
| skippedLine += 1; | ||
| } else { | ||
| if (rowIgnore) { | ||
| Matcher matcher = linePattern.matcher(line); |
There was a problem hiding this comment.
line pattern needs to judge null
There was a problem hiding this comment.
Modified line pattern to avoid NullPointException. Thanks for your good advice.
| return resultMap; | ||
| } catch (IOException e) { | ||
| // ing | ||
| throw new RuntimeException(e); |
There was a problem hiding this comment.
Used ECMErrorException instead of RuntimeException. Thanks for your good advice.
| line = randomAndReversedReadLine(randomReader, reversedReader); | ||
| } | ||
|
|
||
| IOUtils.closeQuietly(randomReader); |
There was a problem hiding this comment.
Already remove repeat code, Thanks .
|
|
||
| File logPath = new File(engineConnLogDir, logType); | ||
| if (!logPath.exists() || !logPath.isFile()) { | ||
| throw new WarnException( |
There was a problem hiding this comment.
Should use ECMErrorException
There was a problem hiding this comment.
use ECMErrorException instead of WarnException.Thanks for your good advice.
|
|
||
| import java.util.List; | ||
|
|
||
| public class RequestExpectedResourceAndWait { |
There was a problem hiding this comment.
Already remove RequestExpectedResourceAndWait class.
|
|
||
| import java.util.List; | ||
|
|
||
| public class RequestExpectedResource { |
There was a problem hiding this comment.
Already remove RequestExpectedResource class,Thanks.
|
|
||
| import java.util.List; | ||
|
|
||
| public class RequestExpectedResource { |
There was a problem hiding this comment.
Already remove RequestExpectedResource class.
| @@ -15,6 +15,6 @@ | |||
| * limitations under the License. | |||
| */ | |||
|
|
|||
| package org.apache.linkis.manager.common.protocol | |||
| package org.apache.linkis.manager.common.protocol; | |||
|
|
|||
There was a problem hiding this comment.
Already remove ServiceState class,Thanks.
|
|
||
| package org.apache.linkis.manager.common.protocol; | ||
|
|
||
| public abstract class ServiceHealthReport { |
There was a problem hiding this comment.
Already remove ServiceHealthReport class.
| RMLabelContainer rMLabelContainer = labelResourceService.enrichLabels(engineNode.getLabels()); | ||
|
|
||
| PersistenceResource persistenceResource = | ||
| labelResourceService.getPersistenceResource(rMLabelContainer.getEngineInstanceLabel()); |
There was a problem hiding this comment.
The ticketID can be obtained through ecResourceInfo.getTicketId, check the resource if it is only null
There was a problem hiding this comment.
Because ecResourceInfo is already been null in above line, so that I get Ticket id from labelResourceService ,please correct me if I was wrong ,thanks.
|
|
||
| String engineAskAsyncId = getAsyncId(); | ||
| CompletableFuture<EngineNode> createNodeThread = | ||
| CompletableFuture.supplyAsync( |
There was a problem hiding this comment.
need to add Executors (thread pools)
There was a problem hiding this comment.
Already add Executors (thread pools), Thanks.
| "%s ticketID:%s Failed to initialize engine, reason: %s ", | ||
| engineNode.getServiceInstance(), resourceTicketId, errorInfo.getKey())); | ||
| } | ||
| throw new LinkisRetryException( |
There was a problem hiding this comment.
should to use AMErrorException
There was a problem hiding this comment.
Already used AMErrorException instead of LinkisRetryException ,Thanks for your good advice.
| } catch (Throwable t) { | ||
| logger.info( | ||
| "Failed to reuse engineConn time taken " + (System.currentTimeMillis() - startTime), t); | ||
| throw new RuntimeException(t); |
There was a problem hiding this comment.
need to use linkisRunitmeException
There was a problem hiding this comment.
Already use AMErrorException instead ,thanks for your time.
| stopEngine(engineStopRequest, Sender.getSender(Sender.getThisServiceInstance())), | ||
| String.format("async stop engineFailed %s", engineStopRequest), | ||
| logger); | ||
| }); |
There was a problem hiding this comment.
Already add add executor, Thanks for your excellent advice.
| String.format("asyncStopEngineWithUpdateMetrics with error: %s", e.getMessage()), | ||
| e); | ||
| } | ||
| }); |
There was a problem hiding this comment.
Already add add executor, Thanks for your excellent advice.
| engineNodes.add(node); | ||
| } | ||
| } | ||
| dealECNodes(engineNodes); |
There was a problem hiding this comment.
add try catch for dealECNodes(engineNodes) method, thanks .
| */ | ||
| @Override | ||
| public void run() { | ||
| logger.info("Start to check the health of the node"); |
There was a problem hiding this comment.
add try catch for dealECNodes(engineNodes) method, thanks .
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| public class Utils { |
There was a problem hiding this comment.
It is recommended to move to common and modify it to LinkisUtils
There was a problem hiding this comment.
Already modify it to LinkisUtils, Thanks for your advice.
| EngineStopRequest stopEngineRequest = | ||
| new EngineStopRequest(engineNode.getServiceInstance(), ManagerUtils.getAdminUser()); | ||
| engineStopService.asyncStopEngine(stopEngineRequest); | ||
| throw new RuntimeException(t); |
There was a problem hiding this comment.
Should LinkisRuntimeException
There was a problem hiding this comment.
used LinkisRuntimie instead of RuntimeException, thanks.
| combinedLabelBuilder.build("", Lists.newArrayList(userCreatorLabel, engineTypeLabel)); | ||
| } catch (LabelErrorException e) { | ||
| logger.warn("getAllUserResource failed", e); | ||
| throw new RuntimeException(e); |
There was a problem hiding this comment.
used LinkisRuntimie instead of RuntimeException, thanks.
| } | ||
|
|
||
| @Override | ||
| public void run() { |
There was a problem hiding this comment.
Already added try catch inside of method run().
| () -> { | ||
| Pair<YarnResource, YarnResource> yarnResource = | ||
| getResources(rmWebAddress, realQueueName, queueName); | ||
| getResources(rmWebAddress, realQueueName, queueName); |
There was a problem hiding this comment.
removed duplicate code ,thanks for your good advice.
| return nodeResource; | ||
| }, | ||
| t -> { | ||
| throw new RuntimeException(t); |
There was a problem hiding this comment.
should we use linkis exception ?
There was a problem hiding this comment.
yeah , already used RMErrorException instead of RuntimeException. Thanks for your time.
| if (queueValue.isPresent()) { | ||
| JsonNode jsonNode = queueValue.get(); | ||
| double absoluteCapacity = jsonNode.path("absoluteCapacity").asDouble(); | ||
| double effectiveResource = absoluteCapacity; |
There was a problem hiding this comment.
can use absoluteCapacity directly?
There was a problem hiding this comment.
yes, already deleted line double effectiveResource = absoluteCapacity, and assign absoluteCapacity = jsonNode.path("absoluteCapacity").asDouble() directly.
| try { | ||
| JsonNode metrics = getResponseByUrl("metrics", rmWebAddress); | ||
| long totalMemory = metrics.path("clusterMetrics").path("totalMB").asLong(); | ||
| long totalCores = metrics.path("clusterMetrics").path("totalVirtualCores").asLong(); |
There was a problem hiding this comment.
can parse clusterMetrics once and used for totalMemory and totalCores ?
There was a problem hiding this comment.
Yeah , thanks for your good advice. I have modified it .
| } | ||
|
|
||
| public static JsonNode getChildQueues(JsonNode resp) { | ||
| JsonNode queues = resp.get("childQueues").get("queue"); |
There was a problem hiding this comment.
is queue sub-attribute of childQueues ?
There was a problem hiding this comment.
this method is translate from pre scala version like bellow:
There was a problem hiding this comment.
you can double confirm with actual json values
There was a problem hiding this comment.
already modified getChildQueues method，thanks.
| String schedulerType = resp.path("scheduler").path("schedulerInfo").path("type").asText(); | ||
| if ("capacityScheduler".equals(schedulerType)) { | ||
| realQueueName = queueName; | ||
| JsonNode childQueues = getChildQueuesOfCapacity(resp.path("scheduler").path("schedulerInfo")); |
There was a problem hiding this comment.
can we parse resp.path("scheduler").path("schedulerInfo") once and used latter ?
There was a problem hiding this comment.
Yeah , thanks for your good advice. I have modified it .
| String id = app.get("id").asText(); | ||
| String user = app.get("user").asText(); | ||
| String applicationType = app.get("applicationType").asText(); | ||
| Optional<YarnResource> yarnResource = getYarnResource(Optional.ofNullable(app), queueName); |
There was a problem hiding this comment.
does this getYarnResource work ?
There was a problem hiding this comment.
Method getYarnResource works ,I trace this method return using arthas when I submited spark task.
There was a problem hiding this comment.
app not memory and cores?
There was a problem hiding this comment.
Sorry for my misunderstanding ,I added method getAllocatedYarnResource to fix above problem.
| } | ||
| JsonNode jsonNode = null; | ||
| try { | ||
| jsonNode = objectMapper.readTree(entityString); |
There was a problem hiding this comment.
did you test with three level child queue , not sure if it works on fairscheduler
There was a problem hiding this comment.
I haven't tested three level child queue yet, I would tested in next week and reply the result ,thanks for your time.
There was a problem hiding this comment.
Already tested with three level child queue with following unit test and fairscheduler config
unit test:
fairscheduler config:
There was a problem hiding this comment.
you can try with multi child queue, for examples, you can define root.production.productionchild1,root.production.productionchild2,root.production.productionchild3, and test whether all this child queues parsed correctly.
There was a problem hiding this comment.
Already passed three level child queue parse test.
-
the following is fair-scheduler.xml
-
here is the submit parameters
-
the following is child queues parsing result
| } | ||
| return Optional.empty(); | ||
| } catch (Exception e) { | ||
| // e.printStackTrace(); |
There was a problem hiding this comment.
Already added print log.
| String id = app.get("id").asText(); | ||
| String user = app.get("user").asText(); | ||
| String applicationType = app.get("applicationType").asText(); | ||
| Optional<YarnResource> yarnResource = getYarnResource(Optional.ofNullable(app), queueName); |
There was a problem hiding this comment.
app not memory and cores?
| httpResponse = response; | ||
| } | ||
|
|
||
| ObjectMapper objectMapper = new ObjectMapper(); |
There was a problem hiding this comment.
Already use static var instead.
| } catch (Throwable t) { | ||
| if (t instanceof FatalException) { | ||
| logger.error("Fatal error, system exit...", t); | ||
| throw (FatalException) t; |
There was a problem hiding this comment.
need to add System.exit(fatal.getErrCode)?
There was a problem hiding this comment.
need to add
···
case e: VirtualMachineError =>
logger.error("Fatal error, system exit...", e)
System.exit(-1)
throw e
case exp
if (null != exp.getCause && (exp.getCause.isInstanceOf[FatalException] || exp.getCause
.isInstanceOf[VirtualMachineError])) =>
logger.error("Caused by fatal error, system exit...", exp)
System.exit(-1)
throw exp
case er: Error =>
logger.error("Throw error", er)
throw er
case t => catchOp(t)
···
There was a problem hiding this comment.
need to add
···
case e: VirtualMachineError =>
logger.error("Fatal error, system exit...", e)
System.exit(-1)
throw e
case exp
if (null != exp.getCause && (exp.getCause.isInstanceOf[FatalException] || exp.getCause
.isInstanceOf[VirtualMachineError])) =>
logger.error("Caused by fatal error, system exit...", exp)
System.exit(-1)
throw exp
case er: Error =>
logger.error("Throw error", er)
throw er
case t => catchOp(t)
···
There was a problem hiding this comment.
Already added the above exception.
| t -> { | ||
| if (t instanceof ErrorException) { | ||
| ErrorException error = (ErrorException) t; | ||
| log.warn( |
There was a problem hiding this comment.
Use log.error instead ,thanks for your time.
| } catch (WarnException t) { | ||
| WarnException warn = (WarnException) t; | ||
| log.warn( | ||
| "Warning code（警告码）: {}, Warning message（警告信息）: {}.", warn.getErrCode(), warn.getDesc()); |
There was a problem hiding this comment.
Is this method the corresponding org.apache.linkis.common.utils.Utils#tryAndErrorMsg?
There was a problem hiding this comment.
Yes, this method correspond to org.apache.linkis.common.utils.Utils#tryAndErrorMsg.
linkis-engineconn-plugins/pom.xml
Outdated
| <module>python</module> | ||
| <module>openlookeng</module> | ||
| <module>io_file</module> | ||
| <!-- <module>python</module>--> |
There was a problem hiding this comment.
Uncommented, thanks for your good advice.
What is the purpose of the change
Reduce mixlanguage module linkismange. Translate linkis-application-manager and linkis-manager-common module from Scala to Java.
Related issues/PRs
Related issues: #4463
Related pr: #4463
Brief change log
Checklist