Skip to content
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.

Sign up for GitHub

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

Added support for accessing cached offsets. #208

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Added support for accessing cached offsets. #208

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
47c18fc
Added support for accessing cached offsets.
Fischiii Jun 21, 2023
fa6b983
Applied code review suggestions.
Fischiii Jul 5, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 32 additions & 0 deletions include/kafka/KafkaConsumer.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,24 +130,32 @@ class KafkaConsumer: public KafkaClient
/**
* Get the first offset for the given partitions.
* This method does not change the current consumer position of the partitions.
* If the timeout is 0 cached offsets will be loaded and no call to the broker is made.
* Throws KafkaException with errors:
* - RD_KAFKA_RESP_ERR__FAIL: Generic failure
*/
std::map<TopicPartition, Offset> beginningOffsets(const TopicPartitions& topicPartitions,
std::chrono::milliseconds timeout = std::chrono::milliseconds(DEFAULT_QUERY_TIMEOUT_MS)) const
{
if (timeout == 0) {
return getOffsetsCached(topicPartitions, true);
}
return getOffsets(topicPartitions, true, timeout);
}

/**
* Get the last offset for the given partitions. The last offset of a partition is the offset of the upcoming message, i.e. the offset of the last available message + 1.
* This method does not change the current consumer position of the partitions.
* If the timeout is 0 cached offsets will be loaded and no call to the broker is made.
* Throws KafkaException with errors:
* - RD_KAFKA_RESP_ERR__FAIL: Generic failure
*/
std::map<TopicPartition, Offset> endOffsets(const TopicPartitions& topicPartitions,
std::chrono::milliseconds timeout = std::chrono::milliseconds(DEFAULT_QUERY_TIMEOUT_MS)) const
{
if (timeout == 0) {
return getOffsetsCached(topicPartitions, false);
}
return getOffsets(topicPartitions, false, timeout);
}

Expand Down Expand Up @@ -279,6 +287,8 @@ class KafkaConsumer: public KafkaClient
std::map<TopicPartition, Offset> getOffsets(const TopicPartitions& topicPartitions,
bool atBeginning,
std::chrono::milliseconds timeout) const;
std::map<TopicPartition, Offset> getOffsetsCached(const TopicPartitions& topicPartitions,
bool atBeginning) const;

enum class PartitionsRebalanceEvent { Assign, Revoke, IncrementalAssign, IncrementalUnassign };
void changeAssignment(PartitionsRebalanceEvent event, const TopicPartitions& tps);
Expand Down Expand Up @@ -752,6 +762,28 @@ KafkaConsumer::getOffsets(const TopicPartitions& topicPartitions,
return result;
}

inline std::map<TopicPartition, Offset>
KafkaConsumer::getOffsetsCached(const TopicPartitions& topicPartitions,
bool atBeginning) const
{
std::map<TopicPartition, Offset> result;

for (const auto& topicPartition: topicPartitions)
{
Offset beginning{}, end{};
const Error error{ rd_kafka_get_watermark_offsets(getClientHandle(),
topicPartition.first.c_str(),
topicPartition.second,
&beginning,
&end) };
KAFKA_THROW_IF_WITH_ERROR(error);

result[topicPartition] = (atBeginning ? beginning : end);
}

return result;
}

// Commit
inline void
KafkaConsumer::commit(const TopicPartitionOffsets& topicPartitionOffsets, CommitType type)
Expand Down