[Fix-3.0](iot) Avoid coredump because of NullPointerException #55597

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed

zclllyybb wants to merge 572 commits into apache:branch-3.1 from zclllyybb:iot30

.clang-format-ignore

-Original file line number
+Diff line change
@@ Expand Up / @@ -9,3 +9,4 @@ be/src/util/sse2neon.h @@
     be/src/util/mustache/mustache.h
     be/src/util/mustache/mustache.cc
     be/src/util/utf8_check.cpp
+    cloud/src/common/defer.h

.github/actions/setup-maven

  
  Submodule setup-maven

    deleted from
      417e1a

.github/workflows/be-ut-mac.yml

-Original file line number
+Diff line change
@@ Expand Up / @@ -85,14 +85,8 @@ jobs: @@
               brew install "${cellars[@]}" || true
               pushd thirdparty
-              branch="${{ github.base_ref }}"
-              if [[ -z "${branch}" ]] || [[ "${branch}" == 'master' ]]; then
-                curl -L https://github.com/apache/doris-thirdparty/releases/download/automation/doris-thirdparty-prebuilt-darwin-x86_64.tar.xz \
-                  -o doris-thirdparty-prebuilt-darwin-x86_64.tar.xz
-              else
-                curl -L "https://github.com/apache/doris-thirdparty/releases/download/automation-${branch/branch-/}/doris-thirdparty-prebuilt-darwin-x86_64.tar.xz" \
-                  -o doris-thirdparty-prebuilt-darwin-x86_64.tar.xz
-              fi
+              curl -L https://github.com/apache/doris-thirdparty/releases/download/automation-3.0/doris-thirdparty-prebuilt-darwin-x86_64.tar.xz \
+                -o doris-thirdparty-prebuilt-darwin-x86_64.tar.xz
               tar -xvf doris-thirdparty-prebuilt-darwin-x86_64.tar.xz
               popd
@@ Expand Down @@

.github/workflows/build-extension.yml

-Original file line number
+Diff line change
@@ Expand Up / @@ -66,14 +66,7 @@ jobs: @@
           - name: Setup thrift
             run: |
               pushd thirdparty
-              branch="${{ github.base_ref }}"
-              if [[ -z "${branch}" ]] || [[ "${branch}" == 'master' || "${branch}" == 'branch-3.0' || "${branch}" == 'branch-2.1' ]]; then
-                curl -L https://github.com/apache/doris-thirdparty/releases/download/automation/doris-thirdparty-prebuilt-linux-x86_64.tar.xz \
-                  -o doris-thirdparty-prebuilt-linux-x86_64.tar.xz
-              else
-                curl -L "https://github.com/apache/doris-thirdparty/releases/download/automation-${branch/branch-/}/doris-thirdparty-prebuilt-linux-x86_64.tar.xz" \
-                  -o doris-thirdparty-prebuilt-linux-x86_64.tar.xz
-              fi
+              curl -L "https://github.com/apache/doris-thirdparty/releases/download/automation-3.0/doris-thirdparty-prebuilt-linux-x86_64.tar.xz" -o doris-thirdparty-prebuilt-linux-x86_64.tar.xz
               tar -xvf doris-thirdparty-prebuilt-linux-x86_64.tar.xz
               popd
               export PATH="${DEFAULT_DIR}/ldb-toolchain/bin/:$(pwd)/thirdparty/installed/bin/:${PATH}"
@@ Expand Down @@

.github/workflows/build-thirdparty.yml

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -72,7 +72,7 @@ jobs:
  
          - name: Download

            run: |

              cd thirdparty

              curl -L https://github.com/apache/doris-thirdparty/releases/download/automation/doris-thirdparty-source.tgz \

              curl -L https://github.com/apache/doris-thirdparty/releases/download/automation-3.0/doris-thirdparty-source.tgz \

                -o doris-thirdparty-source.tgz

              tar -zxvf doris-thirdparty-source.tgz

    @@ -138,7 +138,7 @@ jobs:
  
          - name: Download

            run: |

              cd thirdparty

              curl -L https://github.com/apache/doris-thirdparty/releases/download/automation/doris-thirdparty-source.tgz \

              curl -L https://github.com/apache/doris-thirdparty/releases/download/automation-3.0/doris-thirdparty-source.tgz \

                -o doris-thirdparty-source.tgz

              tar -zxvf doris-thirdparty-source.tgz

    @@ -196,7 +196,7 @@ jobs:
  
          - name: Download

            run: |

              cd thirdparty

              curl -L https://github.com/apache/doris-thirdparty/releases/download/automation/doris-thirdparty-source.tgz \

              curl -L https://github.com/apache/doris-thirdparty/releases/download/automation-3.0/doris-thirdparty-source.tgz \

                -o doris-thirdparty-source.tgz

              tar -zxvf doris-thirdparty-source.tgz

.github/workflows/checkstyle.yaml

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -42,11 +42,13 @@ jobs:
  
                  - 'gensrc/proto/**'

                  - 'gensrc/thrift/**'

          - name: Setup Maven Action

          - name: Set up JDK with Maven

            if: steps.filter.outputs.fe_changes == 'true'

            uses: ./.github/actions/setup-maven

            uses: actions/setup-java@v4

            with:

              maven-version: 3.8.4

              distribution: 'temurin'

              java-version: '17'

              cache: 'maven'    

          - name: Run java checkstyle

            if: steps.filter.outputs.fe_changes == 'true'

.github/workflows/code-checks.yml

-Original file line number
+Diff line change
@@ Expand Up / @@ -99,7 +99,7 @@ jobs: @@
                 sudo DEBIAN_FRONTEND=noninteractive apt install --yes tzdata byacc
                 pushd thirdparty
-                curl -L https://github.com/apache/doris-thirdparty/releases/download/automation/doris-thirdparty-prebuilt-linux-x86_64.tar.xz \
+                curl -L https://github.com/apache/doris-thirdparty/releases/download/automation-3.0/doris-thirdparty-prebuilt-linux-x86_64.tar.xz \
                   -o doris-thirdparty-prebuilt-linux-x86_64.tar.xz
                 tar -xvf doris-thirdparty-prebuilt-linux-x86_64.tar.xz
                 popd
@@ Expand Down @@

.gitmodules

-Original file line number
+Diff line change
@@ Expand Up / @@ -4,9 +4,6 @@ @@
     [submodule ".github/actions/get-workflow-origin"]
     	path = .github/actions/get-workflow-origin
     	url = https://github.com/potiuk/get-workflow-origin.git
-    [submodule ".github/actions/setup-maven"]
-    	path = .github/actions/setup-maven
-    	url = https://github.com/stCarolas/setup-maven.git
     [submodule ".github/actions/paths-filter"]
     	path = .github/actions/paths-filter
     	url = https://github.com/dorny/paths-filter
@@ Expand Down @@

be/src/agent/agent_server.cpp

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -43,6 +43,7 @@
  
    #include "olap/snapshot_manager.h"

    #include "olap/storage_engine.h"

    #include "runtime/exec_env.h"

    #include "util/work_thread_pool.hpp"

    using std::string;

    using std::vector;

    @@ -169,8 +170,8 @@ void AgentServer::start_workers(StorageEngine& engine, ExecEnv* exec_env) {
  
        _workers[TTaskType::ALTER] = std::make_unique<TaskWorkerPool>(

                "ALTER_TABLE", config::alter_tablet_worker_count, [&engine](auto&& task) { return alter_tablet_callback(engine, task); });

        _workers[TTaskType::CLONE] = std::make_unique<TaskWorkerPool>(

                "CLONE", config::clone_worker_count, [&engine, &cluster_info = _cluster_info](auto&& task) { return clone_callback(engine, cluster_info, task); });

        _workers[TTaskType::CLONE] = std::make_unique<PriorTaskWorkerPool>(

                "CLONE", config::clone_worker_count,config::clone_worker_count, [&engine, &cluster_info = _cluster_info](auto&& task) { return clone_callback(engine, cluster_info, task); });

        _workers[TTaskType::STORAGE_MEDIUM_MIGRATE] = std::make_unique<TaskWorkerPool>(

                "STORAGE_MEDIUM_MIGRATE", config::storage_medium_migrate_count, [&engine](auto&& task) { return storage_medium_migrate_callback(engine, task); });

    @@ -196,6 +197,8 @@ void AgentServer::start_workers(StorageEngine& engine, ExecEnv* exec_env) {
  
        _report_workers.push_back(std::make_unique<ReportWorker>(

                "REPORT_OLAP_TABLET", _cluster_info, config::report_tablet_interval_seconds,[&engine, &cluster_info = _cluster_info] { report_tablet_callback(engine, cluster_info); }));

        // clang-format on

        exec_env->storage_engine().to_local().workers = &_workers;

    }

    void AgentServer::cloud_start_workers(CloudStorageEngine& engine, ExecEnv* exec_env) {

be/src/agent/heartbeat_server.cpp

-Original file line number
+Diff line change
@@ Expand Up / @@ -25,7 +25,9 @@ @@
     #include <memory>
     #include <ostream>
     #include <string>
+    #include <vector>
+    #include "cloud/cloud_cluster_info.h"
     #include "cloud/cloud_tablet_mgr.h"
     #include "cloud/config.h"
     #include "common/config.h"
@@ Expand Down Expand Up @@
                           << " " << st;
             }
-            if (master_info.meta_service_endpoint != config::meta_service_endpoint &&
-                config::enable_meta_service_endpoint_consistency_check) {
+            if (master_info.meta_service_endpoint != config::meta_service_endpoint) {
                 LOG(WARNING) << "Detected mismatch in meta_service_endpoint configuration between FE "
                                 "and BE. "
                              << "FE meta_service_endpoint: " << master_info.meta_service_endpoint
                              << ", BE meta_service_endpoint: " << config::meta_service_endpoint;
-                return Status::InvalidArgument<false>(
-                        "fe and be do not work in same mode or meta_service_endpoint mismatch,"
-                        "fe meta_service_endpoint: {}, be meta_service_endpoint: {}",
-                        master_info.meta_service_endpoint, config::meta_service_endpoint);
+                std::vector<std::string> old_endpoints =
+                        doris::split(config::meta_service_endpoint, ",");
+                std::vector<std::string> new_endpoints =
+                        doris::split(master_info.meta_service_endpoint, ",");
+                auto has_intersection = false;
+                for (auto endpoint : new_endpoints) {
+                    if (std::find(old_endpoints.begin(), old_endpoints.end(), endpoint) !=
+                        old_endpoints.end()) {
+                        has_intersection = true;
+                    }
+                }
+                if (has_intersection) {
+                    auto st = config::set_config("meta_service_endpoint",
+                                                 master_info.meta_service_endpoint, true);
+                    LOG(INFO) << "change config meta_service_endpoint to "
+                              << master_info.meta_service_endpoint << " " << st;
+                }
+                if (!has_intersection && config::enable_meta_service_endpoint_consistency_check) {
+                    return Status::InvalidArgument<false>(
+                            "fe and be do not work in same mode or meta_service_endpoint mismatch,"
+                            "fe meta_service_endpoint: {}, be meta_service_endpoint: {}",
+                            master_info.meta_service_endpoint, config::meta_service_endpoint);
+                }
             }
         }
@@ Expand All @@
             LOG(INFO) << "set config cloud_unique_id " << master_info.cloud_unique_id << " " << st;
         }
+        if (master_info.__isset.cloud_cluster_info &&
+            master_info.cloud_cluster_info.__isset.isStandby) {
+            auto* cloud_cluster_info = static_cast<CloudClusterInfo*>(_cluster_info);
+            cloud_cluster_info->set_is_in_standby(master_info.cloud_cluster_info.isStandby);
+        }
         if (master_info.__isset.tablet_report_inactive_duration_ms) {
             doris::g_tablet_report_inactive_duration_ms =
                     master_info.tablet_report_inactive_duration_ms;
@@ Expand Down @@

be/src/agent/task_worker_pool.cpp

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -17,6 +17,7 @@
  
    #include "agent/task_worker_pool.h"

    #include <brpc/controller.h>

    #include <fmt/format.h>

    #include <gen_cpp/AgentService_types.h>

    #include <gen_cpp/DataSinks_types.h>

    @@ -84,6 +85,7 @@
  
    #include "runtime/memory/global_memory_arbitrator.h"

    #include "runtime/snapshot_loader.h"

    #include "service/backend_options.h"

    #include "util/brpc_client_cache.h"

    #include "util/debug_points.h"

    #include "util/doris_metrics.h"

    #include "util/jni-util.h"

    @@ -291,7 +293,7 @@ void alter_cloud_tablet(CloudStorageEngine& engine, const TAgentTaskRequest& age
  
                    job.process_alter_tablet(agent_task_req.alter_tablet_req_v2),

                    [&](const doris::Exception& ex) {

                        DorisMetrics::instance()->create_rollup_requests_failed->increment(1);

                        job.clean_up_on_failed();

                        job.clean_up_on_failure();

                    });

            return Status::OK();

        }();

    @@ -604,6 +606,52 @@ Status PriorTaskWorkerPool::submit_task(const TAgentTaskRequest& task) {
  
        });

    }

    Status PriorTaskWorkerPool::submit_high_prior_and_cancel_low(const TAgentTaskRequest& task) {

        const TTaskType::type task_type = task.task_type;

        int64_t signature = task.signature;

        std::string type_str;

        EnumToString(TTaskType, task_type, type_str);

        auto req = std::make_unique<TAgentTaskRequest>(task);

        DCHECK(req->__isset.priority && req->priority == TPriority::HIGH);

        do {

            std::lock_guard lock(s_task_signatures_mtx);

            auto& set = s_task_signatures[task_type];

            if (!set.contains(signature)) {

                // If it doesn't exist, put it directly into the priority queue

                add_task_count(*req, 1);

                set.insert(signature);

                std::lock_guard lock(_mtx);

                _high_prior_queue.push_back(std::move(req));

                _high_prior_condv.notify_one();

                _normal_condv.notify_one();

                break;

            } else {

                std::lock_guard lock(_mtx);

                for (auto it = _normal_queue.begin(); it != _normal_queue.end();) {

                    // If it exists in the normal queue, cancel the task in the normal queue

                    if ((*it)->signature == signature) {

                        _normal_queue.erase(it);                     // cancel the original task

                        _high_prior_queue.push_back(std::move(req)); // add the new task to the queue

                        _high_prior_condv.notify_one();

                        _normal_condv.notify_one();

                        break;

                    } else {

                        ++it; // doesn't meet the condition, continue to the next one

                    }

                }

                // If it exists in the high priority queue, no operation is needed

                LOG_INFO("task has already existed in high prior queue.").tag("signature", signature);

            }

        } while (false);

        // Set the receiving time of task so that we can determine whether it is timed out later

        (const_cast<TAgentTaskRequest&>(task)).__set_recv_time(time(nullptr));

        LOG_INFO("successfully submit task").tag("type", type_str).tag("signature", signature);

        return Status::OK();

    }

    void PriorTaskWorkerPool::normal_loop() {

        while (true) {

            std::unique_ptr<TAgentTaskRequest> req;

    @@ -1432,23 +1480,15 @@ void update_s3_resource(const TStorageResource& param, io::RemoteFileSystemSPtr
  
            DCHECK_EQ(existed_fs->type(), io::FileSystemType::S3) << param.id << ' ' << param.name;

            auto client = static_cast<io::S3FileSystem*>(existed_fs.get())->client_holder();

            auto new_s3_conf = S3Conf::get_s3_conf(param.s3_storage_param);

            S3ClientConf conf {

                    .endpoint {},

                    .region {},

                    .ak = std::move(new_s3_conf.client_conf.ak),

                    .sk = std::move(new_s3_conf.client_conf.sk),

                    .token = std::move(new_s3_conf.client_conf.token),

                    .bucket {},

                    .provider = new_s3_conf.client_conf.provider,

            };

            S3ClientConf conf = std::move(new_s3_conf.client_conf);

            st = client->reset(conf);

            fs = std::move(existed_fs);

        }

        if (!st.ok()) {

            LOG(WARNING) << "update s3 resource failed: " << st;

        } else {

            LOG_INFO("successfully update hdfs resource")

            LOG_INFO("successfully update s3 resource")

                    .tag("resource_id", param.id)

                    .tag("resource_name", param.name);

            put_storage_resource(param.id, {std::move(fs)}, param.version);

    @@ -1893,8 +1933,9 @@ void PublishVersionWorkerPool::publish_version_callback(const TAgentTaskRequest&
  
                        if (!tablet->tablet_meta()->tablet_schema()->disable_auto_compaction()) {

                            tablet->published_count.fetch_add(1);

                            int64_t published_count = tablet->published_count.load();

                            int32_t max_version_config = tablet->max_version_config();

                            if (tablet->exceed_version_limit(

                                        config::max_tablet_version_num *

                                        max_version_config *

                                        config::load_trigger_compaction_version_percent / 100) &&

                                published_count % 20 == 0) {

                                auto st = _engine.submit_compaction_task(

be/src/agent/task_worker_pool.h

-Original file line number
+Diff line change
@@ Expand Up / @@ -89,6 +89,8 @@ class PriorTaskWorkerPool final : public TaskWorkerPoolIf { @@
         Status submit_task(const TAgentTaskRequest& task) override;
+        Status submit_high_prior_and_cancel_low(const TAgentTaskRequest& task);
     private:
         void normal_loop();
@@ Expand Down @@

be/src/agent/workload_group_listener.cpp

-Original file line number
+Diff line change
@@ Expand Up @@
             auto wg =
                     _exec_env->workload_group_mgr()->get_or_create_workload_group(workload_group_info);
-            // 3 set cpu soft hard limit switch
-            _exec_env->workload_group_mgr()->_enable_cpu_hard_limit.store(
-                    workload_group_info.enable_cpu_hard_limit);
             // 4 create and update task scheduler
             wg->upsert_task_scheduler(&workload_group_info, _exec_env);
             // 5 upsert io throttle
             wg->upsert_scan_io_throttle(&workload_group_info);
             LOG(INFO) << "[topic_publish_wg]update workload group finish, wg info="
-                      << wg->debug_string() << ", enable_cpu_hard_limit="
-                      << (_exec_env->workload_group_mgr()->enable_cpu_hard_limit() ? "true" : "false")
+                      << wg->debug_string()
                       << ", cgroup cpu_shares=" << workload_group_info.cgroup_cpu_shares
                       << ", cgroup cpu_hard_limit=" << workload_group_info.cgroup_cpu_hard_limit
                       << ", cgroup home path=" << config::doris_cgroup_cpu_path
@@ Expand Down @@

be/src/apache-orc

Submodule apache-orc updated 4 files

+8 −0		c++/src/ByteRLE.cc
+51 −0		c++/src/Compression.cc
+5 −0		c++/src/Reader.cc
+78 −0		c++/test/TestWriter.cc

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Fix-3.0](iot) Avoid coredump because of NullPointerException #55597

Uh oh!

Diff view

Diff view

Uh oh!

There are no files selected for viewing

Uh oh!