// 源码位置：seatunnel-engine/spark/src/main/java/org/apache/seatunnel/spark/SparkCommandArgs.java
@EqualsAndHashCode(callSuper = true)
@Data
public class SparkCommandArgs extends AbstractCommandArgs {

   @Parameter(
           names = {"-e", "--deploy-mode"},
           description = "Spark deploy mode, support [cluster, client]",
           converter = SparkDeployModeConverter.class)
   private DeployMode deployMode = DeployMode.CLIENT;

   @Parameter(
           names = {"-m", "--master"},
           description =
                   "Spark master, support [spark://host:port, mesos://host:port, yarn, "
                           + "k8s://https://host:port, local], default local[*]")
   private String master = "local[*]";

   @Override
   public Command<?> buildCommand() {
      Common.setDeployMode(getDeployMode());
      if (checkConfig) {
         return new SparkConfValidateCommand(this);
      }
      if (encrypt) {
         return new ConfEncryptCommand(this);
      }
      if (decrypt) {
         return new ConfDecryptCommand(this);
      }
      return new SparkTaskExecuteCommand(this);
   }
}

// 源码位置：seatunnel-plugin-discovery/src/main/java/org/apache/seatunnel/plugin/discovery/AbstractPluginDiscovery.java
public static List<Path> findPluginJars(Config config) {
    // 1. 从META-INF/seatunnel/plugins.index读取插件声明
    Enumeration<URL> indexes = ClassLoader.getSystemResources("META-INF/seatunnel/plugins.index");
    
    // 2. 递归解析传递依赖（通过pom.xml的<dependencies>）
    return resolveDependencies(indexes)
        .stream()
        .filter(jar -> !jar.contains("org.apache.seatunnel:seatunnel-core")) // 过滤核心包
        .collect(Collectors.toList());
}

// 插件加载隔离机制
public class PluginClassLoader extends URLClassLoader {
    private final String pluginName;
    
    public PluginClassLoader(String pluginName, URL[] urls, ClassLoader parent) {
        super(urls, parent);
        this.pluginName = pluginName;
    }
    
    @Override
    protected Class<?> loadClass(String name, boolean resolve) throws ClassNotFoundException {
        synchronized (getClassLoadingLock(name)) {
            // 优先从当前插件JAR加载类
            Class<?> c = findLoadedClass(name);
            if (c == null) {
                try {
                    c = findClass(name);
                } catch (ClassNotFoundException e) {
                    // 回退到父类加载器
                    c = super.loadClass(name, resolve);
                }
            }
            if (resolve) {
                resolveClass(c);
            }
            return c;
        }
    }
}

@Override
protected Factory loadPluginInstance(
        PluginIdentifier pluginIdentifier, ClassLoader classLoader) {
   ServiceLoader<Factory> serviceLoader =
           ServiceLoader.load(getPluginBaseClass(), classLoader);
   for (Factory factory : serviceLoader) {
      if (factoryClass.isInstance(factory)) {
         String factoryIdentifier = factory.factoryIdentifier();
         String pluginName = pluginIdentifier.getPluginName();
         if (StringUtils.equalsIgnoreCase(factoryIdentifier, pluginName)) {
            return factory;
         }
      }
   }
   return null;
}