[CELEBORN-1540] Simplify Tez plugin implementation and improve integr…

…ation testing Co-authored-by: hongguangwei <[email protected]>
apache · Oct 23, 2024 · 1937914 · 1937914
1 parent 4d5c7e1
commit 1937914
Show file tree

Hide file tree

Showing 34 changed files with 1,313 additions and 3,908 deletions.
diff --git a/client-tez/tez-shaded/pom.xml b/client-tez/tez-shaded/pom.xml
@@ -38,25 +38,6 @@
 
   <build>
     <plugins>
-      <plugin>
-        <groupId>org.apache.maven.plugins</groupId>
-        <artifactId>maven-assembly-plugin</artifactId>
-        <configuration>
-          <descriptors>
-            <descriptor>src/main/resources/assembly.xml</descriptor>
-          </descriptors>
-          <appendAssemblyId>false</appendAssemblyId>
-        </configuration>
-        <executions>
-          <execution>
-            <id>make-assembly</id>
-            <goals>
-              <goal>single</goal>
-            </goals>
-            <phase>package</phase>
-          </execution>
-        </executions>
-      </plugin>
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-shade-plugin</artifactId>
@@ -69,31 +50,22 @@
             <relocation>
               <pattern>com.google.common</pattern>
               <shadedPattern>${shading.prefix}.com.google.common</shadedPattern>
+              <excludes>
+                <exclude>com.google.common.util.concurrent.*</exclude>
+              </excludes>
             </relocation>
             <relocation>
               <pattern>io.netty</pattern>
               <shadedPattern>${shading.prefix}.io.netty</shadedPattern>
             </relocation>
             <relocation>
-              <pattern>org.scala-lang</pattern>
-              <shadedPattern>${shading.prefix}.org.scala-lang</shadedPattern>
+              <pattern>org.apache.commons</pattern>
+              <shadedPattern>${shading.prefix}.org.apache.commons</shadedPattern>
             </relocation>
             <relocation>
               <pattern>org.roaringbitmap</pattern>
               <shadedPattern>${shading.prefix}.org.roaringbitmap</shadedPattern>
             </relocation>
-            <relocation>
-              <pattern>org.apache.commons.lang3</pattern>
-              <shadedPattern>${shading.prefix}.org.apache.commons.lang3</shadedPattern>
-            </relocation>
-            <relocation>
-              <pattern>org.apache.tez.runtime.library.shuffle.impl.ShuffleUserPayloads</pattern>
-              <shadedPattern>${shading.prefix}.org.apache.tez.runtime.library.shuffle.impl.ShuffleUserPayloads</shadedPattern>
-            </relocation>
-            <relocation>
-              <pattern>org.apache.tez.common.TezCommonUtils</pattern>
-              <shadedPattern>${shading.prefix}.org.apache.tez.common.TezCommonUtils</shadedPattern>
-            </relocation>
           </relocations>
           <artifactSet>
             <includes>
@@ -103,10 +75,10 @@
               <include>com.google.guava:failureaccess</include>
               <include>io.netty:*</include>
               <include>org.apache.commons:commons-lang3</include>
+              <include>org.roaringbitmap:RoaringBitmap</include>
               <include>org.scala-lang:scala-library</include>
+              <include>org.scala-lang:scala-reflect</include>
               <include>org.lz4:lz4-java</include>
-              <include>com.github.luben:zstd-jni</include>
-              <include>org.roaringbitmap:RoaringBitmap</include>
             </includes>
           </artifactSet>
           <filters>

diff --git a/client-tez/tez/pom.xml b/client-tez/tez/pom.xml
@@ -188,6 +188,11 @@
         </exclusion>
       </exclusions>
     </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
 </project>
diff --git a/client-tez/tez/src/main/java/org/apache/celeborn/tez/plugin/util/CelebornTezUtils.java b/client-tez/tez/src/main/java/org/apache/celeborn/tez/plugin/util/CelebornTezUtils.java
@@ -32,7 +32,6 @@ public class CelebornTezUtils {
   public static final String TEZ_PREFIX = "tez.";
   public static final String TEZ_CELEBORN_LM_HOST = "celeborn.lifecycleManager.host";
   public static final String TEZ_CELEBORN_LM_PORT = "celeborn.lifecycleManager.port";
-  public static final String TEZ_CELEBORN_USER = "celeborn.lifecycleManager.user";
   public static final String TEZ_CELEBORN_APPLICATION_ID = "celeborn.applicationId";
   public static final String TEZ_SHUFFLE_ID = "celeborn.tez.shuffle.id";
   public static final String TEZ_BROADCAST_OR_ONETOONE = "celeborn.tez.broadcastOrOneToOne";
@@ -59,6 +58,26 @@ public static Object getPrivateField(Object object, String name) {
     }
   }
 
+  public static Object getParentPrivateField(Object object, String name) {
+    try {
+      Field f = object.getClass().getSuperclass().getDeclaredField(name);
+      f.setAccessible(true);
+      return f.get(object);
+    } catch (Exception e) {
+      throw new CelebornRuntimeException(e.getMessage(), e);
+    }
+  }
+
+  public static void setParentPrivateField(Object object, String name, Object value) {
+    try {
+      Field f = object.getClass().getSuperclass().getDeclaredField(name);
+      f.setAccessible(true);
+      f.set(object, value);
+    } catch (Exception e) {
+      throw new CelebornRuntimeException(e.getMessage(), e);
+    }
+  }
+
   public static String uniqueIdentifierToAttemptId(String uniqueIdentifier) {
     if (uniqueIdentifier == null) {
       throw new CelebornRuntimeException("uniqueIdentifier should not be null");

diff --git a/client-tez/tez/src/main/java/org/apache/tez/dag/app/CelebornDagAppMaster.java b/client-tez/tez/src/main/java/org/apache/tez/dag/app/CelebornDagAppMaster.java
@@ -30,9 +30,6 @@
 
 import scala.Tuple2;
 
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.GnuParser;
-import org.apache.commons.cli.Options;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.UserGroupInformation;
@@ -191,9 +188,6 @@ protected DAG createDAG(DAGProtos.DAGPlan dagPB) {
                   edgeSourceConf.set(CelebornTezUtils.TEZ_CELEBORN_LM_HOST, lifecycleManagerHost);
                   edgeSourceConf.setInt(
                       CelebornTezUtils.TEZ_CELEBORN_LM_PORT, lifecycleManagerPort);
-                  edgeSourceConf.set(
-                      CelebornTezUtils.TEZ_CELEBORN_USER,
-                      lifecycleManager.getUserIdentifier().toString());
                   edgeSourceConf.set(
                       CelebornTezUtils.TEZ_BROADCAST_OR_ONETOONE,
                       String.valueOf(broadCastOrOneToOne));
@@ -333,9 +327,19 @@ public void serviceStop() throws Exception {
 
   public static void main(String[] args) {
     try {
-      String s = "org.apache.tez.runtime.library.output.OrderedPartitionedKVOutput";
-      if (s.equalsIgnoreCase(OrderedPartitionedKVOutput.class.getName())) {
-        System.out.println("this is truth----");
+
+      boolean sessionModeCliOption = false;
+      for (int i = 0; i < args.length; i++) {
+        if (args[i].startsWith("-D")) {
+          String[] property = args[i].split("=");
+          if (property.length < 2) {
+            System.setProperty(property[0].substring(2), "");
+          } else {
+            System.setProperty(property[0].substring(2), property[1]);
+          }
+        } else if (args[i].contains("--session") || args[i].contains("-s")) {
+          sessionModeCliOption = true;
+        }
       }
 
       Thread.setDefaultUncaughtExceptionHandler(new YarnUncaughtExceptionHandler());
@@ -360,16 +364,6 @@ public static void main(String[] args) {
 
       String jobUserName = System.getenv(ApplicationConstants.Environment.USER.name());
 
-      // Command line options
-      Options opts = new Options();
-      opts.addOption(
-          TezConstants.TEZ_SESSION_MODE_CLI_OPTION,
-          false,
-          "Run Tez Application Master in Session mode");
-
-      CommandLine cliParser = new GnuParser().parse(opts, args);
-      boolean sessionModeCliOption = cliParser.hasOption(TezConstants.TEZ_SESSION_MODE_CLI_OPTION);
-
       Logger.info(
           "Creating CelebornDAGAppMaster for "
               + "applicationId={}"