This is an automated email from the ASF dual-hosted git repository.
liyang pushed a commit to branch kylin5
in repository https://gitbox.apache.org/repos/asf/kylin.git
commit 2695c5da83d9ccfb2a3177b9947e6eda4e6c1715
Author: sibingzhang <74443791+sibingzh...@users.noreply.github.com>
AuthorDate: Fri Aug 25 15:18:08 2023 +0800
KYLIN-5790 Security of kafka-clients
Co-authored-by: sibing.zhang <sibing.zh...@qq.com>
---
.../java/org/apache/kylin/rest/KafkaCondition.java | 32 ++++++++++++++++++++++
.../apache/kylin/rest/service/KafkaService.java | 8 ++++--
.../apache/kylin/rest/service/TableService.java | 2 +-
.../kylin/rest/controller/KafkaController.java | 2 +-
4 files changed, 39 insertions(+), 5 deletions(-)
diff --git
a/src/datasource-service/src/main/java/org/apache/kylin/rest/KafkaCondition.java
b/src/datasource-service/src/main/java/org/apache/kylin/rest/KafkaCondition.java
new file mode 100644
index 0000000000..c4388b8cd4
--- /dev/null
+++
b/src/datasource-service/src/main/java/org/apache/kylin/rest/KafkaCondition.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.apache.kylin.common.KylinConfig;
+import org.springframework.context.annotation.Condition;
+import org.springframework.context.annotation.ConditionContext;
+import org.springframework.core.type.AnnotatedTypeMetadata;
+
+public class KafkaCondition implements Condition {
+ @Override
+ public boolean matches(ConditionContext context, AnnotatedTypeMetadata
metadata) {
+ KylinConfig config = KylinConfig.getInstanceFromEnv();
+ return config.streamingEnabled();
+ }
+}
diff --git
a/src/datasource-service/src/main/java/org/apache/kylin/rest/service/KafkaService.java
b/src/datasource-service/src/main/java/org/apache/kylin/rest/service/KafkaService.java
index 2fec7df8cc..496f2f90ad 100644
---
a/src/datasource-service/src/main/java/org/apache/kylin/rest/service/KafkaService.java
+++
b/src/datasource-service/src/main/java/org/apache/kylin/rest/service/KafkaService.java
@@ -40,26 +40,28 @@ import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kylin.common.exception.KylinException;
import org.apache.kylin.common.msg.MsgPicker;
import org.apache.kylin.common.util.HadoopUtil;
+import org.apache.kylin.guava30.shaded.common.collect.Maps;
+import org.apache.kylin.guava30.shaded.common.collect.Sets;
import org.apache.kylin.loader.ParserClassLoaderState;
import org.apache.kylin.metadata.jar.JarInfoManager;
import org.apache.kylin.metadata.project.EnhancedUnitOfWork;
import org.apache.kylin.metadata.streaming.DataParserInfo;
import org.apache.kylin.metadata.streaming.DataParserManager;
import org.apache.kylin.metadata.streaming.KafkaConfig;
+import org.apache.kylin.rest.KafkaCondition;
import org.apache.kylin.rest.util.AclEvaluate;
import org.apache.kylin.sample.KafkaSourceHandler;
import org.apache.kylin.sample.StreamingSourceHandler;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Conditional;
import org.springframework.stereotype.Component;
-import org.apache.kylin.guava30.shaded.common.collect.Maps;
-
-import org.apache.kylin.guava30.shaded.common.collect.Sets;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
@Slf4j
@Component("kafkaService")
+@Conditional(KafkaCondition.class)
public class KafkaService extends BasicService {
@Autowired
diff --git
a/src/datasource-service/src/main/java/org/apache/kylin/rest/service/TableService.java
b/src/datasource-service/src/main/java/org/apache/kylin/rest/service/TableService.java
index b6498f1a6f..b2ae72706b 100644
---
a/src/datasource-service/src/main/java/org/apache/kylin/rest/service/TableService.java
+++
b/src/datasource-service/src/main/java/org/apache/kylin/rest/service/TableService.java
@@ -217,7 +217,7 @@ public class TableService extends BasicService {
@Autowired
private AccessService accessService;
- @Autowired
+ @Autowired(required = false)
@Qualifier("kafkaService")
private KafkaService kafkaService;
diff --git
a/src/metadata-server/src/main/java/org/apache/kylin/rest/controller/KafkaController.java
b/src/metadata-server/src/main/java/org/apache/kylin/rest/controller/KafkaController.java
index 163eb0de50..90981e0f8c 100644
---
a/src/metadata-server/src/main/java/org/apache/kylin/rest/controller/KafkaController.java
+++
b/src/metadata-server/src/main/java/org/apache/kylin/rest/controller/KafkaController.java
@@ -46,7 +46,7 @@ import org.springframework.web.bind.annotation.ResponseBody;
@RequestMapping(value = "/api/kafka", produces = { HTTP_VND_APACHE_KYLIN_JSON,
HTTP_VND_APACHE_KYLIN_V4_PUBLIC_JSON })
public class KafkaController extends NBasicController {
- @Autowired
+ @Autowired(required = false)
@Qualifier("kafkaService")
private KafkaService kafkaService;
