lanyuanxiaoyao/hudi-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(command): 修复没有使用推送队列的方式启动压缩

Browse Source
This commit is contained in:
v-zhangjc9
2024-03-19 16:33:08 +08:00
parent 0b5cea7053
commit 59acf837ad
1 changed files with 5 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
service-command/src/main/java/com/lanyuanxiaoyao/service/command/commands/CompactionCommand.java
View File

@@ -5,10 +5,8 @@ import com.lanyuanxiaoyao.service.command.utils.CommandLineUtils;
import com.lanyuanxiaoyao.service.command.utils.TableUtils; import com.lanyuanxiaoyao.service.command.utils.TableUtils;
import com.lanyuanxiaoyao.service.common.Constants; import com.lanyuanxiaoyao.service.common.Constants;
import com.lanyuanxiaoyao.service.common.entity.TableMeta; import com.lanyuanxiaoyao.service.common.entity.TableMeta;
import com.lanyuanxiaoyao.service.configuration.entity.yarn.YarnClusters;
import com.lanyuanxiaoyao.service.forest.service.InfoService; import com.lanyuanxiaoyao.service.forest.service.InfoService;
import com.lanyuanxiaoyao.service.forest.service.launcher.LauncherService; import com.lanyuanxiaoyao.service.forest.service.ScheduleService;
import com.lanyuanxiaoyao.service.forest.service.launcher.LaunchersService;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.shell.standard.ShellComponent; import org.springframework.shell.standard.ShellComponent;
@@ -28,12 +26,11 @@ public class CompactionCommand extends AbstractUtilShellComponent {
private final static String RUN_CONFIRMATION_MESSAGE = "操作将会提交压缩任务, 请再次确认操作"; private final static String RUN_CONFIRMATION_MESSAGE = "操作将会提交压缩任务, 请再次确认操作";
private final InfoService infoService; private final InfoService infoService;
private final LauncherService launcherService; private final ScheduleService scheduleService;
public CompactionCommand(InfoService infoService, YarnClusters yarnClusters, LaunchersService launchersService) { public CompactionCommand(InfoService infoService, ScheduleService scheduleService) {
this.infoService = infoService; this.infoService = infoService;
logger.info("Default sync cluster: {}", yarnClusters.getDefaultSyncCluster()); this.scheduleService = scheduleService;
this.launcherService = launchersService.getService(yarnClusters.getDefaultSyncCluster());
} }
@ShellMethod("启动表压缩任务") @ShellMethod("启动表压缩任务")
@@ -53,7 +50,7 @@ public class CompactionCommand extends AbstractUtilShellComponent {
} else { } else {
System.out.println(TableUtils.makeTableMeta(meta)); System.out.println(TableUtils.makeTableMeta(meta));
if (doubleCheck(RUN_CONFIRMATION_MESSAGE, ignoreCheck)) { if (doubleCheck(RUN_CONFIRMATION_MESSAGE, ignoreCheck)) {
launcherService.compactionStop(meta.getJob().getId(), meta.getAlias()); scheduleService.scheduleTable(meta.getJob().getId(), meta.getAlias());
return Constants.OPERATION_DONE; return Constants.OPERATION_DONE;
} else { } else {
return Constants.OPERATION_CANCEL; return Constants.OPERATION_CANCEL;