feat: add interface guidance for enabling HDFS HA and YARN HA

bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/enums/Command.java

@@ -39,6 +39,8 @@ public enum Command {
     CHECK("check", "Check"),
     CONFIGURE("configure", "Configure"),
     CUSTOM("custom", "Custom"),
+    ENABLE_YARN_RM_HA("enable-yarn-rm-ha", "EnableYarnRmHa"),
+    ENABLE_HDFS_HA("enable-hdfs-ha", "EnableHdfsHa"),
 
     // Internal use only, not available for API call
     INIT("init", "Init"),

bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/factory/component/ComponentConfigureJobFactory.java

@@ -0,0 +1,48 @@
+/*
+ * 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
+ *
+ *    https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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 org.apache.bigtop.manager.server.command.factory.component;
+
+import org.apache.bigtop.manager.common.enums.Command;
+import org.apache.bigtop.manager.server.command.CommandIdentifier;
+import org.apache.bigtop.manager.server.command.job.Job;
+import org.apache.bigtop.manager.server.command.job.JobContext;
+import org.apache.bigtop.manager.server.command.job.component.ComponentConfigureJob;
+import org.apache.bigtop.manager.server.enums.CommandLevel;
+
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+import org.springframework.context.annotation.Scope;
+import org.springframework.stereotype.Component;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Component
+@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
+public class ComponentConfigureJobFactory extends AbstractComponentJobFactory {
+
+    @Override
+    public CommandIdentifier getCommandIdentifier() {
+        return new CommandIdentifier(CommandLevel.COMPONENT, Command.CONFIGURE);
+    }
+
+    @Override
+    public Job createJob(JobContext jobContext) {
+        return new ComponentConfigureJob(jobContext);
+    }
+}

bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/factory/component/ComponentCustomJobFactory.java

@@ -0,0 +1,48 @@
+/*
+ * 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
+ *
+ *    https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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 org.apache.bigtop.manager.server.command.factory.component;
+
+import org.apache.bigtop.manager.common.enums.Command;
+import org.apache.bigtop.manager.server.command.CommandIdentifier;
+import org.apache.bigtop.manager.server.command.job.Job;
+import org.apache.bigtop.manager.server.command.job.JobContext;
+import org.apache.bigtop.manager.server.command.job.component.ComponentCustomJob;
+import org.apache.bigtop.manager.server.enums.CommandLevel;
+
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+import org.springframework.context.annotation.Scope;
+import org.springframework.stereotype.Component;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Component
+@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
+public class ComponentCustomJobFactory extends AbstractComponentJobFactory {
+
+    @Override
+    public CommandIdentifier getCommandIdentifier() {
+        return new CommandIdentifier(CommandLevel.COMPONENT, Command.CUSTOM);
+    }
+
+    @Override
+    public Job createJob(JobContext jobContext) {
+        return new ComponentCustomJob(jobContext);
+    }
+}

bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/factory/service/EnableHdfsHaJobFactory.java

@@ -0,0 +1,75 @@
+/*
+ * 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
+ *
+ *    https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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 org.apache.bigtop.manager.server.command.factory.service;
+
+import org.apache.bigtop.manager.common.enums.Command;
+import org.apache.bigtop.manager.server.command.CommandIdentifier;
+import org.apache.bigtop.manager.server.command.job.Job;
+import org.apache.bigtop.manager.server.command.job.JobContext;
+import org.apache.bigtop.manager.server.command.job.service.EnableHdfsHaJob;
+import org.apache.bigtop.manager.server.enums.CommandLevel;
+
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+import org.springframework.context.annotation.Scope;
+import org.springframework.stereotype.Component;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Component
+@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
+public class EnableHdfsHaJobFactory extends AbstractServiceJobFactory {
+
+    @Override
+    public CommandIdentifier getCommandIdentifier() {
+        return new CommandIdentifier(CommandLevel.SERVICE, Command.ENABLE_HDFS_HA);
+    }
+
+    @Override
+    public Job createJob(JobContext jobContext) {
+        log.info(
+                "EnableHdfsHaJobFactory creating job, jobFactoryClassSource={}, jobClassSource={}, stageClassSource={}, taskClassSource={}, commandDTO.command={}, commandDTO.customCommandLen={}",
+                getCodeSource(EnableHdfsHaJobFactory.class),
+                getCodeSource(EnableHdfsHaJob.class),
+                getCodeSource(org.apache.bigtop.manager.server.command.stage.ComponentCustomStage.class),
+                getCodeSource(org.apache.bigtop.manager.server.command.task.ComponentCustomTask.class),
+                jobContext == null || jobContext.getCommandDTO() == null
+                        ? null
+                        : jobContext.getCommandDTO().getCommand(),
+                jobContext == null
+                                || jobContext.getCommandDTO() == null
+                                || jobContext.getCommandDTO().getCustomCommand() == null
+                        ? null
+                        : jobContext.getCommandDTO().getCustomCommand().length());
+        return new EnableHdfsHaJob(jobContext);
+    }
+
+    private static String getCodeSource(Class<?> clazz) {
+        try {
+            if (clazz == null
+                    || clazz.getProtectionDomain() == null
+                    || clazz.getProtectionDomain().getCodeSource() == null) {
+                return "null";
+            }
+            return String.valueOf(clazz.getProtectionDomain().getCodeSource().getLocation());
+        } catch (Exception e) {
+            return "error:" + e.getMessage();
+        }
+    }
+}

bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/factory/service/EnableYarnRmHaJobFactory.java

@@ -0,0 +1,48 @@
+/*
+ * 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
+ *
+ *    https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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 org.apache.bigtop.manager.server.command.factory.service;
+
+import org.apache.bigtop.manager.common.enums.Command;
+import org.apache.bigtop.manager.server.command.CommandIdentifier;
+import org.apache.bigtop.manager.server.command.job.Job;
+import org.apache.bigtop.manager.server.command.job.JobContext;
+import org.apache.bigtop.manager.server.command.job.service.EnableYarnRmHaJob;
+import org.apache.bigtop.manager.server.enums.CommandLevel;
+
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+import org.springframework.context.annotation.Scope;
+import org.springframework.stereotype.Component;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Component
+@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
+public class EnableYarnRmHaJobFactory extends AbstractServiceJobFactory {
+
+    @Override
+    public CommandIdentifier getCommandIdentifier() {
+        return new CommandIdentifier(CommandLevel.SERVICE, Command.ENABLE_YARN_RM_HA);
+    }
+
+    @Override
+    public Job createJob(JobContext jobContext) {
+        return new EnableYarnRmHaJob(jobContext);
+    }
+}

bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/helper/JobCacheHelper.java

@@ -44,6 +44,8 @@
 import org.apache.bigtop.manager.server.model.dto.StackDTO;
 import org.apache.bigtop.manager.server.utils.StackUtils;
 
+import org.apache.commons.lang3.StringUtils;
+
 import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.HashMap;
@@ -75,6 +77,12 @@ private static void initialize() {
     }
 
     public static void sendJobCache(Long jobId, List<String> hostnames) {
+        sendJobCache(jobId, hostnames, null);
+    }
+
+    public static void sendJobCache(
+            Long jobId, List<String> hostnames, Map<String, List<String>> overrideComponentHosts) {
+        final Map<String, List<String>> finalOverrideComponentHosts = overrideComponentHosts;
         if (!INITIALIZED.get()) {
             initialize();
         }
@@ -102,6 +110,7 @@ public static void sendJobCache(Long jobId, List<String> hostnames) {
                 JobCachePayload copiedPayload =
                         JsonUtils.readFromString(JsonUtils.writeAsString(payload), JobCachePayload.class);
                 genClusterPayload(copiedPayload, clusterId);
+                mergeOverrideComponentHosts(copiedPayload, finalOverrideComponentHosts);
                 JobCacheRequest request = JobCacheRequest.newBuilder()
                         .setJobId(jobId)
                         .setPayload(JsonUtils.writeAsString(copiedPayload))
@@ -162,6 +171,31 @@ private static void genClusterPayload(JobCachePayload payload, Long clusterId) {
         payload.setHosts(hosts);
     }
 
+    private static void mergeOverrideComponentHosts(
+            JobCachePayload payload, Map<String, List<String>> overrideComponentHosts) {
+        if (overrideComponentHosts == null || overrideComponentHosts.isEmpty()) {
+            return;
+        }
+        if (payload.getComponentHosts() == null) {
+            payload.setComponentHosts(new HashMap<>());
+        }
+        final Map<String, List<String>> target = payload.getComponentHosts();
+
+        overrideComponentHosts.forEach((component, hosts) -> {
+            if (StringUtils.isBlank(component) || hosts == null || hosts.isEmpty()) {
+                return;
+            }
+            List<String> filtered = hosts.stream()
+                    .filter(StringUtils::isNotBlank)
+                    .map(String::trim)
+                    .distinct()
+                    .toList();
+            if (!filtered.isEmpty()) {
+                target.put(component.toLowerCase(), new ArrayList<>(filtered));
+            }
+        });
+    }
+
     private static void genGlobalPayload(JobCachePayload payload) {
         List<RepoPO> repoPOList = repoDao.findAll();
         Map<String, Map<String, String>> serviceConfigMap = getServiceConfigMap(0L);
@@ -220,13 +254,11 @@ private static Map<String, List<String>> getComponentHostMap(Long clusterId) {
         List<ComponentPO> componentPOList = componentDao.findByQuery(query);
         Map<String, List<String>> hostMap = new HashMap<>();
         componentPOList.forEach(x -> {
-            if (hostMap.containsKey(x.getName())) {
-                hostMap.get(x.getName()).add(x.getHostname());
-            } else {
-                List<String> list = new ArrayList<>();
-                list.add(x.getHostname());
-                hostMap.put(x.getName(), list);
+            if (StringUtils.isBlank(x.getHostname())) {
+                return;
             }
+            hostMap.computeIfAbsent(x.getName(), k -> new ArrayList<>())
+                    .add(x.getHostname().trim());
         });
 
         return hostMap;

bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/job/AbstractJob.java

@@ -42,6 +42,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.LinkedBlockingQueue;
 
 @Slf4j
@@ -75,6 +76,10 @@ public AbstractJob(JobContext jobContext) {
         if (CollectionUtils.isEmpty(stages)) {
             throw new ApiException(ApiExceptionEnum.JOB_HAS_NO_STAGES);
         }
+
+        for (Stage stage : stages) {
+            stage.init();
+        }
     }
 
     protected void injectBeans() {
@@ -92,6 +97,26 @@ protected void beforeCreateStages() {
 
     protected abstract void createStages();
 
+    protected Map<String, List<String>> buildOverrideComponentHosts() {
+        Map<String, List<String>> m = new java.util.HashMap<>();
+        if (jobContext == null
+                || jobContext.getCommandDTO() == null
+                || jobContext.getCommandDTO().getComponentCommands() == null) {
+            return m;
+        }
+        for (org.apache.bigtop.manager.server.model.dto.command.ComponentCommandDTO cc :
+                jobContext.getCommandDTO().getComponentCommands()) {
+            if (cc == null
+                    || cc.getComponentName() == null
+                    || cc.getHostnames() == null
+                    || cc.getHostnames().isEmpty()) {
+                continue;
+            }
+            m.put(cc.getComponentName().toLowerCase(), cc.getHostnames());
+        }
+        return m;
+    }
+
     @Override
     public void beforeRun() {
         jobPO.setState(JobState.PROCESSING.getName());
@@ -113,7 +138,7 @@ public void run() {
                     .flatMap(List::stream)
                     .distinct()
                     .toList();
-            JobCacheHelper.sendJobCache(jobPO.getId(), hostnames);
+            JobCacheHelper.sendJobCache(jobPO.getId(), hostnames, buildOverrideComponentHosts());
 
             LinkedBlockingQueue<Stage> queue = new LinkedBlockingQueue<>(stages);
             while (!queue.isEmpty()) {