This is an automated email from the ASF dual-hosted git repository. liyang-kylin pushed a commit to branch kylin5 in repository https://gitbox.apache.org/repos/asf/kylin.git
commit 7140baa80bb69eb06b70bedae2a28692ac541f92 Author: Li Yang <[email protected]> AuthorDate: Sat Nov 7 13:53:09 2026 +0800 KYLIN-6120 harden streaming job termination --- .../kylin/rest/service/ModelSemanticHelper.java | 16 +++++++ .../kylin/rest/service/ModelServiceTest.java | 16 +++++++ .../org/apache/kylin/streaming/util/JobKiller.java | 31 ++++++++++---- .../apache/kylin/streaming/util/JobKillerTest.java | 50 +++++++++++++++++++++- 4 files changed, 103 insertions(+), 10 deletions(-) diff --git a/src/modeling-service/src/main/java/org/apache/kylin/rest/service/ModelSemanticHelper.java b/src/modeling-service/src/main/java/org/apache/kylin/rest/service/ModelSemanticHelper.java index 1f10a19091..17546f7c36 100644 --- a/src/modeling-service/src/main/java/org/apache/kylin/rest/service/ModelSemanticHelper.java +++ b/src/modeling-service/src/main/java/org/apache/kylin/rest/service/ModelSemanticHelper.java @@ -17,6 +17,7 @@ */ package org.apache.kylin.rest.service; +import static org.apache.kylin.common.exception.ServerErrorCode.INVALID_PARAMETER; import static org.apache.kylin.common.exception.code.ErrorCodeServer.SIMPLIFIED_MEASURES_MISSING_ID; import java.io.IOException; @@ -32,6 +33,7 @@ import java.util.Map; import java.util.Objects; import java.util.Optional; import java.util.Set; +import java.util.UUID; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.BiConsumer; import java.util.function.Consumer; @@ -145,6 +147,7 @@ public class ModelSemanticHelper extends BasicService { } public NDataModel convertToDataModel(ModelRequest modelRequest) { + validateModelUuid(modelRequest.getUuid()); List<SimplifiedMeasure> simplifiedMeasures = modelRequest.getSimplifiedMeasures(); NDataModel dataModel; try { @@ -177,6 +180,19 @@ public class ModelSemanticHelper extends BasicService { return dataModel; } + private void validateModelUuid(String modelUuid) { + if (modelUuid == null) { + return; + } + try { + if (!UUID.fromString(modelUuid).toString().equalsIgnoreCase(modelUuid)) { + throw new IllegalArgumentException("Non-canonical UUID"); + } + } catch (IllegalArgumentException e) { + throw new KylinException(INVALID_PARAMETER, "Invalid model UUID."); + } + } + /** * expand model request, add hidden internal measures from current model * diff --git a/src/modeling-service/src/test/java/org/apache/kylin/rest/service/ModelServiceTest.java b/src/modeling-service/src/test/java/org/apache/kylin/rest/service/ModelServiceTest.java index 1761163098..8aca4ea771 100644 --- a/src/modeling-service/src/test/java/org/apache/kylin/rest/service/ModelServiceTest.java +++ b/src/modeling-service/src/test/java/org/apache/kylin/rest/service/ModelServiceTest.java @@ -19,6 +19,7 @@ package org.apache.kylin.rest.service; import static org.apache.kylin.common.exception.ServerErrorCode.FAILED_EXECUTE_MODEL_SQL; +import static org.apache.kylin.common.exception.ServerErrorCode.INVALID_PARAMETER; import static org.apache.kylin.common.exception.ServerErrorCode.INVALID_PARTITION_COLUMN; import static org.apache.kylin.common.exception.code.ErrorCodeServer.COMPUTED_COLUMN_CONFLICT; import static org.apache.kylin.common.exception.code.ErrorCodeServer.COMPUTED_COLUMN_CONFLICT_ADJUST_INFO; @@ -4706,6 +4707,21 @@ public class ModelServiceTest extends SourceTestCase { Assert.assertEquals(model.getAllNamedColumns().get(4).getName(), model1.getAllNamedColumns().get(4).getName()); } + @Test + public void testRejectNonCanonicalModelUuid() { + String project = "streaming_test"; + NDataModelManager modelManager = NDataModelManager.getInstance(KylinConfig.getInstanceFromEnv(), project); + NDataModel model = modelManager.getDataModelDesc("4965c827-fbb4-4ea1-a744-3f341a3b030d"); + ModelRequest modelRequest = new ModelRequest(model); + modelRequest.setProject(project); + modelRequest.setUuid("test';touch /tmp/not-executed;echo'"); + + KylinException exception = Assert.assertThrows(KylinException.class, + () -> semanticService.convertToDataModel(modelRequest)); + + Assert.assertEquals(INVALID_PARAMETER.toErrorCode(), exception.getErrorCode()); + } + @Test public void testUpdateModelColumns() { String project = "streaming_test"; diff --git a/src/streaming/src/main/java/org/apache/kylin/streaming/util/JobKiller.java b/src/streaming/src/main/java/org/apache/kylin/streaming/util/JobKiller.java index 1f33ceeb01..b7b67a545e 100644 --- a/src/streaming/src/main/java/org/apache/kylin/streaming/util/JobKiller.java +++ b/src/streaming/src/main/java/org/apache/kylin/streaming/util/JobKiller.java @@ -19,8 +19,8 @@ package org.apache.kylin.streaming.util; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; -import java.util.Locale; import org.apache.commons.lang3.StringUtils; import org.apache.kylin.cluster.ClusterManagerFactory; @@ -42,7 +42,7 @@ import lombok.val; public class JobKiller { private static final Logger logger = LoggerFactory.getLogger(JobKiller.class); - private static final String GREP_CMD = "ps -ef|grep '%s' | grep -v grep|awk '{print $2}'"; + private static final List<String> PROCESS_LIST_COMMAND = Arrays.asList("ps", "-eo", "pid=,args="); private static boolean isYarnEnv = StreamingUtils.isJobOnCluster(KylinConfig.getInstanceFromEnv()); @@ -156,17 +156,30 @@ public class JobKiller { } public static int grepProcess(CliCommandExecutor exec, StringLogger strLogger, String jobId) throws ShellException { - String cmd = String.format(Locale.getDefault(), GREP_CMD, jobId); - val result = exec.execute(cmd, strLogger).getCode(); - logger.info("grep process cmd={}, result ={} ", cmd, result); + StringLogger processOutput = new StringLogger(); + int result = exec.execute(PROCESS_LIST_COMMAND, null, processOutput, null).getCode(); + for (String line : processOutput.getContents()) { + String[] pidAndCommand = line.trim().split("\\s+", 2); + if (pidAndCommand.length == 2 && StringUtils.isNumeric(pidAndCommand[0]) + && pidAndCommand[1].contains(jobId)) { + if (strLogger != null) { + strLogger.log(pidAndCommand[0]); + } + } + } + logger.info("grep process jobId={}, result={} ", jobId, result); return result; } public static int doKillProcess(CliCommandExecutor exec, String jobId, boolean forced) throws ShellException { - String cmd = String.format(Locale.getDefault(), GREP_CMD, jobId); - val force = forced ? " -9" : StringUtils.EMPTY; - val result = exec.execute(cmd + "|xargs kill" + force, null).getCode(); - logger.info("kill process cmd={}, result ={} ", cmd, result); + StringLogger pidLogger = new StringLogger(); + int result = grepProcess(exec, pidLogger, jobId); + String signal = forced ? "-9" : "-TERM"; + for (String pid : pidLogger.getContents()) { + exec.execute(Arrays.asList("kill", signal, pid), null, null, jobId); + } + logger.info("kill process jobId={}, forced={}, pids={}, result={} ", jobId, forced, + pidLogger.getContents(), result); return result; } diff --git a/src/streaming/src/test/java/org/apache/kylin/streaming/util/JobKillerTest.java b/src/streaming/src/test/java/org/apache/kylin/streaming/util/JobKillerTest.java index 29e330076d..de0e6bbd34 100644 --- a/src/streaming/src/test/java/org/apache/kylin/streaming/util/JobKillerTest.java +++ b/src/streaming/src/test/java/org/apache/kylin/streaming/util/JobKillerTest.java @@ -18,8 +18,12 @@ package org.apache.kylin.streaming.util; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.List; +import java.util.Map; +import org.apache.commons.lang3.StringUtils; import org.apache.kylin.common.util.AddressUtil; import org.apache.kylin.common.util.CliCommandExecutor; import org.apache.kylin.common.util.Logger; @@ -204,7 +208,8 @@ public class JobKillerTest extends StreamingTestCase { val mgr = StreamingJobManager.getInstance(getTestConfig(), PROJECT); val streamingJobMeta = mgr.getStreamingJobByUuid(id); val exec = new CliCommandExecutor() { - public CliCmdExecResult execute(String command, Logger logAppender) throws ShellException { + public CliCmdExecResult execute(List<String> command, Map<String, String> environment, Logger logAppender, + String jobId) throws ShellException { throw new ShellException("test"); } }; @@ -227,6 +232,34 @@ public class JobKillerTest extends StreamingTestCase { } } + @Test + public void testGrepProcessDoesNotInterpretJobIdAsShell() throws Exception { + RecordingCliCommandExecutor exec = new RecordingCliCommandExecutor(); + String jobId = "test';touch /tmp/not-executed;echo'"; + exec.processOutput = " 123 java -jar app.jar " + jobId + "\n 456 java -jar other.jar\n"; + val strLogger = new JobKiller.StringLogger(); + + int resultCode = JobKiller.grepProcess(exec, strLogger, jobId); + + Assert.assertEquals(0, resultCode); + Assert.assertEquals(Collections.singletonList("123"), strLogger.getContents()); + Assert.assertEquals(Collections.singletonList(Arrays.asList("ps", "-eo", "pid=,args=")), exec.commands); + } + + @Test + public void testDoKillProcessUsesNumericPidArguments() throws Exception { + RecordingCliCommandExecutor exec = new RecordingCliCommandExecutor(); + String jobId = "test';touch /tmp/not-executed;echo'"; + exec.processOutput = " 123 java -jar app.jar " + jobId + "\n 456 java -jar other.jar\n"; + + int resultCode = JobKiller.doKillProcess(exec, jobId, true); + + Assert.assertEquals(0, resultCode); + Assert.assertEquals(2, exec.commands.size()); + Assert.assertEquals(Arrays.asList("ps", "-eo", "pid=,args="), exec.commands.get(0)); + Assert.assertEquals(Arrays.asList("kill", "-9", "123"), exec.commands.get(1)); + } + @Test public void testDoKillProcess() { val config = getTestConfig(); @@ -253,4 +286,19 @@ public class JobKillerTest extends StreamingTestCase { log.log("test"); Assert.assertFalse(log.getContents().isEmpty()); } + + private static class RecordingCliCommandExecutor extends CliCommandExecutor { + private final List<List<String>> commands = new ArrayList<>(); + private String processOutput = StringUtils.EMPTY; + + @Override + public CliCmdExecResult execute(List<String> command, Map<String, String> environment, Logger logAppender, + String jobId) { + commands.add(new ArrayList<>(command)); + if (logAppender != null && "ps".equals(command.get(0))) { + Arrays.stream(processOutput.split("\n")).forEach(logAppender::log); + } + return new CliCmdExecResult(0, StringUtils.EMPTY, "1"); + } + } }
