Skip to content

Commit kafka offset in routine load - #1734

Merged
morningman merged 1 commit into
apache:masterfrom
HangyuanLiu:routine_load_commit_kafka_offset
Sep 10, 2019
Merged

Commit kafka offset in routine load#1734
morningman merged 1 commit into
apache:masterfrom
HangyuanLiu:routine_load_commit_kafka_offset

Conversation

@HangyuanLiu

@HangyuanLiuHangyuanLiu commented Sep 2, 2019

Copy link
Copy Markdown
Contributor

Commit kafka offset in routine load

Kafka will decide whether to delete data based on whether all consumer group is commit offset or not.
If there is no commit offset, the kafka server disk may be full


for (auto& kv : ctx->kafka_info->cmt_offset) {
RdKafka::TopicPartition* tp1 = RdKafka::TopicPartition::create(
ctx->kafka_info->topic, kv.first, kv.second);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need to delete this tp1 after commit?

switch (ctx->load_src_type) {
case TLoadSourceType::KAFKA: {
std::shared_ptr<DataConsumer> consumer;
HANDLE_ERROR(_data_consumer_pool.get_consumer(ctx, &consumer), "failed to get consumers");

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's better not return any error if we failed to do something after txn is committed.

virtual Status reset() override;
virtual bool match(StreamLoadContext* ctx) override;
// commit kafka offset
Status commit(std::vector<RdKafka::TopicPartition*> offset);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
Statuscommit(std::vector<RdKafka::TopicPartition*>offset);
Statuscommit(conststd::vector<RdKafka::TopicPartition*>&offset);

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

librdkafka api is " ErrorCode commitSync (std::vector<TopicPartition*> &offsets)" , so here can't be const

@HangyuanLiu
HangyuanLiuforce-pushed the routine_load_commit_kafka_offset branch from a9cd126 to cba0073CompareSeptember 9, 2019 17:19

@morningmanmorningman left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM

@morningman
morningman merged commit 235cdb0 into apache:masterSep 10, 2019
@imayimay mentioned this pull request Sep 26, 2019
swjtu-zhanglei pushed a commit to swjtu-zhanglei/incubator-doris that referenced this pull request Jul 25, 2023
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants

@HangyuanLiu@imay@morningman