UnorderedOffsetCommitQueue.h

#pragma once
 
#include <kafka/Project.h>
 
#include <kafka/KafkaClient.h>
#include <kafka/Types.h>
 
#include <deque>
#include <vector>
 
namespace KAFKA_API { namespace clients { namespace consumer {
 
template <typename T>
class Heap
{
public:
    bool empty() const { return data.empty(); }
    std::size_t size() const { return data.size(); }
 
    const T& front() const { return data[0]; }
 
    void push(const T& t)
    {
        data.emplace_back(t);
 
        for (std::size_t indexCurrent = data.size() - 1; indexCurrent > 0;)
        {
            const std::size_t indexParent = (indexCurrent + 1) / 2 - 1;
 
            if (!(data[indexCurrent] < data[indexParent])) return;
 
            std::swap(data[indexCurrent], data[indexParent]);
            indexCurrent = indexParent;
        }
    }
 
    void pop_front()
    {
        data[0] = data.back();
        data.pop_back();
 
        if (data.empty()) return;
 
        for (std::size_t indexCurrent = 0;;)
        {
            const std::size_t indexRightChild = (indexCurrent + 1) * 2;
            const std::size_t indexLeftChild  = indexRightChild - 1;
 
            if (indexLeftChild >= data.size()) return;
 
            const std::size_t indexMinChild = (indexRightChild >= data.size() || data[indexLeftChild] < data[indexRightChild]) ? indexLeftChild : indexRightChild;
 
            if (!(data[indexMinChild] < data[indexCurrent])) return;
 
            std::swap(data[indexCurrent], data[indexMinChild]);
            indexCurrent = indexMinChild;
        }
    }
 
private:
    std::vector<T> data;
};
 
 
class UnorderedOffsetCommitQueue
{
public:
    UnorderedOffsetCommitQueue(const Topic& topic, Partition partition)
        : _partitionInfo(std::string("topic[").append(topic).append("], paritition[").append(std::to_string(partition)).append("]"))
    {
    }
    UnorderedOffsetCommitQueue() = default;
 
    std::size_t size() const { return _offsetsReceived.size(); }
 
    void waitOffset(Offset offset)
    {
        if (offset < 0 || (!_offsetsReceived.empty() && offset <= _offsetsReceived.back()))
        {
            // Invalid offset (might be fetched from the record which had no valid offset)
            KAFKA_API_LOG(Log::Level::Err, "Got invalid offset to wait[%lld]! %s", offset, (_partitionInfo.empty() ? "" : _partitionInfo.c_str()));
            return;
        }
 
        _offsetsReceived.emplace_back(offset);
    }
 
    void ackOffset(Offset offset)
    {
        const Offset maxOffsetReceived = _offsetsReceived.back();
        if (offset > maxOffsetReceived)
        {
            // Runtime error
            KAFKA_API_LOG(Log::Level::Err, "Got invalid ack offset[%lld]! Even larger than all offsets received[%lld]! %s", offset, maxOffsetReceived, (_partitionInfo.empty() ? "" : _partitionInfo.c_str()));
        }
 
        _offsetsToCommit.push(offset);
        do
        {
            const Offset minOffsetToCommit = _offsetsToCommit.front();
            const Offset expectedOffset    = _offsetsReceived.front();
            if (minOffsetToCommit == expectedOffset)
            {
                _toCommit = expectedOffset + 1;
                _offsetsToCommit.pop_front();
                _offsetsReceived.pop_front();
            }
            else if (minOffsetToCommit < expectedOffset)
            {
                // Inconsist error (might be caused by duplicated ack)
                KAFKA_API_LOG(Log::Level::Err, "Got invalid ack offset[%lld]! Even smaller than expected[%lld]! %s", minOffsetToCommit, expectedOffset, (_partitionInfo.empty() ? "" : _partitionInfo.c_str()));
                _offsetsToCommit.pop_front();
            }
            else
            {
                break;
            }
        } while (!_offsetsToCommit.empty());
    }
 
    Optional<Offset> popOffsetToCommit()
    {
        Optional<Offset> ret;
        if (_committed != _toCommit)
        {
            ret = _committed = _toCommit;
        }
        return ret;
    }
 
    Optional<Offset> lastPoppedOffset()
    {
        Optional<Offset> ret;
        if (_committed != INVALID_OFFSET)
        {
            ret = _committed;
        }
        return ret;
    }
 
private:
    std::deque<Offset> _offsetsReceived;
    Heap<Offset>       _offsetsToCommit;
    Offset             _toCommit        = {INVALID_OFFSET};
    Offset             _committed       = {INVALID_OFFSET};
    std::string        _partitionInfo;
 
    static constexpr Offset INVALID_OFFSET = -1;
};
 
} } } // end of KAFKA_API::clients::consumer