You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by xx...@apache.org on 2023/01/04 02:34:54 UTC
[kylin] branch kylin5 updated (2e29cffe56 -> c0290399fc)
This is an automated email from the ASF dual-hosted git repository.
xxyu pushed a change to branch kylin5
in repository https://gitbox.apache.org/repos/asf/kylin.git
from 2e29cffe56 KYLIN-5367 fix updata layoutHits error in fusion model
new 676fdf7691 KYLIN-5370 DDL operations ignore table name case
new aaa1debe72 KYLIN-5369 Fix job is running, the job fails after the KE restart and failure information is displayed for several times
new c0290399fc KYLIN-5368 Fix when stop ke job on yarn are not be killed for yarn cluster mode
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../rest/ISmartApplicationListenerForSystem.java} | 17 +++---
.../rest/config/initialize/BootstrapCommand.java | 6 ++-
.../config/initialize/EpochChangedListener.java | 5 +-
.../org/apache/kylin/common/msg/CnMessage.java | 2 +-
.../job/impl/threadpool/NDefaultScheduler.java | 8 +++
.../apache/kylin/job/runners/FetcherRunner.java | 8 +++
.../apache/kylin/metadata/epoch/EpochManager.java | 5 ++
.../kylin/metadata/epoch/EpochManagerTest.java | 22 ++++++++
.../org/apache/kylin/rest/service/JobService.java | 34 +++++++++++-
.../apache/kylin/rest/service/SparkDDLTest.java | 2 +-
.../apache/kylin/rest/service/JobServiceTest.java | 61 ++++++++++++++++++++++
.../org/apache/kylin/rest/BootstrapServer.java | 9 +++-
.../apache/kylin/spark/ddl/SourceTableCheck.java | 14 +++--
13 files changed, 173 insertions(+), 20 deletions(-)
copy src/{server/src/main/java/org/apache/kylin/rest/ProxyUriTemplateHandler.java => common-service/src/main/java/org/apache/kylin/rest/ISmartApplicationListenerForSystem.java} (62%)
[kylin] 02/03: KYLIN-5369 Fix job is running, the job fails after the KE restart and failure information is displayed for several times
Posted by xx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xxyu pushed a commit to branch kylin5
in repository https://gitbox.apache.org/repos/asf/kylin.git
commit aaa1debe72f3f9ab915b54b4bb4c5d399842852d
Author: sibingzhang <74...@users.noreply.github.com>
AuthorDate: Tue Nov 1 15:51:43 2022 +0800
KYLIN-5369 Fix job is running, the job fails after the KE restart and failure information is displayed for several times
---
.../org/apache/kylin/rest/config/initialize/BootstrapCommand.java | 6 ++++--
.../apache/kylin/rest/config/initialize/EpochChangedListener.java | 5 +++--
.../org/apache/kylin/job/impl/threadpool/NDefaultScheduler.java | 8 ++++++++
.../src/main/java/org/apache/kylin/job/runners/FetcherRunner.java | 8 ++++++++
4 files changed, 23 insertions(+), 4 deletions(-)
diff --git a/src/common-service/src/main/java/org/apache/kylin/rest/config/initialize/BootstrapCommand.java b/src/common-service/src/main/java/org/apache/kylin/rest/config/initialize/BootstrapCommand.java
index 955b59c3cb..f5a682bb8f 100644
--- a/src/common-service/src/main/java/org/apache/kylin/rest/config/initialize/BootstrapCommand.java
+++ b/src/common-service/src/main/java/org/apache/kylin/rest/config/initialize/BootstrapCommand.java
@@ -29,6 +29,8 @@ import org.springframework.stereotype.Component;
import lombok.val;
import lombok.extern.slf4j.Slf4j;
+import java.util.concurrent.atomic.AtomicBoolean;
+
@Slf4j
@Component
public class BootstrapCommand implements Runnable {
@@ -50,8 +52,8 @@ public class BootstrapCommand implements Runnable {
}
void initProject(KylinConfig config, final ProjectInstance project) {
+ NDefaultScheduler scheduler = NDefaultScheduler.getInstance(project.getName());
EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
- NDefaultScheduler scheduler = NDefaultScheduler.getInstance(project.getName());
scheduler.init(new JobEngineConfig(config));
if (!scheduler.hasStarted()) {
throw new RuntimeException("Scheduler for " + project.getName() + " has not been started");
@@ -59,7 +61,7 @@ public class BootstrapCommand implements Runnable {
return 0;
}, project.getName(), 1, UnitOfWork.DEFAULT_EPOCH_ID);
-
+ scheduler.setHasFinishedTransactions(new AtomicBoolean(true));
log.info("init project {} finished", project.getName());
}
}
diff --git a/src/common-service/src/main/java/org/apache/kylin/rest/config/initialize/EpochChangedListener.java b/src/common-service/src/main/java/org/apache/kylin/rest/config/initialize/EpochChangedListener.java
index e86435d6d3..6165322f40 100644
--- a/src/common-service/src/main/java/org/apache/kylin/rest/config/initialize/EpochChangedListener.java
+++ b/src/common-service/src/main/java/org/apache/kylin/rest/config/initialize/EpochChangedListener.java
@@ -18,6 +18,7 @@
package org.apache.kylin.rest.config.initialize;
import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.persistence.ResourceStore;
@@ -93,8 +94,8 @@ public class EpochChangedListener {
}
log.info("start thread of project: {}", project);
+ NDefaultScheduler scheduler = NDefaultScheduler.getInstance(project);
EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
- NDefaultScheduler scheduler = NDefaultScheduler.getInstance(project);
scheduler.init(new JobEngineConfig(kylinConfig));
if (!scheduler.hasStarted()) {
throw new RuntimeException("Scheduler for " + project + " has not been started");
@@ -104,7 +105,6 @@ public class EpochChangedListener {
if (!ss.getHasStarted().get()) {
throw new RuntimeException("Streaming Scheduler for " + project + " has not been started");
}
-
QueryHistoryTaskScheduler qhAccelerateScheduler = QueryHistoryTaskScheduler.getInstance(project);
qhAccelerateScheduler.init();
@@ -115,6 +115,7 @@ public class EpochChangedListener {
recommendationUpdateScheduler.addProject(project);
return 0;
}, project, 1);
+ scheduler.setHasFinishedTransactions(new AtomicBoolean(true));
} else {
//TODO need global leader
CreateAdminUserUtils.createAllAdmins(userService, env);
diff --git a/src/core-job/src/main/java/org/apache/kylin/job/impl/threadpool/NDefaultScheduler.java b/src/core-job/src/main/java/org/apache/kylin/job/impl/threadpool/NDefaultScheduler.java
index 8297b92239..d3efe67bb9 100644
--- a/src/core-job/src/main/java/org/apache/kylin/job/impl/threadpool/NDefaultScheduler.java
+++ b/src/core-job/src/main/java/org/apache/kylin/job/impl/threadpool/NDefaultScheduler.java
@@ -29,6 +29,7 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import lombok.Setter;
import org.apache.commons.lang3.RandomUtils;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.util.ExecutorServiceUtil;
@@ -69,6 +70,9 @@ public class NDefaultScheduler implements Scheduler<AbstractExecutable> {
private ExecutableContext context;
private AtomicBoolean initialized = new AtomicBoolean(false);
private AtomicBoolean hasStarted = new AtomicBoolean(false);
+
+ @Setter
+ private AtomicBoolean hasFinishedTransactions = new AtomicBoolean(false);
@Getter
private JobEngineConfig jobEngineConfig;
@Getter
@@ -226,6 +230,10 @@ public class NDefaultScheduler implements Scheduler<AbstractExecutable> {
return hasStarted.get();
}
+ public boolean hasFinishedTransactions() {
+ return hasFinishedTransactions.get();
+ }
+
public static double currentAvailableMem() {
return 1.0 * memoryRemaining.availablePermits();
}
diff --git a/src/core-job/src/main/java/org/apache/kylin/job/runners/FetcherRunner.java b/src/core-job/src/main/java/org/apache/kylin/job/runners/FetcherRunner.java
index 93f9d89bdc..ab659c48a8 100644
--- a/src/core-job/src/main/java/org/apache/kylin/job/runners/FetcherRunner.java
+++ b/src/core-job/src/main/java/org/apache/kylin/job/runners/FetcherRunner.java
@@ -45,6 +45,8 @@ public class FetcherRunner extends AbstractDefaultSchedulerRunner {
private final ScheduledExecutorService fetcherPool;
+ private boolean reSchedule = true;
+
public FetcherRunner(NDefaultScheduler nDefaultScheduler, ExecutorService jobPool,
ScheduledExecutorService fetcherPool) {
super(nDefaultScheduler);
@@ -95,6 +97,12 @@ public class FetcherRunner extends AbstractDefaultSchedulerRunner {
@Override
public void doRun() {
try {
+ // Job schedule is only limited to the transaction in the NDefaultScheduler once
+ // Avoid that if the first transaction fails, jobs in the project cannot be scheduled
+ if (!nDefaultScheduler.hasFinishedTransactions() && reSchedule) {
+ reSchedule = false;
+ return;
+ }
val executableManager = NExecutableManager.getInstance(KylinConfig.getInstanceFromEnv(), project);
Map<String, Executable> runningJobs = context.getRunningJobs();
[kylin] 03/03: KYLIN-5368 Fix when stop ke job on yarn are not be killed for yarn cluster mode
Posted by xx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xxyu pushed a commit to branch kylin5
in repository https://gitbox.apache.org/repos/asf/kylin.git
commit c0290399fcd514b05aafdc792dd01b9773176f06
Author: sibingzhang <74...@users.noreply.github.com>
AuthorDate: Tue Nov 1 15:51:55 2022 +0800
KYLIN-5368 Fix when stop ke job on yarn are not be killed for yarn cluster mode
---
.../rest/ISmartApplicationListenerForSystem.java | 32 ++++++++++++
.../apache/kylin/metadata/epoch/EpochManager.java | 5 ++
.../kylin/metadata/epoch/EpochManagerTest.java | 22 ++++++++
.../org/apache/kylin/rest/service/JobService.java | 34 +++++++++++-
.../apache/kylin/rest/service/JobServiceTest.java | 61 ++++++++++++++++++++++
.../org/apache/kylin/rest/BootstrapServer.java | 9 +++-
6 files changed, 160 insertions(+), 3 deletions(-)
diff --git a/src/common-service/src/main/java/org/apache/kylin/rest/ISmartApplicationListenerForSystem.java b/src/common-service/src/main/java/org/apache/kylin/rest/ISmartApplicationListenerForSystem.java
new file mode 100644
index 0000000000..8fc5f026e7
--- /dev/null
+++ b/src/common-service/src/main/java/org/apache/kylin/rest/ISmartApplicationListenerForSystem.java
@@ -0,0 +1,32 @@
+/*
+ * 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
+ *
+ * http://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.kylin.rest;
+
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.context.ApplicationEvent;
+import org.springframework.context.event.ContextClosedEvent;
+import org.springframework.context.event.SmartApplicationListener;
+
+public interface ISmartApplicationListenerForSystem extends SmartApplicationListener {
+
+ @Override
+ default boolean supportsEventType(Class<? extends ApplicationEvent> eventType) {
+ return eventType == ApplicationReadyEvent.class || eventType == ContextClosedEvent.class;
+ }
+
+}
\ No newline at end of file
diff --git a/src/core-metadata/src/main/java/org/apache/kylin/metadata/epoch/EpochManager.java b/src/core-metadata/src/main/java/org/apache/kylin/metadata/epoch/EpochManager.java
index bb7f2c3d2f..cbe1ba9590 100644
--- a/src/core-metadata/src/main/java/org/apache/kylin/metadata/epoch/EpochManager.java
+++ b/src/core-metadata/src/main/java/org/apache/kylin/metadata/epoch/EpochManager.java
@@ -814,4 +814,9 @@ public class EpochManager {
return null;
});
}
+
+ public List<Epoch> getOwnedEpochs() {
+ return epochStore.list().stream().filter(this::checkEpochOwnerOnly).collect(Collectors.toList());
+ }
+
}
diff --git a/src/core-metadata/src/test/java/org/apache/kylin/metadata/epoch/EpochManagerTest.java b/src/core-metadata/src/test/java/org/apache/kylin/metadata/epoch/EpochManagerTest.java
index 164d952020..d488dfe67b 100644
--- a/src/core-metadata/src/test/java/org/apache/kylin/metadata/epoch/EpochManagerTest.java
+++ b/src/core-metadata/src/test/java/org/apache/kylin/metadata/epoch/EpochManagerTest.java
@@ -214,6 +214,28 @@ class EpochManagerTest {
}
+ @Test
+ void testGetOwnedEpochs() {
+ Epoch e1 = new Epoch();
+ e1.setEpochTarget("test1");
+ e1.setCurrentEpochOwner("owner1");
+
+ Epoch e2 = new Epoch();
+ e2.setEpochTarget("test2");
+ e2.setCurrentEpochOwner("owner2");
+
+ Epoch e3 = new Epoch();
+ e3.setEpochTarget("test3");
+ e3.setCurrentEpochOwner("owner2");
+
+ getEpochStore().insertBatch(Arrays.asList(e1, e2, e3));
+
+ EpochManager epochManager = EpochManager.getInstance();
+ epochManager.setIdentity("owner2");
+
+ Assertions.assertEquals(2, epochManager.getOwnedEpochs().size());
+ }
+
@Test
void testForceUpdateEpoch() {
EpochManager epochManager = EpochManager.getInstance();
diff --git a/src/data-loading-service/src/main/java/org/apache/kylin/rest/service/JobService.java b/src/data-loading-service/src/main/java/org/apache/kylin/rest/service/JobService.java
index 08ef663424..22424681c5 100644
--- a/src/data-loading-service/src/main/java/org/apache/kylin/rest/service/JobService.java
+++ b/src/data-loading-service/src/main/java/org/apache/kylin/rest/service/JobService.java
@@ -47,6 +47,7 @@ import java.util.stream.Stream;
import javax.servlet.http.HttpServletRequest;
+import org.apache.kylin.metadata.epoch.EpochManager;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.io.IOUtils;
@@ -66,6 +67,7 @@ import org.apache.kylin.common.metrics.MetricsGroup;
import org.apache.kylin.common.metrics.MetricsName;
import org.apache.kylin.common.msg.Message;
import org.apache.kylin.common.msg.MsgPicker;
+import org.apache.kylin.common.persistence.metadata.Epoch;
import org.apache.kylin.common.persistence.transaction.UnitOfWork;
import org.apache.kylin.common.persistence.transaction.UnitOfWorkContext;
import org.apache.kylin.common.scheduler.EventBusFactory;
@@ -107,6 +109,7 @@ import org.apache.kylin.metadata.model.Segments;
import org.apache.kylin.metadata.model.TableDesc;
import org.apache.kylin.metadata.project.EnhancedUnitOfWork;
import org.apache.kylin.metadata.project.ProjectInstance;
+import org.apache.kylin.rest.ISmartApplicationListenerForSystem;
import org.apache.kylin.rest.constant.Constant;
import org.apache.kylin.rest.request.JobFilter;
import org.apache.kylin.rest.request.JobUpdateRequest;
@@ -122,6 +125,8 @@ import org.apache.kylin.rest.util.SparkHistoryUIUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationEvent;
+import org.springframework.context.event.ContextClosedEvent;
import org.springframework.http.HttpHeaders;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.stereotype.Component;
@@ -141,7 +146,7 @@ import lombok.val;
import lombok.var;
@Component("jobService")
-public class JobService extends BasicService implements JobSupporter {
+public class JobService extends BasicService implements JobSupporter, ISmartApplicationListenerForSystem {
@Autowired
private ProjectService projectService;
@@ -1352,6 +1357,33 @@ public class JobService extends BasicService implements JobSupporter {
MsgPicker.setMsg("en");
}
}
+
+ @Override
+ public void onApplicationEvent(ApplicationEvent event) {
+ if (event instanceof ContextClosedEvent) {
+ logger.info("Stop kyligence node, kill job on yarn for yarn cluster mode");
+ EpochManager epochManager = EpochManager.getInstance();
+ KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
+ List<Epoch> ownedEpochs = epochManager.getOwnedEpochs();
+
+ for (Epoch epoch : ownedEpochs) {
+ String project = epoch.getEpochTarget();
+ NExecutableManager executableManager = NExecutableManager.getInstance(kylinConfig, project);
+ if (executableManager != null) {
+ List<ExecutablePO> allJobs = executableManager.getAllJobs();
+ for (ExecutablePO executablePO : allJobs) {
+ executableManager.cancelRemoteJob(executablePO);
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ public int getOrder() {
+ return HIGHEST_PRECEDENCE;
+ }
+
@Setter
@Getter
static class ExecutablePOSortBean {
diff --git a/src/job-service/src/test/java/org/apache/kylin/rest/service/JobServiceTest.java b/src/job-service/src/test/java/org/apache/kylin/rest/service/JobServiceTest.java
index ee6db5a9be..71266afd4c 100644
--- a/src/job-service/src/test/java/org/apache/kylin/rest/service/JobServiceTest.java
+++ b/src/job-service/src/test/java/org/apache/kylin/rest/service/JobServiceTest.java
@@ -57,11 +57,14 @@ import java.util.stream.Collectors;
import javax.servlet.http.HttpServletRequest;
+import org.apache.kylin.metadata.epoch.EpochManager;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.fs.Path;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.exception.KylinException;
+import org.apache.kylin.common.persistence.metadata.Epoch;
+import org.apache.kylin.common.persistence.metadata.EpochStore;
import org.apache.kylin.common.persistence.transaction.TransactionException;
import org.apache.kylin.common.persistence.transaction.UnitOfWork;
import org.apache.kylin.common.util.ClassUtil;
@@ -89,6 +92,7 @@ import org.apache.kylin.job.execution.JobTypeEnum;
import org.apache.kylin.job.execution.NExecutableManager;
import org.apache.kylin.job.execution.StageBase;
import org.apache.kylin.job.execution.SucceedChainedTestExecutable;
+import org.apache.kylin.job.execution.SucceedDagTestExecutable;
import org.apache.kylin.job.execution.SucceedTestExecutable;
import org.apache.kylin.metadata.cube.model.NBatchConstants;
import org.apache.kylin.metadata.cube.model.NDataflowManager;
@@ -125,7 +129,9 @@ import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.context.ApplicationEvent;
import org.springframework.context.annotation.ClassPathScanningCandidateComponentProvider;
+import org.springframework.context.event.ContextClosedEvent;
import org.springframework.core.type.filter.AssignableTypeFilter;
import org.springframework.http.HttpHeaders;
import org.springframework.security.authentication.TestingAuthenticationToken;
@@ -177,6 +183,9 @@ public class JobServiceTest extends NLocalFileMetadataTestCase {
@Mock
private ProjectService projectService = Mockito.spy(ProjectService.class);
+ @Mock
+ private ApplicationEvent applicationEvent = Mockito.mock(ContextClosedEvent.class);
+
@Rule
public ExpectedException thrown = ExpectedException.none();
@@ -1901,4 +1910,56 @@ public class JobServiceTest extends NLocalFileMetadataTestCase {
Assert.assertEquals(0.0, executable.getProgress(), 0);
Assert.assertEquals("", executable.getRelatedSegment());
}
+
+ @Test
+ public void tstOnApplicationEvent() {
+ final String PROJECT1 = "test1";
+ final String PROJECT2 = "test2";
+ final String PROJECT3 = "test3";
+ KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
+ NExecutableManager manager1 = NExecutableManager.getInstance(kylinConfig, PROJECT2);
+
+ Epoch e1 = new Epoch();
+ e1.setEpochTarget(PROJECT1);
+ e1.setCurrentEpochOwner("owner1");
+
+ Epoch e2 = new Epoch();
+ e2.setEpochTarget(PROJECT2);
+ e2.setCurrentEpochOwner("owner2");
+
+ Epoch e3 = new Epoch();
+ e3.setEpochTarget(PROJECT3);
+ e3.setCurrentEpochOwner("owner2");
+
+ try {
+ EpochStore.getEpochStore(kylinConfig).insertBatch(Arrays.asList(e1, e2, e3));
+ } catch (Exception e) {
+ throw new RuntimeException("cannnot init epoch store!");
+ }
+
+ EpochManager epochManager = EpochManager.getInstance();
+ epochManager.setIdentity("owner2");
+
+ val job = new DefaultExecutable();
+ job.setProject(PROJECT2);
+ job.setJobType(JobTypeEnum.INDEX_BUILD);
+
+ val executable1 = new SucceedDagTestExecutable();
+ executable1.setProject(PROJECT2);
+ job.addTask(executable1);
+
+ val executable2 = new SucceedDagTestExecutable();
+ executable2.setProject(PROJECT2);
+ job.addTask(executable2);
+
+ executable1.setNextSteps(Sets.newHashSet(executable2.getId()));
+ executable2.setPreviousStep(executable1.getId());
+
+ val executablePO = NExecutableManager.toPO(job, PROJECT2);
+ manager1.addJob(executablePO);
+ manager1.updateJobOutput(job.getId(), ExecutableState.RUNNING);
+
+ jobService.onApplicationEvent(applicationEvent);
+ }
+
}
diff --git a/src/server/src/main/java/org/apache/kylin/rest/BootstrapServer.java b/src/server/src/main/java/org/apache/kylin/rest/BootstrapServer.java
index c18973ed27..474ccad787 100644
--- a/src/server/src/main/java/org/apache/kylin/rest/BootstrapServer.java
+++ b/src/server/src/main/java/org/apache/kylin/rest/BootstrapServer.java
@@ -38,7 +38,6 @@ import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
import org.springframework.cloud.loadbalancer.annotation.LoadBalancerClient;
import org.springframework.cloud.zookeeper.discovery.ZookeeperInstance;
import org.springframework.context.ApplicationEvent;
-import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ImportResource;
import org.springframework.context.event.ContextClosedEvent;
@@ -60,7 +59,7 @@ import lombok.val;
@EnableDiscoveryClient
@LoadBalancerClient(name = "spring-boot-provider", configuration = org.apache.kylin.rest.LoadBalanced.class)
@EnableSpringHttpSession
-public class BootstrapServer implements ApplicationListener<ApplicationEvent> {
+public class BootstrapServer implements ISmartApplicationListenerForSystem {
private static final Logger logger = LoggerFactory.getLogger(BootstrapServer.class);
@@ -135,4 +134,10 @@ public class BootstrapServer implements ApplicationListener<ApplicationEvent> {
EpochManager.getInstance().releaseOwnedEpochs();
}
}
+
+ @Override
+ public int getOrder() {
+ return LOWEST_PRECEDENCE;
+ }
+
}
[kylin] 01/03: KYLIN-5370 DDL operations ignore table name case
Posted by xx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xxyu pushed a commit to branch kylin5
in repository https://gitbox.apache.org/repos/asf/kylin.git
commit 676fdf7691e530052ef6f11d67a21335307e64ec
Author: ChenLiang <31...@users.noreply.github.com>
AuthorDate: Mon Oct 31 21:14:21 2022 +0800
KYLIN-5370 DDL operations ignore table name case
---
.../main/java/org/apache/kylin/common/msg/CnMessage.java | 2 +-
.../java/org/apache/kylin/rest/service/SparkDDLTest.java | 2 +-
.../java/org/apache/kylin/spark/ddl/SourceTableCheck.java | 14 ++++++++++----
3 files changed, 12 insertions(+), 6 deletions(-)
diff --git a/src/core-common/src/main/java/org/apache/kylin/common/msg/CnMessage.java b/src/core-common/src/main/java/org/apache/kylin/common/msg/CnMessage.java
index 3306e85266..af50c20b53 100644
--- a/src/core-common/src/main/java/org/apache/kylin/common/msg/CnMessage.java
+++ b/src/core-common/src/main/java/org/apache/kylin/common/msg/CnMessage.java
@@ -1757,7 +1757,7 @@ public class CnMessage extends Message {
@Override
public String getDDLUnSupported() {
- return "不支持的 DDL 语法，仅支持 `create view`, `drop view`, `alter view`, `show create view` 语法";
+ return "不支持的 DDL 语法，仅支持 `create view`, `drop view`, `alter view`, `show create table` 语法";
}
@Override
diff --git a/src/datasource-service/src/test/java/org/apache/kylin/rest/service/SparkDDLTest.java b/src/datasource-service/src/test/java/org/apache/kylin/rest/service/SparkDDLTest.java
index df4af60863..fbdb21528f 100644
--- a/src/datasource-service/src/test/java/org/apache/kylin/rest/service/SparkDDLTest.java
+++ b/src/datasource-service/src/test/java/org/apache/kylin/rest/service/SparkDDLTest.java
@@ -45,7 +45,7 @@ public class SparkDDLTest extends NLocalFileMetadataTestCase {
private final IUserGroupService userGroupService = Mockito.spy(NUserGroupService.class);
private static final String CREATEVIEW_SQL1 =
- "CREATE VIEW `ssb`.`ke_order_view` as select LO_ORDERKEY, C_NAME from SSB.P_LINEORDER t1 left join "
+ "CREATE VIEW `ssb`.`ke_order_view` as select LO_ORDERKEY, C_NAME from SSB.p_lineorder t1 left join "
+ "SSB. CUSTOMER t2 on t1. LO_CUSTKEY = t2. C_CUSTKEY";
private static final String CREATEVIEW_SQL2 = "CREATE VIEW `ssb`.`order_view2` as select * from SSB.P_LINEORDER";
private static final String CREATEVIEW_SQL3 = "CREATE VIEW `ssb`.`order_view2` as abc";
diff --git a/src/spark-project/spark-ddl-plugin/src/main/java/org/apache/kylin/spark/ddl/SourceTableCheck.java b/src/spark-project/spark-ddl-plugin/src/main/java/org/apache/kylin/spark/ddl/SourceTableCheck.java
index c3c54623e5..beb157bddc 100644
--- a/src/spark-project/spark-ddl-plugin/src/main/java/org/apache/kylin/spark/ddl/SourceTableCheck.java
+++ b/src/spark-project/spark-ddl-plugin/src/main/java/org/apache/kylin/spark/ddl/SourceTableCheck.java
@@ -17,6 +17,9 @@
*/
package org.apache.kylin.spark.ddl;
+import java.util.List;
+import java.util.stream.Collectors;
+
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.msg.MsgPicker;
import org.apache.kylin.metadata.model.ISourceAware;
@@ -62,12 +65,15 @@ public class SourceTableCheck implements DDLCheck {
for (LogicalPlan plan : scala.collection.JavaConverters.seqAsJavaListConverter(relationLeaves).asJava()) {
if (plan instanceof UnresolvedRelation) {
val tableName = ((UnresolvedRelation) plan).tableName();
- TableDesc tableDesc = tableManager.getTableDesc(tableName);
- if (tableDesc == null) {
+ List<TableDesc> loadTable = tableManager.listAllTables().stream()
+ .filter(table -> table.getTableAlias().equalsIgnoreCase(tableName))
+ .collect(Collectors.toList());
+ if (loadTable.isEmpty()) {
throwException(MsgPicker.getMsg().getDDLTableNotLoad(tableName));
}
- if (ISourceAware.ID_HIVE != tableDesc.getSourceType()
- && ISourceAware.ID_SPARK != tableDesc.getSourceType()) {
+ TableDesc table = loadTable.get(0);
+ if (ISourceAware.ID_HIVE != table.getSourceType()
+ && ISourceAware.ID_SPARK != table.getSourceType()) {
throwException(MsgPicker.getMsg().getDDLTableNotSupport(tableName));
}
}