apache · imay · Jun 7, 2019 · May 29, 2019 · May 29, 2019 · May 29, 2019
diff --git a/be/src/runtime/routine_load/data_consumer.cpp b/be/src/runtime/routine_load/data_consumer.cpp
@@ -236,7 +236,7 @@ Status KafkaDataConsumer::get_partition_meta(std::vector<int32_t>* partition_ids
     // create topic
     std::string errstr;
     RdKafka::Topic *topic = RdKafka::Topic::create(_k_consumer, _topic, tconf, errstr);
-    if (!topic) {
+    if (topic == nullptr) {
         std::stringstream ss;
         ss << "failed to create topic: " << errstr;
         LOG(WARNING) << ss.str();
@@ -246,7 +246,7 @@ Status KafkaDataConsumer::get_partition_meta(std::vector<int32_t>* partition_ids
     DeferOp delete_topic(std::bind<void>(topic_deleter));
 
     // get topic metadata
-    RdKafka::Metadata *metadata;
+    RdKafka::Metadata* metadata = nullptr;
     RdKafka::ErrorCode err = _k_consumer->metadata(true/* for this topic */, topic, &metadata, 5000);
     if (err != RdKafka::ERR_NO_ERROR) {
         std::stringstream ss;

diff --git a/be/src/runtime/routine_load/routine_load_task_executor.cpp b/be/src/runtime/routine_load/routine_load_task_executor.cpp
@@ -34,15 +34,27 @@
 namespace doris {
 
 Status RoutineLoadTaskExecutor::get_kafka_partition_meta(
-        const TKafkaMetaProxyRequest& request, std::vector<int32_t>* partition_ids) {
-    DCHECK(request.__isset.kafka_info);
+        const PKafkaMetaProxyRequest& request, std::vector<int32_t>* partition_ids) {
+    DCHECK(request.has_kafka_info());
 
     // This context is meaningless, just for unifing the interface
     StreamLoadContext ctx(_exec_env);
     ctx.load_type = TLoadType::ROUTINE_LOAD;
     ctx.load_src_type = TLoadSourceType::KAFKA;
     ctx.label = "NaN";
-    ctx.kafka_info = new KafkaLoadInfo(request.kafka_info);
+
+    // convert PKafkaInfo to TKafkaLoadInfo
+    TKafkaLoadInfo t_info;
+    t_info.brokers = request.kafka_info().brokers();
+    t_info.topic = request.kafka_info().topic();
+    std::map<std::string, std::string> properties;
+    for (int i = 0; i < request.kafka_info().properties_size(); ++i) {
+        const PStringPair& pair = request.kafka_info().properties(i);
+        properties.emplace(pair.key(), pair.val());
+    }
+    t_info.__set_properties(std::move(properties));
+
+    ctx.kafka_info = new KafkaLoadInfo(t_info);
     ctx.need_rollback = false;
 
     std::shared_ptr<DataConsumer> consumer;
@@ -63,7 +75,8 @@ Status RoutineLoadTaskExecutor::submit_task(const TRoutineLoadTask& task) {
         return Status::OK;
     }
 
-    if (_thread_pool.get_queue_size() > 100) {
+    // the max queue size of thread pool is 100, here we use 80 as a very conservative limit
+    if (_thread_pool.get_queue_size() >= 80) {
         LOG(INFO) << "too many tasks in queue: " << _thread_pool.get_queue_size() << ", reject task: " << UniqueId(task.id);
         return Status("too many tasks");
     }

diff --git a/be/src/runtime/routine_load/routine_load_task_executor.h b/be/src/runtime/routine_load/routine_load_task_executor.h
@@ -25,6 +25,8 @@
 #include "util/thread_pool.hpp"
 #include "util/uid_util.h"
 
+#include "gen_cpp/internal_service.pb.h"
+
 namespace doris {
 
 class ExecEnv;
@@ -56,7 +58,7 @@ class RoutineLoadTaskExecutor {
     // submit a routine load task
     Status submit_task(const TRoutineLoadTask& task);
 
-    Status get_kafka_partition_meta(const TKafkaMetaProxyRequest& request, std::vector<int32_t>* partition_ids);
+    Status get_kafka_partition_meta(const PKafkaMetaProxyRequest& request, std::vector<int32_t>* partition_ids);
 
 private:
     // execute the task

diff --git a/be/src/runtime/small_file_mgr.cpp b/be/src/runtime/small_file_mgr.cpp
@@ -110,7 +110,7 @@ Status SmallFileMgr::get_file(
         Status st = _check_file(entry, md5);
         if (!st.ok()) {
             // check file failed, we should remove this cache and download it from FE again
-            if (!remove(entry.path.c_str())) {
+            if (remove(entry.path.c_str()) != 0) {
                 std::stringstream ss;
                 ss << "failed to remove file: " << file_id << ", err: "<< std::strerror(errno);
                 return Status(ss.str());

diff --git a/be/src/service/backend_service.cpp b/be/src/service/backend_service.cpp
@@ -245,21 +245,4 @@ void BackendService::submit_routine_load_task(
     return Status::OK.to_thrift(&t_status);
 }
 
-void BackendService::get_info(TProxyResult& result, const TProxyRequest& request) {
-    result.status.status_code = TStatusCode::OK;
-    if (request.__isset.kafka_meta_request) {
-        std::vector<int32_t> partition_ids;
-        Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(request.kafka_meta_request, &partition_ids);
-        if (!st.ok()) {
-            st.to_thrift(&result.status);
-            return;
-        } else {
-            TKafkaMetaProxyResult kafka_result;
-            kafka_result.__set_partition_ids(std::move(partition_ids));
-            result.__set_kafka_meta_result(std::move(kafka_result));
-            return;
-        }
-    }
-}
-
 } // namespace doris
diff --git a/be/src/service/backend_service.h b/be/src/service/backend_service.h
@@ -148,9 +148,6 @@ class BackendService : public BackendServiceIf {
     virtual void get_tablet_stat(TTabletStatResult& result) override;
 
     virtual void submit_routine_load_task(TStatus& t_status, const std::vector<TRoutineLoadTask>& tasks) override;
-
-    virtual void get_info(TProxyResult& result, const TProxyRequest& request) override;
-
 private:
     Status start_plan_fragment_execution(const TExecPlanFragmentParams& exec_params);
 

diff --git a/be/src/service/internal_service.cpp b/be/src/service/internal_service.cpp
@@ -28,6 +28,7 @@
 #include "util/thrift_util.h"
 #include "runtime/buffer_control_block.h"
 #include "runtime/result_buffer_mgr.h"
+#include "runtime/routine_load/routine_load_task_executor.h"
 
 namespace doris {
 
@@ -183,6 +184,30 @@ void PInternalServiceImpl<T>::trigger_profile_report(
     st.to_protobuf(result->mutable_status());
 }
 
+template<typename T>
+void PInternalServiceImpl<T>::get_info(
+    google::protobuf::RpcController* controller,
+    const PProxyRequest* request,
+    PProxyResult* response,
+    google::protobuf::Closure* done) {
+
+    brpc::ClosureGuard closure_guard(done);
+    if (request->has_kafka_meta_request()) {
+        std::vector<int32_t> partition_ids;
+        Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(request->kafka_meta_request(), &partition_ids);
+        if (st.ok()) {
+            PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
+            for (int32_t id : partition_ids) {
+                kafka_result->add_partition_ids(id);
+            } 
+        }
+        st.to_protobuf(response->mutable_status());
+        return;
+    }
+    Status::OK.to_protobuf(response->mutable_status());
+}
+
+
 template class PInternalServiceImpl<PBackendService>;
 template class PInternalServiceImpl<palo::PInternalService>;
 

diff --git a/be/src/service/internal_service.h b/be/src/service/internal_service.h
@@ -80,6 +80,12 @@ class PInternalServiceImpl : public T {
         PTriggerProfileReportResult* result,
         google::protobuf::Closure* done) override;
 
+    void get_info(
+        google::protobuf::RpcController* controller,
+        const PProxyRequest* request,
+        PProxyResult* response,
+        google::protobuf::Closure* done) override;
+
 private:
     Status _exec_plan_fragment(brpc::Controller* cntl);
 private:

@@ -36,6 +36,7 @@ under the License.
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <maven.compiler.source>1.8</maven.compiler.source>
         <maven.compiler.target>1.8</maven.compiler.target>
+        <jprotobuf.version>2.2.11</jprotobuf.version>
     </properties>
 
     <profiles>
@@ -242,7 +243,8 @@ under the License.
         <dependency>
             <groupId>com.baidu</groupId>
             <artifactId>jprotobuf</artifactId>
-            <version>1.11.0</version>
+            <version>${jprotobuf.version}</version>
+            <classifier>jar-with-dependencies</classifier>
         </dependency>
 
         <!-- https://mvnrepository.com/artifact/com.baidu/jprotobuf-rpc-common -->
@@ -256,7 +258,7 @@ under the License.
         <dependency>
             <groupId>com.baidu</groupId>
             <artifactId>jprotobuf-rpc-core</artifactId>
-            <version>3.5.17</version>
+            <version>3.5.21</version>
         </dependency>
 
         <!-- https://mvnrepository.com/artifact/org.json/json -->
@@ -358,7 +360,7 @@ under the License.
         <dependency>
             <groupId>com.google.protobuf</groupId>
             <artifactId>protobuf-java</artifactId>
-            <version>2.5.0</version>
+            <version>3.5.1</version>
         </dependency>
 
         <!-- https://mvnrepository.com/artifact/com.squareup/protoparser -->
@@ -576,6 +578,7 @@ under the License.
             </plugin>
 
             <!-- run make to generate Version and builtin -->
+            <!-- also parse the proto for FE -->
             <plugin>
                 <groupId>org.codehaus.mojo</groupId>
                 <artifactId>exec-maven-plugin</artifactId>
@@ -596,6 +599,24 @@ under the License.
                             <skip>${skip.plugin}</skip>
                         </configuration>
                     </execution>
+                    <execution>
+                        <id>gen_proto</id>
+                        <phase>generate-sources</phase>
+                        <goals>
+                            <!-- DO NOT use goal 'java', it will terminate the VM after done -->
+                            <goal>exec</goal>
+                        </goals>
+                        <configuration>
+                            <executable>java</executable>
+                            <arguments>
+                                <argument>-jar</argument>
+                                <argument>${settings.localRepository}/com/baidu/jprotobuf/${jprotobuf.version}/jprotobuf-${jprotobuf.version}-jar-with-dependencies.jar</argument>
+                                <argument>--java_out=${palo.home}/gensrc/build/java/</argument>
+                                <argument>${palo.home}/gensrc/proto/internal_service.proto</argument>
+                            </arguments>
+                            <skip>${skip.plugin}</skip>
+                        </configuration>
+                    </execution>
                 </executions>
             </plugin>
 

diff --git a/fe/src/main/java/org/apache/doris/common/Status.java b/fe/src/main/java/org/apache/doris/common/Status.java
@@ -17,7 +17,7 @@
 
 package org.apache.doris.common;
 
-import org.apache.doris.rpc.PStatus;
+import org.apache.doris.proto.PStatus;
 import org.apache.doris.thrift.TStatus;
 import org.apache.doris.thrift.TStatusCode;
 
@@ -82,9 +82,9 @@ public void setStatus(String msg) {
     }
 
     public void setPstatus(PStatus status) {
-        this.errorCode = TStatusCode.findByValue(status.code);
-        if (status.msgs != null && !status.msgs.isEmpty()) {
-            this.errorMsg = status.msgs.get(0);
+        this.errorCode = TStatusCode.findByValue(status.status_code);
+        if (status.error_msgs != null && !status.error_msgs.isEmpty()) {
+            this.errorMsg = status.error_msgs.get(0);
         }
     }
 

diff --git a/fe/src/main/java/org/apache/doris/common/proc/CurrentQueryInfoProvider.java b/fe/src/main/java/org/apache/doris/common/proc/CurrentQueryInfoProvider.java
@@ -17,26 +17,31 @@
 
 package org.apache.doris.common.proc;
 
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
 import org.apache.doris.catalog.Catalog;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.Pair;
 import org.apache.doris.common.util.Counter;
 import org.apache.doris.common.util.DebugUtil;
 import org.apache.doris.common.util.RuntimeProfile;
+import org.apache.doris.proto.PTriggerProfileReportResult;
+import org.apache.doris.proto.PUniqueId;
 import org.apache.doris.qe.QueryStatisticsItem;
-import org.apache.doris.rpc.*;
+import org.apache.doris.rpc.BackendServiceProxy;
+import org.apache.doris.rpc.PTriggerProfileReportRequest;
+import org.apache.doris.rpc.RpcException;
 import org.apache.doris.system.Backend;
 import org.apache.doris.thrift.TNetworkAddress;
 import org.apache.doris.thrift.TStatusCode;
 import org.apache.doris.thrift.TUniqueId;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 import java.util.Collection;
-import java.util.Formatter;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
@@ -198,7 +203,9 @@ private void triggerProfileReport(Collection<QueryStatisticsItem> items, boolean
                 }
                 // specified query instance which will report.
                 if (!allQuery) {
-                    final PUniqueId pUId = new PUniqueId(instanceInfo.getInstanceId());
+                    final PUniqueId pUId = new PUniqueId();
+                    pUId.hi = instanceInfo.getInstanceId().hi;
+                    pUId.lo = instanceInfo.getInstanceId().lo;
                     request.addInstanceId(pUId);
                 }
             }
@@ -230,11 +237,11 @@ private void recvResponse(List<Pair<Request, Future<PTriggerProfileReportResult>
             try {
                 final PTriggerProfileReportResult result
                         = pair.second.get(2, TimeUnit.SECONDS);
-                final TStatusCode code = TStatusCode.findByValue(result.status.code);
+                final TStatusCode code = TStatusCode.findByValue(result.status.status_code);
                 if (code != TStatusCode.OK) {
                     String errMsg = "";
-                    if (result.status.msgs != null && !result.status.msgs.isEmpty()) {
-                        errMsg = result.status.msgs.get(0);
+                    if (result.status.error_msgs != null && !result.status.error_msgs.isEmpty()) {
+                        errMsg = result.status.error_msgs.get(0);
                     }
                     throw new AnalysisException(reasonPrefix + " backend:" + pair.first.getAddress()
                             + " reason:" + errMsg);

diff --git a/fe/src/main/java/org/apache/doris/common/util/DebugUtil.java b/fe/src/main/java/org/apache/doris/common/util/DebugUtil.java
@@ -18,7 +18,7 @@
 package org.apache.doris.common.util;
 
 import org.apache.doris.common.Pair;
-import org.apache.doris.rpc.PUniqueId;
+import org.apache.doris.proto.PUniqueId;
 import org.apache.doris.thrift.TUniqueId;
 
 import java.io.PrintWriter;