[feature](RoutineLoad) Support the Amazon Kinesis

be/cmake/thirdparty.cmake

@@ -140,6 +140,7 @@ add_thirdparty(aws-c-s3 LIB64)
 add_thirdparty(aws-c-sdkutils LIB64)
 add_thirdparty(aws-cpp-sdk-identity-management LIB64)
 add_thirdparty(aws-cpp-sdk-sts LIB64)
+add_thirdparty(aws-cpp-sdk-kinesis LIB64)
 if (NOT OS_MACOSX)
     add_thirdparty(aws-s2n LIBNAME "lib/libs2n.a")
 endif()

be/src/common/metrics/doris_metrics.cpp

@@ -167,6 +167,13 @@ DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(routine_load_get_msg_latency, MetricUnit::M
 DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(routine_load_get_msg_count, MetricUnit::NOUNIT);
 DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(routine_load_consume_rows, MetricUnit::ROWS);
 DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(routine_load_consume_bytes, MetricUnit::BYTES);
+DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(routine_load_kinesis_get_records_latency,
+                                     MetricUnit::MILLISECONDS);
+DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(routine_load_kinesis_get_records_count, MetricUnit::NOUNIT);
+DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(routine_load_kinesis_throttle_count, MetricUnit::NOUNIT);
+DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(routine_load_kinesis_retriable_error_count,
+                                     MetricUnit::NOUNIT);
+DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(routine_load_kinesis_closed_shard_count, MetricUnit::NOUNIT);
 
 DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(memtable_flush_total, MetricUnit::OPERATIONS);
 DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(memtable_flush_duration_us, MetricUnit::MICROSECONDS);
@@ -353,6 +360,11 @@ DorisMetrics::DorisMetrics() : _metric_registry(_s_registry_name) {
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, routine_load_get_msg_count);
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, routine_load_consume_bytes);
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, routine_load_consume_rows);
+    INT_COUNTER_METRIC_REGISTER(_server_metric_entity, routine_load_kinesis_get_records_latency);
+    INT_COUNTER_METRIC_REGISTER(_server_metric_entity, routine_load_kinesis_get_records_count);
+    INT_COUNTER_METRIC_REGISTER(_server_metric_entity, routine_load_kinesis_throttle_count);
+    INT_COUNTER_METRIC_REGISTER(_server_metric_entity, routine_load_kinesis_retriable_error_count);
+    INT_COUNTER_METRIC_REGISTER(_server_metric_entity, routine_load_kinesis_closed_shard_count);
 
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, memtable_flush_total);
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, memtable_flush_duration_us);

be/src/common/metrics/doris_metrics.h

@@ -139,6 +139,11 @@ class DorisMetrics {
     IntCounter* routine_load_get_msg_count = nullptr;
     IntCounter* routine_load_consume_bytes = nullptr;
     IntCounter* routine_load_consume_rows = nullptr;
+    IntCounter* routine_load_kinesis_get_records_latency = nullptr;
+    IntCounter* routine_load_kinesis_get_records_count = nullptr;
+    IntCounter* routine_load_kinesis_throttle_count = nullptr;
+    IntCounter* routine_load_kinesis_retriable_error_count = nullptr;
+    IntCounter* routine_load_kinesis_closed_shard_count = nullptr;
 
     IntCounter* memtable_flush_total = nullptr;
     IntCounter* memtable_flush_duration_us = nullptr;

be/src/io/fs/kafka_consumer_pipe.h

@@ -27,21 +27,6 @@ class KafkaConsumerPipe : public StreamLoadPipe {
             : StreamLoadPipe(max_buffered_bytes, min_chunk_size) {}
 
     ~KafkaConsumerPipe() override = default;
-
-    virtual Status append_with_line_delimiter(const char* data, size_t size) {
-        Status st = append(data, size);
-        if (!st.ok()) {
-            return st;
-        }
-
-        // append the line delimiter
-        st = append("\n", 1);
-        return st;
-    }
-
-    virtual Status append_json(const char* data, size_t size) {
-        return append_and_flush(data, size);
-    }
 };
 } // namespace io
 } // end namespace doris

be/src/io/fs/kinesis_consumer_pipe.h

@@ -0,0 +1,34 @@
+// 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.
+
+#pragma once
+
+#include "io/fs/stream_load_pipe.h"
+
+namespace doris {
+namespace io {
+
+class KinesisConsumerPipe : public StreamLoadPipe {
+public:
+    KinesisConsumerPipe(size_t max_buffered_bytes = 1024 * 1024, size_t min_chunk_size = 64 * 1024)
+            : StreamLoadPipe(max_buffered_bytes, min_chunk_size) {}
+
+    ~KinesisConsumerPipe() override = default;
+};
+
+} // namespace io
+} // namespace doris

be/src/io/fs/stream_load_pipe.h

@@ -53,6 +53,15 @@ class StreamLoadPipe : public MessageBodySink, public FileReader {
     Status append(const char* data, size_t size) override;
     Status append(const ByteBufferPtr& buf) override;
 
+    virtual Status append_with_line_delimiter(const char* data, size_t size) {
+        RETURN_IF_ERROR(append(data, size));
+        return append("\n", 1);
+    }
+
+    virtual Status append_json(const char* data, size_t size) {
+        return append_and_flush(data, size);
+    }
+
     const Path& path() const override { return _path; }
 
     size_t size() const override { return 0; }

be/src/load/routine_load/consumer_group_helpers.h

@@ -0,0 +1,96 @@
+// 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.
+
+#pragma once
+
+#include <map>
+#include <string>
+#include <vector>
+
+namespace doris {
+
+// Helper class for partitioning work items (partitions/shards) across consumers
+template <typename KeyType, typename ValueType>
+class WorkPartitioner {
+public:
+    // Divide work items equally across N consumers using round-robin
+    static std::vector<std::map<KeyType, ValueType>> partition_round_robin(
+            const std::map<KeyType, ValueType>& work_items, int consumer_count) {
+        std::vector<std::map<KeyType, ValueType>> result(consumer_count);
+        int i = 0;
+        for (const auto& [key, value] : work_items) {
+            int idx = i % consumer_count;
+            result[idx].emplace(key, value);
+            i++;
+        }
+        return result;
+    }
+};
+
+// Helper class for managing format-based append operations
+class FormatAppender {
+public:
+    // Get the appropriate append function pointer based on format type
+    template <typename PipeType>
+    static auto get_append_function(TFileFormatType::type format)
+            -> Status (PipeType::*)(const char*, size_t) {
+        return (format == TFileFormatType::FORMAT_JSON) ? &PipeType::append_json
+                                                        : &PipeType::append_with_line_delimiter;
+    }
+};
+
+// Helper class for tracking consumption progress
+class ConsumptionProgress {
+public:
+    ConsumptionProgress(int64_t max_time_ms, int64_t max_rows, int64_t max_bytes)
+            : _initial_time(max_time_ms),
+              _initial_rows(max_rows),
+              _initial_bytes(max_bytes),
+              _left_time(max_time_ms),
+              _left_rows(max_rows),
+              _left_bytes(max_bytes) {}
+
+    // Check if any limit is reached
+    bool is_limit_reached() const { return _left_time <= 0 || _left_rows <= 0 || _left_bytes <= 0; }
+
+    // Update progress after consuming one item
+    void consume_item(int64_t bytes) {
+        _left_rows--;
+        _left_bytes -= bytes;
+    }
+
+    // Update time progress
+    void update_time(int64_t elapsed_us) { _left_time = _initial_time - elapsed_us / 1000; }
+
+    // Getters
+    int64_t left_time() const { return _left_time; }
+    int64_t left_rows() const { return _left_rows; }
+    int64_t left_bytes() const { return _left_bytes; }
+    int64_t consumed_rows() const { return _initial_rows - _left_rows; }
+    int64_t consumed_bytes() const { return _initial_bytes - _left_bytes; }
+    int64_t consumed_time() const { return _initial_time - _left_time; }
+
+private:
+    int64_t _initial_time;
+    int64_t _initial_rows;
+    int64_t _initial_bytes;
+    int64_t _left_time;
+    int64_t _left_rows;
+    int64_t _left_bytes;
+};
+
+} // namespace doris

be/src/load/routine_load/consumer_helpers.cpp

@@ -0,0 +1,61 @@
+// 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.
+
+#include "load/routine_load/consumer_helpers.h"
+
+#include <algorithm>
+#include <thread>
+
+#include "common/metrics/doris_metrics.h"
+
+namespace doris {
+
+// ConsumerMetrics implementation
+void ConsumerMetrics::track_get_msg(int64_t latency_us) {
+    _get_msg_count++;
+    DorisMetrics::instance()->routine_load_get_msg_count->increment(1);
+    DorisMetrics::instance()->routine_load_get_msg_latency->increment(latency_us / 1000);
+}
+
+void ConsumerMetrics::track_consume_bytes(int64_t bytes) {
+    _consume_bytes += bytes;
+    DorisMetrics::instance()->routine_load_consume_bytes->increment(bytes);
+}
+
+void ConsumerMetrics::track_consume_rows(int64_t rows) {
+    _consume_rows += rows;
+    DorisMetrics::instance()->routine_load_consume_rows->increment(rows);
+}
+
+// RetryPolicy implementation
+void RetryPolicy::retry_with_backoff() {
+    _retry_count++;
+    if (_retry_count <= _max_retries) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(_backoff_ms));
+    }
+}
+
+// ThrottleBackoff implementation
+void ThrottleBackoff::backoff_and_sleep() {
+    _throttle_count++;
+    // Exponential backoff: initial_ms * 2^(count-1), capped at max_ms
+    int backoff_ms = std::min(_initial_backoff_ms * (1 << std::min(_throttle_count - 1, 3)),
+                              _max_backoff_ms);
+    std::this_thread::sleep_for(std::chrono::milliseconds(backoff_ms));
+}
+
+} // namespace doris

be/src/load/routine_load/consumer_helpers.h

@@ -0,0 +1,119 @@
+// 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.
+
+#pragma once
+
+#include <chrono>
+#include <string>
+#include <unordered_map>
+
+namespace doris {
+
+// Helper class for tracking metrics in data consumers
+class ConsumerMetrics {
+public:
+    ConsumerMetrics() = default;
+
+    // Track a single message fetch operation
+    void track_get_msg(int64_t latency_us);
+
+    // Track consumed bytes
+    void track_consume_bytes(int64_t bytes);
+
+    // Track consumed rows
+    void track_consume_rows(int64_t rows);
+
+    // Get accumulated metrics
+    int64_t get_msg_count() const { return _get_msg_count; }
+    int64_t consume_bytes() const { return _consume_bytes; }
+    int64_t consume_rows() const { return _consume_rows; }
+
+private:
+    int64_t _get_msg_count = 0;
+    int64_t _consume_bytes = 0;
+    int64_t _consume_rows = 0;
+};
+
+// Helper class for retry logic with configurable backoff
+class RetryPolicy {
+public:
+    explicit RetryPolicy(int max_retries = 3, int backoff_ms = 200)
+            : _max_retries(max_retries), _backoff_ms(backoff_ms), _retry_count(0) {}
+
+    // Check if should retry
+    bool should_retry() const { return _retry_count < _max_retries; }
+
+    // Increment retry count and sleep with backoff
+    void retry_with_backoff();
+
+    // Reset retry counter (call on success)
+    void reset() { _retry_count = 0; }
+
+    // Get current retry count
+    int retry_count() const { return _retry_count; }
+
+private:
+    int _max_retries;
+    int _backoff_ms;
+    int _retry_count;
+};
+
+// Helper class for exponential backoff (used for throttling)
+class ThrottleBackoff {
+public:
+    explicit ThrottleBackoff(int initial_backoff_ms = 1000, int max_backoff_ms = 10000)
+            : _initial_backoff_ms(initial_backoff_ms),
+              _max_backoff_ms(max_backoff_ms),
+              _throttle_count(0) {}
+
+    // Increment throttle count and sleep with exponential backoff
+    void backoff_and_sleep();
+
+    // Reset throttle counter (call on success)
+    void reset() { _throttle_count = 0; }
+
+    // Get current throttle count
+    int throttle_count() const { return _throttle_count; }
+
+private:
+    int _initial_backoff_ms;
+    int _max_backoff_ms;
+    int _throttle_count;
+};
+
+// Helper class for comparing custom properties between consumers
+class PropertyMatcher {
+public:
+    // Check if two property maps match exactly
+    template <typename MapType1, typename MapType2>
+    static bool properties_match(const MapType1& props1, const MapType2& props2) {
+        if (props1.size() != props2.size()) {
+            return false;
+        }
+
+        for (const auto& [key, value] : props1) {
+            auto it = props2.find(key);
+            if (it == props2.end() || it->second != value) {
+                return false;
+            }
+        }
+
+        return true;
+    }
+};
+
+} // namespace doris