support percent display show when dataX execution

datavane · Apr 15, 2024 · 3ad6010 · 3ad6010
1 parent 37ed518
commit 3ad6010
Show file tree

Hide file tree

Showing 5 changed files with 48 additions and 7 deletions.
diff --git a/tis-assemble/src/main/java/com/qlangtech/tis/rpc/server/FullBuildStatCollectorServer.java b/tis-assemble/src/main/java/com/qlangtech/tis/rpc/server/FullBuildStatCollectorServer.java
@@ -98,7 +98,6 @@ public void run() {
 
  @Override
  public void loadPhaseStatusFromLatest(PSynResTarget request, StreamObserver<PPhaseStatusCollection> responseObserver) {
- // super.loadPhaseStatusFromLatest(request, responseObserver);
  boolean pipeline = request.getPipeline();
  String targetResName = request.getName();
  if (!pipeline) {
@@ -107,7 +106,7 @@ public void loadPhaseStatusFromLatest(PSynResTarget request, StreamObserver<PPha
  }
 
  PhaseStatusCollection statusCollection = DefaultChainContext.loadPhaseStatusFromLatest(targetResName);
- responseObserver.onNext(statusCollection != null ? LogCollectorClient.convertPP(statusCollection) : null);
+ responseObserver.onNext((statusCollection != null) ? LogCollectorClient.convertPP(statusCollection) : null);
  responseObserver.onCompleted();
  }
 

diff --git a/tis-assemble/src/test/java/StartAssembleWeb.java b/tis-assemble/src/test/java/StartAssembleWeb.java
@@ -16,9 +16,6 @@
  * limitations under the License.
  */
 
-import com.alibaba.fastjson.JSONArray;
-import com.alibaba.fastjson.JSONObject;
-import com.qlangtech.tis.common.utils.Assert;
 import com.qlangtech.tis.datax.TimeFormat;
 import com.qlangtech.tis.manage.common.CenterResource;
 import com.qlangtech.tis.manage.common.Config;
@@ -45,7 +42,7 @@ public class StartAssembleWeb extends TestCase {
 
  public void testStart() throws Exception {
 
- System.out.println( this.getClass().getClassLoader().loadClass("com.alibaba.datax.common.statistics.PerfTrace"));
+ System.out.println(this.getClass().getClassLoader().loadClass("com.alibaba.datax.common.statistics.PerfTrace"));
 
  // System.setProperty(com.qlangtech.tis.fullbuild.indexbuild.IRemoteTaskTrigger.KEY_DELTA_STREM_DEBUG, "true");
  System.setProperty(DataxUtils.EXEC_TIMESTAMP, String.valueOf(TimeFormat.getCurrentTimeStamp()));

diff --git a/tis-hadoop-rpc/src/main/java/com/qlangtech/tis/rpc/server/IncrStatusClient.java b/tis-hadoop-rpc/src/main/java/com/qlangtech/tis/rpc/server/IncrStatusClient.java
@@ -175,7 +175,7 @@ public PhaseStatusCollection loadPhaseStatusFromLatest(SynResTarget resTarget) {
  builder.setPipeline(resTarget.isPipeline());
  builder.setName(resTarget.getName());
  PPhaseStatusCollection statusCollection = logCollectorBlockingStub.loadPhaseStatusFromLatest(builder.build());
- if (statusCollection == null) {
+ if (statusCollection == null || statusCollection.getTaskId() < 1) {
  return null;
  }
  return LogCollectorClient.convert(statusCollection, ExecutePhaseRange.fullRange());

diff --git a/tis-hadoop-rpc/src/main/java/com/tis/hadoop/rpc/StatusRpcClientFactory.java b/tis-hadoop-rpc/src/main/java/com/tis/hadoop/rpc/StatusRpcClientFactory.java
@@ -30,6 +30,7 @@
 import com.qlangtech.tis.realtime.yarn.rpc.LaunchReportInfo;
 import com.qlangtech.tis.realtime.yarn.rpc.MasterJob;
 import com.qlangtech.tis.realtime.yarn.rpc.PingResult;
+import com.qlangtech.tis.realtime.yarn.rpc.SynResTarget;
 import com.qlangtech.tis.realtime.yarn.rpc.UpdateCounterMap;
 import com.qlangtech.tis.rpc.grpc.log.DefaultLoggerAppenderClient;
 import com.qlangtech.tis.rpc.grpc.log.ILogReporter;
@@ -502,6 +503,11 @@ private static class MockIncrStatusUmbilicalProtocol implements IncrStatusUmbili
  public void nodeLaunchReport(LaunchReportInfo launchReportInfo) {
  }
 
+ @Override
+ public PhaseStatusCollection loadPhaseStatusFromLatest(SynResTarget resTarget) {
+ return null;
+ }
+
  @Override
  public void close() throws IOException {
  }

diff --git a/tis-hadoop-rpc/src/test/java/com/tis/hadoop/rpc/TestStatusRpcClientFactory.java b/tis-hadoop-rpc/src/test/java/com/tis/hadoop/rpc/TestStatusRpcClientFactory.java
@@ -0,0 +1,39 @@
+/**
+ * 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
+ * <p>
+ * http:https://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 com.tis.hadoop.rpc;
+
+import com.qlangtech.tis.cloud.ITISCoordinator;
+import com.qlangtech.tis.common.utils.Assert;
+import com.qlangtech.tis.fullbuild.phasestatus.PhaseStatusCollection;
+import com.qlangtech.tis.realtime.yarn.rpc.SynResTarget;
+import com.tis.hadoop.rpc.StatusRpcClientFactory.AssembleSvcCompsite;
+import junit.framework.TestCase;
+
+/**
+ * @author: 百岁（[email protected]）
+ * @create: 2024-04-15 14:08
+ **/
+public class TestStatusRpcClientFactory extends TestCase {
+ public void testLoadPhaseStatusFromLatest() throws Exception {
+ RpcServiceReference ref = StatusRpcClientFactory.getService(ITISCoordinator.create());
+ AssembleSvcCompsite svc = ref.get();
+ PhaseStatusCollection statusCollection = svc.statReceiveSvc.loadPhaseStatusFromLatest(SynResTarget.pipeline("mysql5"));
+ // Assert.assertNotNull("statusCollection can not be null", statusCollection);
+ }
+}