consumer_helpers.cpp

// 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